本章节将演示 RocketMQ 的基本使用,以及 RocketMQ 与 Spring Cloud Stream 结合后的基本使用,以广播消费为例。
RocketMQ 基本使用
安装并启动
-
下载 RocketMQ 最新的二进制文件,并解压。
解压后的目录结构如下:
-
启动 NameServer
Linux / Unix / Mac 操作系统,执行命令 :
3. 启动 Broker
Linux / Unix / Mac 操作系统,执行命令 :
测试发送与接受消息
- 创建测试主题(TopicTest):
Linux / Unix / Mac 操作系统,执行命令 :
2. 发送消息:
Linux / Unix / Mac 操作系统,执行命令 :
3. 接收消息:
Linux / Unix / Mac 操作系统,执行命令 :
Windows 操作系统,执行命令:
发送成功后显示:SendResult [sendStatus=SEND_OK, msgId= …
接收成功后显示:ConsumeMessageThread_%d Receive New Messages: [MessageExt…
广播消费示例
广播会发送消息给所有消费者。如果你想同一消费组下所有消费者接收到同一个 topic 下的消息,广播消费非常适合此场景。
引入依赖
需要在 pom.xml
文件中引入 group ID 为 com.alibaba.cloud
和 artifact ID 为 spring-cloud-starter-stream-rocketmq
的 starter:
更新主题(topic)
更新 RocketMQ 中的主题(topic)为广播模式。
Linux / Unix / Mac 操作系统,执行命令 :
Windows 操作系统,执行命令:
创建生产者
- 相关配置
在应用的 /src/main/resources/application.yaml
配置文件中完成 RocketMQ 广播消息生产者的相关配置。如下所示,对服务端口号、RocketMQ 的 NameServer 地址、消息发送的目标进行配置:
- 发送消息
通过@Autowired
注解,将 StreamBridge 注入到应用程序中,以发送消息。在 producer 方法中,使用循环发送 100 条消息。每条消息都包含一个 key 和 headers,并创建一个 GenericMessage 对象作为消息体,然后使用 streamBridge.send 方法发送消息到名为 “producer-out-0” 的目标。
创建消费者
启动两个消费者实例。
消费者 1
- 相关配置
在应用的 /src/main/resources/application.yaml
配置文件中完成 RocketMQ 广播消息消费者 1 的相关配置。如下所示,对服务端口号、消费者函数、RocketMQ 的 NameServer 地址、消息消费模式以及消费者分组进行配置:
- 接收消息
通过 @Bean
注解标记的 consumer 方法返回一个 Consumer 对象,用于处理接收到的消息。在 consumer 方法中,通过 Lambda 表达式定义了一个消费者函数,当接收到新消息时,打印出消息内容。
消费者 2
消费者 2 创建过程的详细解释可参考上述消费者 1 的创建过程的解释。
- 相关配置
application.yml
配置
- 接收消息
除了上述的广播消费应用,还可以进行顺序消费,延时消费,过滤消息,事务消息等应用,关于更多 RocketMQ 的简单示例,可以参考 RocketMQ 示例。
此 Demo 仅演示了 RocketMQ 与 Spring Cloud Stream 结合后的基本使用,更多 RocketMQ 的高级特性,请参考 RocketMQ 进阶指南章节!