SpringCloud学习之Stream消息驱动自定义通道

Posted 2021-03-15 xulijun137

如果不清楚本篇内容的，请务必先去看完上一篇再看本篇，否则阅读起来可能会有部分障碍和困难；

上一篇文章《SpringCloud学习之Stream消息驱动【默认通道】（十）》我们简单用自定义通道实现了消息发送和接收，但是用的是Stream给我们提供的默认Source，Sink，接下来我们要自己进行自定义，这种方式在工作中还是用的比较多的，因为我们要往不同的消息通道发消息，必然不能全都叫input,output的，那样的话就乱套了

（一）创建消息生产者【service-sender-stream-8089】

?

MySource.java

package com.xu.serviceconsumer.interfaces;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;

public interface MySource {

    String INPUT_CHANNEL = "myInput";
    String OUTPUT_CHANNEL = "myOutput";

    /**
     * 输入通道
     * @return
     */
    @Input(MySource.INPUT_CHANNEL)
    SubscribableChannel input();

    /**
     * 输出通道
     * @return
     */
    @Output(MySource.OUTPUT_CHANNEL)
    MessageChannel output();

}

?

application.yml

server:
  port: 8089
spring:
  application:
    name: spring-cloud-stream-sender
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment: #配置rabbimq连接环境
            spring:
              rabbitmq:
                host: localhost
                username: guest
                password: guest
                virtual-host: /
      bindings:
        myOutput:       #自定义输出通道
          destination: stream-demo       #exchange名称，交换模式默认是topic
          content-type: text/plain       #消息发送的格式，接收端不用指定格式，但是发送端要

 

发送消息接口不变

?

接口实现需要重新改造：

?

 

(二）消息消费者【service-consumer-stream-8090和service-consumer-stream-8091】

这两个消费客户端的配置基本一模一样的，只是application.yml中的端口略有不同

?

 

因为我们这个测试项目没有公共依赖模块，所以暂时把消息生产端中的MySource.java这个自定义通道类文件复制放到两个客户端模块里

?

 

我们还要重新改造消息消费者里的代码，接口定义不变：

?

?

重新Rebuild三个项目模块，然后重新启动三个模块，打开消息生产者swagger页面http://localhost:8089/swagger-ui.html重新生产一个消息到消息队列，我们依旧可以看到两个客户端也接收到了发送过来的消息：

?

?

?

至此我们完成了自定义通道消息发送和接收，spring cloud stream还有很多东西（比如分组group和分区partition），后面有空我再深入了解后补充说明，谢谢大家。