阅读数:82 评论数:0

目标是屏蔽各种MQ的差异，统一编程模型。spring cloud stream 主要是使用binder和binding。

Binder表示绑定具体哪一个消息中间件，Binding包括Input Binding和Output Binding。Binding在消息中间件与应程序提供的Provider和Consumer之间提供了一个桥梁，实现了开发者只需使用应用程序的Provider或Consumer生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。

yml配置

设置input output 这里的名字可以自己定义，需要跟注解定义的对应起来。

spring:
  cloud:
    stream:
      rocketmq:
        binder:
          namesrv-addr: 127.0.0.1:9876
# 定义name为output的binding
      bindings:
        output1:
          destination: test-topic
          content-type: application/json
# 定义name为input的binding
        input1:
          destination: test-topic
          content-type: application/json
          group: test-group
server:
  port: 1000

public interface StreamClient {

    @Input("input1")
    SubscribableChannel input1();

    @Output("output1")
    MessageChannel output1();
}

编写接收者

@Component
@EnableBinding(StreamClient.class)
@Slf4j
public class MQReciver {

    @StreamListener("input1")
    public void test(String message){       
        log.info(message);
    }
}

编写消息发送者

@RestController
@RequestMapping("test")
public class MQRest {

    @Autowired
    StreamClient streamClient;

    @GetMapping("mq")
    public void test(){
        String now = "now" + new Date();
        streamClient.output1().send(MessageBuilder.withPayload(now).build());
    }
}