关于java:光说不练假把式一起Kafka业务实战

38次阅读

共计 4702 个字符，预计需要花费 12 分钟才能阅读完成。

假如咱们要传输一批订单到另一个零碎，那么订单对应状态的演变是有程序性要求的。

已下单 → 已领取 → 已确认

不容许错乱！

1）全局有序：

串行化。每条通过 kafka 的音讯必须严格保障有序性。

这就要求 kafka 单通道，每个 groupid 下单消费者

极大的影响性能，事实业务下简直没必要

2）部分有序：

业务部分有序。同一条订单有序即可，不同订单能够并行处理。不同订单的程序前后无所谓

充分利用 kafka 多分区的并发性，只须要想方法让须要程序的一批数据进同一分区即可。

1）发送端：

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

2）发送中：

给队列配置多分区保障并发性。

3）读取端：

单消费者：显然不合理

吞吐量显然上不去，kafka 开多个分区还有何意义？

所以开多个消费者指定分区生产，现实情况下，每个分区配一个。

然而，这个吞吐量仍然无限，那如何解决呢？

计划：多线程

在每个消费者上再开多线程，是个解决办法。然而，要警觉程序性被突破！

参考下图：thread 解决后，会将 data 变成 2-1-3

改良：接管后散发二级内存队列

消费者取到音讯后不做解决，依据 key 二次散发到多个阻塞队列。

再开启多个线程，每个队列调配一个线程解决。晋升吞吐量

1）新建一个 sort 队列，2 个分区

2）启动 order 我的项目

源码参考：

SortedProducer（程序性发送端）

SortedConsumer（程序性生产端 – 阻塞队列实现，不便大家了解设计思路）

SortedConsumer2（程序性生产端 – 线程池实现，事实中举荐这种形式！）

3）通过 swagger 申请

先按不同的 id 发送，查看控制台日志，id 被正确散发到对应的队列

同一个 key 调配到同一个 queue，程序性失去保障

假如大数据部门须要大屏来展现用户的打车订单状况，须要把订单数据送入 druid

这里不波及程序，只有下单就传输，然而对实时性和并发量要求较高

在下单实现 mysql 后，通过程序代码打印，间接进入 kafka

或者 logback 和 kafka 集成，通过 log 输送

长处：

更合乎惯例的思维。将数据送给想要的部门

毛病：

耦合度高，将 kafka 发送音讯嵌入了订单下单的主业务，造成代码入侵。

下单不关怀，也不应该关注送入 kafka 的状况，一旦 kafka 不可用，程序受影响

借助 canal，监听订单表的数据变动，不再影响主业务。

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 的具体应用，在大数据篇章里会具体解说。

Kafka Eagle 监控零碎是一款用来监控 Kafka 集群的工具，反对治理多个 Kafka 集群、治理 Kafka 主题（蕴含查看、删除、创立等）、消费者组合消费者实例监控、音讯阻塞告警、Kafka 集群衰弱状态查看等。

举荐 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

拜访：http://52.82.98.209:10907/ke/

默认用户名明码：admin / 123456

如果要删除 topic 等操作，须要治理 token：keadmin

与 km 到底选哪个呢？依据本人习惯，集体认为：

界面好看水平和监控曲线优于 km，有登录权限管制
性能操作上不如 km 简略直白，然而 km 须要配置肯定的连贯信息

本文由传智教育博学谷 – 狂野架构师教研团队公布，转载请注明出处！

如果本文对您有帮忙，欢送关注和点赞；如果您有任何倡议也可留言评论或私信，您的反对是我保持创作的能源

正文完

java

发表至： java

2022-08-16

0

关于java:详细介绍java关键字super

关于java:1wumeismart系统运行

关于java:我们来聊聊锁升级吧

转载Spring-Boot-设置项目名后静态文件相对路径问题

关于macos:ColorWell-web网页颜色代码取色工具

关于java:光说不练假把式一起Kafka业务实战

7.1 程序性场景

7.1.1 场景概述

7.1.2 程序级别

7.1.3 实现计划

7.1.4 代码验证

7.2 海量同步场景

7.2.1 惯例架构

7.2.2 解耦合

7.2.3 部署实现

7.3 kafka 监控

7.3.1 eagle 简介

7.3.2 部署

7.3.3 应用阐明

Just My Socks（注册教程内含优惠码）

关于java:光说不练假把式一起Kafka业务实战

7.1 程序性场景

7.1.1 场景概述

7.1.2 程序级别

7.1.3 实现计划

7.1.4 代码验证

7.2 海量同步场景

7.2.1 惯例架构

7.2.2 解耦合

7.2.3 部署实现

7.3 kafka 监控

7.3.1 eagle 简介

7.3.2 部署

7.3.3 应用阐明

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）