SpringCloud stream連接RabbitMQ收發信息

百度上查的大部分都是一些很簡單的單消費者或者單生產者的例子，並且多是同一個服務器的配置，本文的例子為多服務器配置下的消費生產和消費者配置。

參考資料：https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/#_binder_implementations

 

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

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

2、application.properties

通用配置：

#rabbit的配置信息
spring.rabbitmq.addresses=amqp://10.18.75.231:5672
spring.rabbitmq.username=user_admin
spring.rabbitmq.password=12345678
#下面這個配置優先級太高，在配置中心分模塊(分文件)的場景下后面的binder屬性無法被覆蓋，如果有存在多個vhost的情況下建議將該屬性注釋掉
spring.rabbitmq.virtual-host=boss

當存在多個binder時必須指定一個默認的binder：

# 設置一個默認的binder，如果不配置將報錯
spring.cloud.stream.defaultBinder=boss

消費者配置： 1 # 配置ecm消費者的服務器配置信息

spring.cloud.stream.binders.ecm.type=rabbit  3 #spring.cloud.stream.binders.ecm.environment.spring.rabbitmq.addresses=${spring.rabbitmq.addresses}  4 #spring.cloud.stream.binders.ecm.environment.spring.rabbitmq.username=${spring.rabbitmq.username}  5 #spring.cloud.stream.binders.ecm.environment.spring.rabbitmq.password=${spring.rabbitmq.password}  6 spring.cloud.stream.binders.ecm.environment.spring.rabbitmq.virtual-host=ecm  7 
#交易系統ECM的貨櫃模板變更消費者
spring.cloud.stream.bindings.ecm_shop_template.binder=ecm 10 spring.cloud.stream.bindings.ecm_shop_template.destination=這里填exchange的名字 11 #默認情況下同一個隊列的只能被同一個group的消費者消費
spring.cloud.stream.bindings.ecm_shop_template.group=這里是消費者的名稱 13 spring.cloud.stream.bindings.ecm_shop_template.contentType=text/plain 14 #指定該主題的類型為廣播模式
spring.cloud.stream.rabbit.bindings.ecm_shop_template.consumer.exchangeType=fanout 16 #消費失敗的消息放入dlq隊列
spring.cloud.stream.rabbit.bindings.ecm_shop_template.consumer.autoBindDlq=true 18 spring.cloud.stream.rabbit.bindings.ecm_shop_template.consumer.republishToDlq=true

配置死信隊列會在消費者出現異常的時候重試3(默認為3，可以配置)次后將消息放入死信隊列中，效果如下：

 

生產者配置：

# BOSS消息生產者服務器配置
spring.cloud.stream.binders.boss.type=rabbit
#spring.cloud.stream.binders.boss.environment.spring.rabbitmq.addresses=${spring.rabbitmq.addresses}
#spring.cloud.stream.binders.boss.environment.spring.rabbitmq.username=${spring.rabbitmq.username}
 5 #spring.cloud.stream.binders.boss.environment.spring.rabbitmq.password=${spring.rabbitmq.password}
spring.cloud.stream.binders.boss.environment.spring.rabbitmq.virtual-host=boss

#BOSS基礎信息生產者
spring.cloud.stream.bindings.message_output.destination=exchange的名稱
#exchange的類型為廣播模式
spring.cloud.stream.rabbit.bindings.message_output.producer.exchangeType=fanout

下面是java代碼

1、定義消息的Input和Output配置信息

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

/**
 * mq連接源定義
 * 
 * 其中類中的2個屬性的值和properties里的配置需要一致
 **/
public interface MqMessageSource {
    // BOSS生產者
    String MESSAGE_OUTPUT = "message_output";
    // ECM消費者
    String ECM_SHOP_TEMPLATE_INPUT = "ecm_shop_template";

    @Output(MESSAGE_OUTPUT)
    MessageChannel messageOutput();
    
    @Input(ECM_SHOP_TEMPLATE_INPUT)
    MessageChannel messageInput();

}

2、消息消費

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;

/**
 * MQ消費者
 * @author yangzhilong
 *
 */
@Slf4j
@EnableBinding(MqMessageSource.class)
public class MqMessageConsumer {

    @Autowired
    private XXService xxService;
    
    /**
     * 消費ECM的貨櫃模板變更
     * @param message
     */
    @StreamListener(MqMessageSource.ECM_SHOP_TEMPLATE_INPUT)
    public void receive(Message<String> message) {
        log.info("接收貨櫃模板開始，參數={}", JSONObject.toJSONString(message));
        if (null == message) {
            return;
        }
        try {
            String payload = message.getPayload();
            log.info("具體消息內容= {}", JSONObject.toJSONString(payload));
            JSONObject jsonObject = JSONObject.parseObject(payload);
            ShopReqDto shopReqDto = new ShopReqDto();
            shopReqDto.setCode(jsonObject.getString("shopNo"));
            shopReqDto.setGoodsMarketTemplateId(jsonObject.getLong("goodsMarketTemplateId"));
            shopReqDto.setGoodsMarketTemplateName(jsonObject.getString("goodsMarketTemplateName"));
            ResponseResult<String> responseResult = xxService.updateTemplateIdAndName(shopReqDto);
            if(responseResult.isSuccess()){
                log.info("【MQ消費貨櫃模板更新信息成功】");
            }else{
                log.error("【MQ消費貨櫃模板更新信息失敗】,返回結果信息：" + JSONObject.toJSONString(responseResult));
            }
        } catch (Exception e) {
            log.error("接收處理貨櫃模板MQ時出現異常:{}", e);
            throw new RuntimeException(e);
        }
    }
}

3、消息生產者代碼

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;

/**
 * 消息生產者
 *
 **/
@EnableBinding(MqMessageSource.class)
@Slf4j
public class MqMessageProducer {
    @Autowired
    @Output(MqMessageSource.MESSAGE_OUTPUT)
    private MessageChannel channel;


    //品牌
    public void sendBrandAdd(Brand brand) {
        BossMessage<Brand> message = new BossMessage<>();
        message.setData(brand);
        message.setOpType(MqMessageProducer.ADD);
        message.setDataType(MqMessageProducer.BRAND);
        channel.send(MessageBuilder.withPayload(JSON.toJSONString(message)).build());
        log.info("【MQ發送內容】" + JSON.toJSONString(message));
    }
}