标签：ati start create star 消息队列 代码 cloud ring 为什么
spring stream 是用来做消息队列发送消息使用的。他隔离了各种消息队列的区别，使用统一的编程模型来发送消息。
目前支持：
rabbitmq
kafka
rocketmq
rocketmq 支持windows
start mqnamesrv.cmd start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
</dependency>
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(); }
在应用上增加注解
@EnableBinding({InputOutput.class})
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"); } }
分别接收两类消息
标签：ati start create star 消息队列 代码 cloud ring 为什么
原文地址：https://www.cnblogs.com/yg_zhang/p/12837879.html
