光说不练假把式，一起Kafka业务实战。

2023-11-25 16:10
文章标签 实战 业务 kafka 一起 假把式 光说不练

本文主要是介绍光说不练假把式，一起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

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-servercontainer_name: canalrestart: alwaysports:- "10908:11111"environment:#mysql的链接信息canal.instance.master.address: 52.82.98.209:3389canal.instance.dbUsername: canalcanal.instance.dbPassword: canal#投放到kafka的哪个主题？要提前准备好！canal.mq.topic: canalvolumes:- "/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.13kafka-1:container_name: kafka-1image: wurstmeister/kafka:2.12-2.2.2ports:- 10903:9092- 10913:10913environment:KAFKA_BROKER_ID: 1 HOST_IP: 52.82.98.209KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181#docker部署必须设置外部可访问ip和端口，否则注册进zk的地址将不可达造成外部无法连接KAFKA_ADVERTISED_HOST_NAME: 52.82.98.209KAFKA_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: 10913volumes:- /etc/localtime:/etc/localtimedepends_on:- zookeeper           kafka-2:container_name: kafka-2image: wurstmeister/kafka:2.12-2.2.2ports:- 10904:9092- 10914:10914environment:KAFKA_BROKER_ID: 2 HOST_IP: 52.82.98.209KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_ADVERTISED_HOST_NAME: 52.82.98.209KAFKA_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: 10914volumes:- /etc/localtime:/etc/localtimedepends_on:- zookeeper eagle:image: gui66497/kafka_eaglecontainer_name: kerestart: alwaysdepends_on:- kafka-1- kafka-2ports:- "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需要配置一定的连接信息

专注Java技术干货分享，欢迎志同道合的小伙伴，一起交流学习

这篇关于光说不练假把式，一起Kafka业务实战。的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/424099。

相关文章

PyTorch中的词嵌入层(nn.Embedding)详解与实战应用示例

PyTorch中的词嵌入层(nn.Embedding)详解与实战应用示例

《PyTorch中的词嵌入层(nn.Embedding)详解与实战应用示例》词嵌入解决NLP维度灾难,捕捉语义关系,PyTorch的nn.Embedding模块提供灵活实现,支持参数配置、预训练及变长... 目录一、词嵌入(Word Embedding)简介为什么需要词嵌入？二、PyTorch中的nn.Em
阅读更多...
在IntelliJ IDEA中高效运行与调试Spring Boot项目的实战步骤

在IntelliJ IDEA中高效运行与调试Spring Boot项目的实战步骤

《在IntelliJIDEA中高效运行与调试SpringBoot项目的实战步骤》本章详解SpringBoot项目导入IntelliJIDEA的流程,教授运行与调试技巧,包括断点设置与变量查看,奠定... 目录引言：为良驹配上好鞍一、为何选择IntelliJ IDEA？二、实战：导入并运行你的第一个项目步骤1
阅读更多...
Spring Boot3.0新特性全面解析与应用实战

Spring Boot3.0新特性全面解析与应用实战

《SpringBoot3.0新特性全面解析与应用实战》SpringBoot3.0作为Spring生态系统的一个重要里程碑,带来了众多令人兴奋的新特性和改进,本文将深入解析SpringBoot3.0的... 目录核心变化概览Java版本要求提升迁移至Jakarta EE重要新特性详解1. Native Ima
阅读更多...
Spring Boot 与微服务入门实战详细总结

Spring Boot 与微服务入门实战详细总结

《SpringBoot与微服务入门实战详细总结》本文讲解SpringBoot框架的核心特性如快速构建、自动配置、零XML与微服务架构的定义、演进及优缺点,涵盖开发环境准备和HelloWorld实战... 目录一、Spring Boot 核心概述二、微服务架构详解1. 微服务的定义与演进2. 微服务的优缺点三
阅读更多...
SpringBoot集成MyBatis实现SQL拦截器的实战指南

SpringBoot集成MyBatis实现SQL拦截器的实战指南

《SpringBoot集成MyBatis实现SQL拦截器的实战指南》这篇文章主要为大家详细介绍了SpringBoot集成MyBatis实现SQL拦截器的相关知识,文中的示例代码讲解详细,有需要的小伙伴... 目录一、为什么需要SQL拦截器？二、MyBATis拦截器基础2.1 核心接口：Interceptor
阅读更多...
从入门到进阶讲解Python自动化Playwright实战指南

从入门到进阶讲解Python自动化Playwright实战指南

《从入门到进阶讲解Python自动化Playwright实战指南》Playwright是针对Python语言的纯自动化工具,它可以通过单个API自动执行Chromium,Firefox和WebKit... 目录Playwright 简介核心优势安装步骤观点与案例结合Playwright 核心功能从零开始学习
阅读更多...
Java docx4j高效处理Word文档的实战指南

Java docx4j高效处理Word文档的实战指南

《Javadocx4j高效处理Word文档的实战指南》对于需要在Java应用程序中生成、修改或处理Word文档的开发者来说,docx4j是一个强大而专业的选择,下面我们就来看看docx4j的具体使用... 目录引言一、环境准备与基础配置1.1 Maven依赖配置1.2 初始化测试类二、增强版文档操作示例2.
阅读更多...
MySQL 多列 IN 查询之语法、性能与实战技巧(最新整理)

MySQL 多列 IN 查询之语法、性能与实战技巧(最新整理)

《MySQL多列IN查询之语法、性能与实战技巧(最新整理)》本文详解MySQL多列IN查询,对比传统OR写法,强调其简洁高效,适合批量匹配复合键,通过联合索引、分批次优化提升性能,兼容多种数据库... 目录一、基础语法：多列 IN 的两种写法1. 直接值列表2. 子查询二、对比传统 OR 的写法三、性能分析
阅读更多...
Python办公自动化实战之打造智能邮件发送工具

Python办公自动化实战之打造智能邮件发送工具

《Python办公自动化实战之打造智能邮件发送工具》在数字化办公场景中,邮件自动化是提升工作效率的关键技能,本文将演示如何使用Python的smtplib和email库构建一个支持图文混排,多附件,多... 目录前言一、基础配置：搭建邮件发送框架1.1 邮箱服务准备1.2 核心库导入1.3 基础发送函数二、
阅读更多...
PowerShell中15个提升运维效率关键命令实战指南

PowerShell中15个提升运维效率关键命令实战指南

《PowerShell中15个提升运维效率关键命令实战指南》作为网络安全专业人员的必备技能,PowerShell在系统管理、日志分析、威胁检测和自动化响应方面展现出强大能力,下面我们就来看看15个提升... 目录一、PowerShell在网络安全中的战略价值二、网络安全关键场景命令实战1. 系统安全基线核查
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2