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>
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 直接启动和编译打包后启动。
使用 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);
}
}
结果如下:
spring cloud alibaba RocketMQ 最佳实践 ,至此就结束了,如有疑问,欢迎评论区留言。