当前位置：首页 > news >正文

spring cloud alibaba RocketMQ 最佳实践

news 2025/7/18 16:18:50

概述

github 文档地址
rocket mq example

RocketMQ 版本为 5.1.4

使用

准备工作

阅读此文需要事先准备 RocketMQ ，如有疑问，请移步 RocketMQ 服务搭建

引入依赖

此处，多依赖了一些依赖。

<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
</dependency>

创建Topic

sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic

在这里插入图片描述

代码

配置 Input 和 Output 的 Binding 信息并配合 @EnableBinding 注解使其生效

@EnableBinding({ Source.class, Sink.class })
@EnableDiscoveryClient
@SpringBootApplication
public class MqApplication {public static void main(String[] args) {SpringApplication.run(MqApplication.class);}
}

配置 Binding 信息：

# 配置rocketmq的nameserver地址
spring.cloud.stream.rocketmq.binder.name-server=10.xx.xx.143:9876
# 定义name为output的binding
spring.cloud.stream.bindings.output.destination=test-topic
spring.cloud.stream.bindings.output.content-type=application/json
# 定义name为input的binding
spring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=application/json
spring.cloud.stream.bindings.input.group=test-group

发送消息代码：

public class RocketMQProducer {public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer("producer_group");producer.setNamesrvAddr("10.xx.xx.143:9876");producer.start();Message msg = new Message("test-topic", "tagStr", "message from rocketmq producer".getBytes());producer.send(msg);}
}

应用启动

1.增加配置，在应用的 /src/main/resources/application.properties 中添加基本配置信息

spring:application:name: mq
server:port: 9700

2.启动应用，支持 IDE 直接启动和编译打包后启动。

IDE 直接启动：找到主类 RocketMQApplication，执行 main 方法启动应用。
打包编译后启动：首先执行 mvn clean package 将工程编译打包，然后执行 java -jar rocketmq-example.jar 启动应用。

消息接收

使用 name 为 output 对应的 binding 发送消息到 test-topic 这个 topic。

@Service
public class ReceiveService {@StreamListener("input1")public void receiveInput1(String receiveMsg) {System.out.println("input1 receive: " + receiveMsg);}
}

在这里插入图片描述

再扩展一个

一个服务中要使用多个 Topic ，这种情况如何解决？下面给出解决方案。

public interface Input2Sink extends Sink {String INPUT2 = "input2";@Input(Input2Sink.INPUT2)SubscribableChannel input2();
}

public interface Output2Source extends Source {String OUTPUT2 = "output2";@Output(Output2Source.OUTPUT2)MessageChannel output2();
}

@Service
public class ReceiveService {@StreamListener("input")public void receiveInput1(String receiveMsg) {System.out.println("input receive: " + receiveMsg);}@StreamListener("input2")public void receiveInput2(String receiveMsg) {System.out.println("input2 receive: " + receiveMsg);}
}

@EnableBinding(value = {Source.class, Sink.class, Output2Source.class, Input2Sink.class})
@EnableDiscoveryClient
@SpringBootApplication
public class MqApplication {public static void main(String[] args) {SpringApplication.run(MqApplication.class);}
}