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

7.1 順序性場景 7.1.1 場景概述假設我們要傳輸一批訂單到另一個系統，那麼訂單對應狀態的演變是有順序性要求的。已下單 → 已支付 → 已確認不允許錯亂！ 7.1.2 順序級別 1）全局有序：串列化。每條經過kafka的消息必須嚴格保障有序性。這就要求kafka單通道，每個groupi ...

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

file
改進：接收後分發二級記憶體隊列

消費者取到消息後不做處理，根據key二次分發到多個阻塞隊列。

再開啟多個線程，每個隊列分配一個線程處理。提升吞吐量

file

7.1.4 代碼驗證

1）新建一個sort隊列，2個分區

2）啟動order項目

源碼參考：

SortedProducer（順序性發送端）

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

SortedConsumer2（順序性消費端 - 線程池實現，現實中推薦這種方式！）

3）通過swagger請求

file

先按不同的id發送，查看控制台日誌，id被正確分發到對應的隊列

file

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

file

7.2 海量同步場景

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

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

7.2.1 常規架構

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

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

優點：

更符合常規的思維。將數據送給想要的部門

缺點：

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

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

7.2.2 解耦合

藉助canal，監聽訂單表的數據變化，不再影響主業務。

file

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。

file

註：

關於druid的詳細使用，在大數據篇章里會詳細講解。

7.3 kafka監控

7.3.1 eagle簡介

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

file

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

file

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

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

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