当前位置：首页 > 编程日记 > 正文

active mq topic消费后删除_Spring cloud stream 整合mq

编程日记 2024-06-21 16:00:00

说明：本案例win10环境

测试scs(spring cloud stream)整合kfk(kafka)/rbt(rabbitmq)消息生产消费场景

流程

一、准备中间件环境(kfk/rbt)

后续内容提供kfk与rbt的环境准备流程

二、导入scs的依赖

项目spring boot 版本：2.1.3.RELEASE

引入spring cloud 版本：Greenwich.SR3 支持 spring boot 2.1.x.RELEASE

SR3：Service Release

POM相关依赖

<properties>

三、编写配置文件

spring

四、组织测试代码

五、启动实例测试

启动服务顺序

zookeeper服务
kafka服务
kafka生产者服务
kafka消费者服务
rabbitmq服务
启动boot项目

kfk组件原理

Ⅰ. CONCEPT OF KFK

Broder：kfk实例.
Producer：msg生产者.
Consumer：msg消费者.
Topic：msg主题，用于消息路由策略.
Partition：topic分区，如设置值为3，体现形式为3个文件夹.
Consumer group：消费者组，设置消费组后可以实现msg不重复消费.
Replication：分区的副本，follower和leader在不同机器上，设计与es雷同，如下图Topic C.Ⅱ. DESIGN OF KFK

rbt组件原理

Ⅰ. CONCEPT OF RBT
Message：由消息头和消息体组成.
Publisher：msg生产者.
Consumer：msg消费者.
Exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列.
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
Binding：绑定器，用于关联交换器与消息队列.
Queue：消息队列容器.
Connection：socket连接.
Channel：信道，使用多路复用技术-双向数据流通道 (通信连接复用).
Virtual Host：mini 版的 rbt服务器，拥有独立的队列、交换器、绑定和权限机制，vh是 AMQP 概念的基础.
Broker：mq服务器实体，它的角色作用是维护消息流水线，保证数据能按照指定的方式进行传输.Ⅱ. DESIGN OF RBT

【概念统一】
destination(scs)=topic(kfk)=exchange(rbt)

kfk 环境

一、文件配置 (可选项)
1. zookeeper
A. 配置文件<zoo_sample.cfg>克隆一份名称为<zoo.cfg>
B. 数据目录：# dataDir=D:swzookeeper-3.4.14data
2. kafka
A. 配置日志目录：# log.dirs=kafka-logs

二、应用启动 (安装目录下)
1. zookeeper
$.binzkServer.cmd
2. kafka
$.binwindowskafka-server-start.bat .configserver.properties

三、创建主题
$.binwindowskafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic example

四、查看主题
$.binwindowskafka-topics.bat --list --zookeeper localhost:2181

五、启动生产者
$.binwindowskafka-console-producer.bat --broker-list localhost:9092 --topic example

六、启动消费者
$.binwindowskafka-console-consumer.bat --bootstrap-server localhost:9092 --topic example --from-beginning

七、改变分区数
$.binwindowskafka-topics.bat --zookeeper localhost:2181 --alter --partitions 2 --topic example-kfk3

八、参数说明
--topic 主题名称，scs中概念为 destination
--partitions 指定分区个数，scs中概念为 partition-count
--replication-factor 指定副本因子
--create 创建主题
--alter 改动操作

九、问题日志
（1）kafka与zookeeper版本需要兼容
（2）配置文件中路径的分隔符注意转义
（3）java应用启动如不需要连接数据库加
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)（4）多个监听实例重复消费mq问题需求：一份msg，多个实例监听，只能被一个实例消费，并且为轮流消费策略解决方案：生产者设置分区数量n，并在cfg中设置消费者分组，则可被n个实例轮流不重复消费

rbt 环境

一、下载
下载rbt及依赖的erlang，并配置环境变量

二、添加插件
MQ可视化管理插件：rabbitmq-plugins enable rabbitmq_management

三、访问环境http://localhost:15672/

四、操作步骤
1. 创建exchange
2. 设置推送策略
3. 绑定queue
4. 发送消息

https://www.dkcj.cn/info/11703.html

流程