使用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

<dependency>
      <groupid>com.alibaba.cloud</groupid>
      <artifactid>spring-cloud-stream-binder-rocketmq</artifactid>
    </dependency>

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

分别接收两类消息

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。