光說不練假把式，一起Kafka業務實戰。

7.1 順序性場景

7.1.1 場景概述

假設我們要傳輸一批訂單到另一個系統，那么訂單對應狀態的演變是有順序性要求的，

已下單 → 已支付 → 已確認

不允許錯亂！

7.1.2 順序級別

1）全域有序：

串行化，每條經過kafka的訊息必須嚴格保障有序性，

這就要求kafka單通道，每個groupid下單消費者

極大的影響性能，現實業務下幾乎沒必要

2）區域有序：

業務區域有序，同一條訂單有序即可，不同訂單可以并行處理，不同訂單的順序前后無所謂

充分利用kafka多磁區的并發性，只需要想辦法讓需要順序的一批資料進同一磁區即可，

7.1.3 實作方案

1）發送端：

指定key發送，key=order.id即可，案例回顧：4.2.3，PartitionProducer

2）發送中：

給佇列配置多磁區保障并發性，

3）讀取端：

單消費者：顯然不合理

吞吐量顯然上不去，kafka開多個磁區還有何意義？

所以開多個消費者指定磁區消費，理想狀況下，每個磁區配一個，

但是，這個吞吐量依然有限，那如何處理呢？

方案：多執行緒

在每個消費者上再開多執行緒，是個解決辦法，但是，要警惕順序性被打破！

參考下圖：thread處理后，會將data變成 2-1-3

改進：接收后分發二級記憶體佇列

消費者取到訊息后不做處理，根據key二次分發到多個阻塞佇列，

再開啟多個執行緒，每個佇列分配一個執行緒處理，提升吞吐量

7.1.4 代碼驗證

1）新建一個sort佇列，2個磁區

2）啟動order專案

原始碼參考：

SortedProducer（順序性發送端）

SortedConsumer（順序性消費端 - 阻塞佇列實作，方便大家理解設計思路）

SortedConsumer2（順序性消費端 - 執行緒池實作，現實中推薦這種方式！）

3）通過swagger請求

先按不同的id發送，查看控制臺日志，id被正確分發到對應的佇列

同一個key分配到同一個queue，順序性得到保障

7.2 海量同步場景

假設大資料部門需要大屏來展示用戶的打車訂單情況，需要把訂單資料送入druid

這里不涉及順序，只要下單就傳輸，但是對實時性和并發量要求較高

7.2.1 常規架構

在下單完成mysql后，通程序式代碼列印，直接進入kafka

或者logback和kafka集成，通過log輸送

優點：

更符合常規的思維，將資料送給想要的部門

缺點：

耦合度高，將kafka發送訊息嵌入了訂單下單的主業務，形成代碼入侵，

下單不關心，也不應該關注送入kafka的情況，一旦kafka不可用，程式受影響

7.2.2 解耦合

借助canal，監聽訂單表的資料變化，不再影響主業務，

7.2.3 部署實作

1）mysql部署

注意，需要打開binlog，8.0 默認處于開啟狀態

#啟動mysql8
docker run --name mysql8 -v /opt/data/mysql8:/var/lib/mysql -p 3389:3306 -e TZ=Asia/Shanghai -e MYSQL_ROOT_PASSWORD=thisisprizemysql8db -d daocloud.io/mysql:8.0

連上mysql，執行以下sql，添加canal用戶

CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';

創建訂單表

CREATE TABLE `orders` (
  `id` int unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
);

2）canal部署

#canal.properties
#附帶資料里有，放到服務器 /opt/data/canal/ 目錄下
#修改servers為你的kafka的機器地址
canal.serverMode = kafka
kafka.bootstrap.servers = 52.82.98.209:10903,52.82.98.209:10904

#docker-compose.yml
#附帶資料里有canal.yml，隨便找個目錄，重命名為docker-compose.yml
#修改mysql的鏈接資訊的鏈接資訊
#然后在當前目錄下執行 docker-compose up -d
version: '2'
services:
    canal:
        image: canal/canal-server
        container_name: canal
        restart: always
        ports:
            - "10908:11111"
        environment:
        		#mysql的鏈接資訊
            canal.instance.master.address: 52.82.98.209:3389
            canal.instance.dbUsername: canal
            canal.instance.dbPassword: canal
            #投放到kafka的哪個主題？要提前準備好！
            canal.mq.topic: canal
        volumes:
            - "/opt/data/canal/canal.properties:/home/admin/canal-server/conf/canal.properties"

