使用spring stream发送消息代码实例

使用spring stream发送消息代码实例

为什么使用spring stream ?

spring stream 是用来做消息队列发送消息使用的。他隔离了各种消息队列的区别，使用统一的编程模型来发送消息。

目前支持：

rabbitmq

kafka

rocketmq

启动rocketmq

rocketmq 支持windows

start mqnamesrv.cmd

start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

修改pom.xml

com.alibaba.cloud

spring-cloud-stream-binder-rocketmq

增加发送接收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();

}

在应用上增加注解

@EnableBinding({InputOutput.class})

增加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");

}

}

分别接收两类消息

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。