springcloudstream整合rabbitmq

1、通過docker安裝ribbitmq

拉取鏡像
docker pull rabbitmq:3.8.6-management

運行鏡像

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 --hostname myRabbit  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:3.8.6-management

2、生產者代碼

添加依賴

 <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
            <version>3.0.7.RELEASE</version>
        </dependency>

添加配置文件

spring:
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: 192.168.178.128
                port: 5672
                username: admin
                password: admin
      bindings:
        output: #通道的名稱
          destination: exchange1 #表示要使用的exchange名稱的定義
          content-type: application/json
          binder: defaultRabbit

編寫接口

public interface MessageProvider {
    public String send();
}

編寫實現類

@EnableBinding(Source.class)
public class IMessageSendProvider implements MessageProvider {

    @Resource
    private Source source;
    @Override
    public String send() {
        source.output().send(MessageBuilder.withPayload(UUID.randomUUID().toString()).build());
//        boolean send = messageChannel.send(MessageBuilder.withPayload(UUID.randomUUID().toString()).build());
        return "success";
    }
}

3、消費者代碼

添加依賴

 <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
            <version>3.0.7.RELEASE</version>
        </dependency>

添加配置文件

spring:
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: 192.168.178.128 port: 5672 username: admin password: admin bindings: input: #通道的名稱 destination: exchange1 #表示要使用的exchange名稱的定義 content-type: application/json binder: defaultRabbit

編寫代碼

 

@EnableBinding(Sink.class)
@Service
public class LianxiCuseromer {

    @StreamListener(Sink.INPUT)
    public void test(Message<String> message){
        String payload = message.getPayload();
        System.out.println(payload);
    }
}