3）資料通道驗證

進入kafka容器，用上面3.2.4里的命令列方式監聽canal佇列

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic canal

在mysql上創建orders表，增刪資料試一下

mysql> insert into orders (name) values ('張三');
Query OK, 1 row affected (0.03 sec)

在kafka控制臺，可以看到同步的訊息

{"data":[{"id":"1","name":"張三"}],"database":"canal","es":1611657853000,"id":5,"isDdl":false,"mysqlType":{"id":"int unsigned","name":"varchar(255)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12},"table":"orders","ts":1611657853802,"type":"INSERT"}

資料通道已打通，還缺少的是druid作為消費端來接收訊息

4）druid部署

#druid.yml
#在附帶資料里有
#隨便找個目錄，執行
docker-compose -f druid.yml up -d

5）驗證

配置druid的資料源，從kafka讀取資料，驗證資料可以正確進入druid，

注：

關于druid的詳細使用，在大資料篇章里會詳細講解，

7.3 kafka監控

7.3.1 eagle簡介

Kafka Eagle監控系統是一款用來監控Kafka集群的工具，支持管理多個Kafka集群、管理Kafka主題（包含查看、洗掉、創建等）、消費者組合消費者實體監控、訊息阻塞告警、Kafka集群健康狀態查看等，

7.3.2 部署

推薦docker-compose啟動

將配備的資料中 eagle.yml ， 拷貝到服務器任意目錄

修改對應的ip地址為你服務器的地址

#注意ip地址：52.82.98.209，全部換成你自己服務器的

version: '3'
services:
    zookeeper:
        image: zookeeper:3.4.13

    kafka-1:
        container_name: kafka-1
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10903:9092
            - 10913:10913
        environment:
            KAFKA_BROKER_ID: 1 
            HOST_IP: 52.82.98.209
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            #docker部署必須設定外部可訪問ip和埠，否則注冊進zk的地址將不可達造成外部無法連接
            KAFKA_ADVERTISED_HOST_NAME: 52.82.98.209
            KAFKA_ADVERTISED_PORT: 10903 
            KAFKA_JMX_OPTS: "-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=52.82.98.209 -Dcom.sun.management.jmxremote.rmi.port=10913"
            JMX_PORT: 10913
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper           
    kafka-2:
        container_name: kafka-2
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10904:9092
            - 10914:10914
        environment:
            KAFKA_BROKER_ID: 2 
            HOST_IP: 52.82.98.209
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 52.82.98.209
            KAFKA_ADVERTISED_PORT: 10904 
            KAFKA_JMX_OPTS: "-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=52.82.98.209 -Dcom.sun.management.jmxremote.rmi.port=10914"
            JMX_PORT: 10914
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper 
    eagle:
        image: gui66497/kafka_eagle
        container_name: ke
        restart: always
        depends_on:
            - kafka-1
            - kafka-2
        ports:
            - "10907:8048"
        environment:
            ZKSERVER: "zookeeper:2181"

執行 docker-compose -f eagle.yml up -d

7.3.3 使用說明

訪問 ： http://52.82.98.209:10907/ke/

默認用戶名密碼： admin / 123456

如果要洗掉topic等操作，需要管理token： keadmin

與km到底選哪個呢？根據自己習慣，個人認為：

• 界面美觀程度和監控曲線優于km，有登錄權限控制
• 功能操作上不如km簡單直白，但是km需要配置一定的連接資訊

本文由傳智教育博學谷 - 狂野架構師教研團隊發布
如果本文對您有幫助，歡迎關注和點贊；如果您有任何建議也可留言評論或私信，您的支持是我堅持創作的動力
轉載請注明出處！

標籤：Java

上一篇：Dubbo原始碼(八) - 負載均衡

下一篇：Java學習-第一部分-第二階段-第四節：常用類