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);
    }
}