為什么使用spring stream ?

spring stream 是用來做消息隊列發(fā)送消息使用的。他隔離了各種消息隊列的區(qū)別，使用統(tǒng)一的編程模型來發(fā)送消息。

目前支持：

啟動rocketmq

rocketmq 支持windows

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

修改pom.xml

<dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-stream-binder-rocketmq</artifactId> </dependency>

增加發(fā)送接收J(rèn)AVA代碼

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();}

在應(yīng)用上增加注解

@EnableBinding({InputOutput.class})

增加yml配置

spring: cloud: stream: rocketmq: binder: name-server: 127.0.0.1:9876 bindings: output: destination: bpmmessage group: bpmmessage-groupinput: destination: bpmmessage group: bpmmessage-group-consumermailOutput: destination: mail group: mail-groupmailInput:destination: mailgroup: mail-group-consumer

編寫代碼收發(fā)消息：

MessageModel messageModel=new MessageModel(); messageModel.setMsgType('mail'); messageModel.setContent('helloworld'); inputOutput.mailOutput().send( MessageBuilder.withPayload('mail' ).build()); inputOutput.output().send(MessageBuilder.withPayload( messageModel).build() );

這里發(fā)送的是兩類消息。

接收消息：

@Servicepublic 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'); }}

分別接收兩類消息

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。