Spring Cloud Stream 整合 RabbitMQ

简介

Spring Cloud Stream 是一个构建消息驱动微服务的框架，应用程序经过input（至关于consumer）、output（至关于producer）来与Spring Cloud Stream中Binder交互，而Binder负责与消息中间件交互；所以，咱们只需关注如何与Binder交互便可，而无需关注与具体消息中间件的交互。

快速上手

一、添加依赖

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
  <version>2.1.2.RELEASE</version>
</dependency>
复制代码

二、配置文件

provider配置（采用动态路由键方式）

server:
  port: 7071
spring:
  cloud:
    stream:
      binders:
        pro:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: localhost
                port: 5672
                username: test
                password: test
                virtual-host: test
      bindings:
        myOutPut:
          destination: myOutPut
          content-type: application/json
          default-binder: test
      rabbit:
        bindings:
          myOutPut:
            producer:
              exchangeType: topic
              routing-key-expression: headers.routeId
复制代码

consumer配置

server:
  port: 7072
spring:
  cloud:
    stream:
      rabbit:
        bindings:
          input:
            consumer:
              bindingRoutingKey: routeKey1
              acknowledge-mode: manual
      binders:
              protest:
                type: rabbit
                environment:
                  spring:
                    rabbitmq:
                      addresses: localhost
                      port: 5672
                      username: test
                      password: test
                      virtual-host: test
      bindings:
              input:
                destination: myOutPut
                content-type: application/json
                default-binder: protest
                group: group-cus1
复制代码

三、自定义通道

provider

public interface MqMessageSource {//自定义通道
    String MY_OUT_PUT = "myOutPut";
    @Output(MY_OUT_PUT)
    MessageChannel testOutPut();
}
复制代码

四、消息发送

@EnableBinding(MqMessageSource.class)
public class MessageProviderImpl implements IMessageProvider {
    @Autowired
    @Output(MqMessageSource.MY_OUT_PUT)
    private MessageChannel channel;

    @Override
    public void send(Company company) {
        channel.send(MessageBuilder.withPayload(company).setHeader("routeId", company.getTitle()).build());
    }
}
复制代码

五、消息接收

@Component
@EnableBinding(Sink.class)
public class MessageListener {
    @StreamListener(Sink.INPUT)
    public void input(Message<Company> message) throws IOException {
        Channel channel = (com.rabbitmq.client.Channel)message.getHeaders().get(AmqpHeaders.CHANNEL);
        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
        System.err.println(JSON.toJSONString(message.getPayload()));
    }
}
复制代码