SpringCloud stream连接RabbitMQ收发信息

Posted 2021-01-26 七彩的人生

 1、POM引入spring-cloud-starter-stream-rabbit

注意springboot 版本，用1.5.2.RELEASE 就报错 rabbitmqTemple not found

sb版本切换到1.5.14 就好了

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.14.RELEASE</version>
        <relativePath /> <!-- lookup parent from repository -->
    </parent>

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

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Edgware.SR4</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

View Code

 

 

2、Bindings

public interface OrderStream {

    
    @Output(ORDER_OUT)
    MessageChannel outOrder();

    @Input(ORDER_INPUT)
    SubscribableChannel inputOrder();
    
    
    @Output(ORDER_OUT_R)
    MessageChannel outOrderR();


    @Input(ORDER_INPUT_R)
    SubscribableChannel inputOrderR();
}

 

ORDER_OUT 用用于发送消息，ORderOUT_R 用于接收消息回调

如果input 里面的名称 和 output 配置一样会报错

3、发送
指定bindings

@SuppressWarnings("SpringJavaAutowiringInspection")
@EnableBinding({OrderStream.class})
public class OrderMsgProceducer {


    private static final Logger log = LoggerFactory.getLogger(OrderMsgProceducer.class);
    @Autowired
    @Qualifier(OrderStream.ORDER_OUT)
    private MessageChannel messageChannel;

    public void send(Order timingEvent) {
        Message<Order> message = MessageBuilder.withPayload(timingEvent).build();
        messageChannel.send(message);
        log.info("send Mq:{}",message);
    }
}

 

4、接受 与回调

之所以能够接受到是因为

接受的与发送的是一个

 

@SuppressWarnings("SpringJavaAutowiringInspection")
@EnableBinding({OrderStream.class})
public class OrderMsgConsumer {

    private final static Logger log = LoggerFactory.getLogger(OrderMsgConsumer.class);

    @StreamListener(OrderStream.ORDER_OUT)
    @SendTo(OrderStream.ORDER_OUT_R)
    public String receiver(@Payload Order order) throws Exception {

        log.info("Receive Order To MQ:{}",order);
        log.debug("MQ END:{}",order);
        return "我收到了";
    }
    
    @StreamListener(OrderStream.ORDER_OUT_R)
    public void receiverHuidiao(String msg) throws Exception {
        System.out.println("我接收回调......");
        System.out.println(msg);
    }
}

 

配置文件

spring.rabbitmq.host=1xxx
spring.rabbitmq.port=5673
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.cloud.stream.bindings.ORDER_SEND.content-type=application/json;charset=UTF-8
spring.cloud.stream.bindings.ORDER_SEND.producer.deliveryMode=PERSISTENT
spring.cloud.stream.bindings.ORDER_SEND.destination=ORDER_14
spring.cloud.stream.bindings.ORDER_SEND.group=becom
spring.cloud.stream.bindings.ORDER_SEND.consumer.maxAttempts=3
spring.cloud.stream.bindings.ORDER_SEND.consumer.backOffInitialInterval=1000
spring.cloud.stream.bindings.ORDER_SEND.consumer.concurrency=1

group作用：如果我们启动两个服务，能够确保我们发送一个消息只被消费一次

 
测试

 

 

mq页面

 

项目：