使用 spring stream 发送消息
2021-01-28 13:15
标签：ati start create star 消息队列 代码 cloud ring 为什么 spring stream 是用来做消息队列发送消息使用的。他隔离了各种消息队列的区别，使用统一的编程模型来发送消息。 目前支持： rabbitmq kafka rocketmq rocketmq 支持windows 在应用上增加注解 @EnableBinding({InputOutput.class}) 这里发送的是两类消息。 接收消息： 分别接收两类消息 使用 spring stream 发送消息 标签：ati start create star 消息队列 代码 cloud ring 为什么 原文地址：https://www.cnblogs.com/yg_zhang/p/12837879.html为什么使用spring stream ?
启动rocketmq
start mqnamesrv.cmd
start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true
修改pom.xml
增加发送接收JAVA代码
public interface InputOutput {
String MAIL_OUTPUT = "mailOutput";
String MAIL_INPUT = "mailInput";
String OUTPUT = "output";
String INPUT = "input";
@Output(OUTPUT)
MessageChannel output();
@Input(INPUT)
SubscribableChannel input();
@Output(MAIL_OUTPUT)
MessageChannel mailOutput();
@Input(MAIL_INPUT)
SubscribableChannel mailInput();
}
增加yml配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 127.0.0.1:9876
bindings:
output:
destination: bpmmessage
group: bpmmessage-group
input:
destination: bpmmessage
group: bpmmessage-group-consumer
mailOutput:
destination: mail
group: mail-group
mailInput:
destination: mail
group: mail-group-consumer
编写代码收发消息：
MessageModel messageModel=new MessageModel();
messageModel.setMsgType("mail");
messageModel.setContent("helloworld");
inputOutput.mailOutput().send( MessageBuilder.withPayload(
"mail"
).build());
inputOutput.output().send(
MessageBuilder.withPayload(
messageModel
).build()
);
@Service
public class MessageListener {
@StreamListener(InputOutput.INPUT)
public void receive(MessageModel message) {
System.err.println(message);
System.err.println("ok");
}
@StreamListener(InputOutput.MAIL_INPUT)
public void receive(String message) {
System.err.println(message);
System.err.println("ok");
}
}
文章标题：使用 spring stream 发送消息
文章链接：http://soscw.com/index.php/essay/48244.html