前言

在引入一項技術之前，首先必須清楚的是該技術可以為項目解決什麼問題。個人在瞭解消息隊列(Message Queue)之前，以為消息隊列主是用於發送短信、郵件等消息發送(異步解耦)，但深入理解才發現自己的理解錯了，MQ的作用不止體現在一些用戶接收到的具體消息裡，還可用於其它應用的數據發送、通用的業務處理等。消息隊列從字面上意思解讀就是將消息存放到隊列裡，根據隊列FIFO(先入先出)的特性進行消息消費。在實際開發中，是一種跨進程的通信機制，用於應用間的消息傳遞。

在引入MQ之前，需要了解的優缺點與應用場景

MQ的主要優點為解耦、異步、削峰，以下舉一個簡單的場景來反應這幾個特性。

在微服務項目中，一般會根據核心業務進行系統的垂直拆分再進行單獨部署。在上圖中，各系統在下單業務裡主要負責的內容如下：

• 訂單系統：創建訂單，將下單消息(如訂單id、用戶數據)發送到MQ
• MQ：限制每秒的訂單請求處理數(如每秒接收2000個請求但數據庫只能處理1000個則只處理1000個，處理不過來的先在消息隊列裡堆積)
• 物流系統：創建訂單物流信息
• 積分系統：用戶購物積分信息更新

想象下以上場景沒有MQ的的存在時創建訂單流程中存在的問題：

• 訂單系統創建完訂單信息後要去調用物流系統、積分系統上的業務接口，系統嚴重的耦合在一起(解耦)
• 訂單系統若非通過線程去調用其它系統的接口，還需同步等待返回浪費不少時間(異步，避免創建線程調用的麻煩)
• 用戶高峰期請求過多數據庫處理不過來進而導致應用崩潰(削峰)

任何事物都有兩面性，雖然MQ可以給系統解決不少問題，但也會引入一些問題，如：

• 系統複雜度提高，需要考慮消息重複消費、消息丟失等問題
• 數據一致性問題，如上例中的物流或庫存系統寫庫出現異常如何回滾補償

瞭解了MQ的一些特性後，再討論下幾個適合使用MQ的場景：

• 上游系統不關心下游的執行結果(如用戶註冊成功後用戶系統通過MQ向用戶發送郵件，但發送成不成功用戶系統根本不在意)
• 依賴於數據的定時任務(如下單後24小時內不支付則取消訂單，申請退款72小時內商家不處理則自動退款)

引入MQ後的一些問題解決思路

冪等性：對於同一操作的請求無論請求多少次結果都是一致的，在MQ中的具體體現為同一條消息無論發送都少次都只會被消費一次。

由於網絡抖動(延遲)的原因消息重複發送的問題是不可避免的，如果在消費端消費時沒有做好消息的冪等性保證就有可能出現重複消費，導致同一條消息被多次消費、寫庫多次的情況。比較常見的做法是為消息添加一個唯一標識(ID)，在消費時根據ID查詢數據庫是否存在該消息記錄，如果不存在再插入消息，存在則不進行插入消費。
當生成與消費時間間隔不長時，可使用Redis提高消息冪等性的效率，如：

1. 消費者消費前根據ID去查詢redis是否存在該消息

2. 不存在該消息則消費並寫入redis，存在該消息則不消費返回

關於消息ID：

1. RocketMQ的每條消息都會配有全局唯一的ID

2. 如果消息中間件不會生成ID，可考慮一些ID服務(如雪花算法)生成全局唯一ID

3. 建議ID不與實際業務關聯

如目前個人工作中負責的消息中心應用是基於MongoDB+RocketMQ的技術架構，MongoDB負責存儲各個應用發送過來的消息(主要為Sms、Email等)，每次消費前通過RocketMQ的Message ID查詢Mongo保證消息冪等性避免重複消費，消費成功後更新DB中的消息狀態。

MQ各組件的消息丟失含義都有所不同，導致與解決方案也不一定相同，以kafka、rocket的消息傳遞模型(Producer->Broker->Consumer)為例：

• Producer：消息未持久化到Broker中，或消費者未能成功消費到消息。Kafka可通過更改ack配置解決，rocketMQ中會返回消息發送狀態碼。
• Broker：消息成功傳到到我這裡了，可我因為某些原因(不同的MQ可能因機制問題有不同原因)弄丟了，如果是硬件原因(如宕機、磁盤損壞)建議你copy(集群部署)幾個我
• Consumer：我拿到了消息，但消費失敗了或中途掛掉了沒告訴Broker。可通過各MQ中間件的ACK機制解決。

基於RocketMQ的簡單例子技術框架與業務模型

以下便以一個基於MongoDB+RocketMQ+Eureka+Spring Cloud Config的技術框架並結合使用MQ中的問題搭建一個簡單的消息中心項目案例，其中各組件在項目中的主要作用如下：

• Spring Cloud Config：消息配置(如topic、ConsumerGroup、ProducerGroup)中心。
• Eureka：應用服務註冊中心，負責項目中各服務的發現與提供調用。
• MongoDB：由於消息的事務關係不強且Mongodb格式文檔自由(json存儲，隨意增刪字段)，所以使用Mongodb存儲各個應用發送過來的消息(主要為Sms、Email等)，每次消費前通過RocketMQ的Message ID查詢Mongo保證消息冪等性避免重複消費，消費成功後保存消息。
• RocketMQ：消息接收、存儲、發送。

下圖為該項目的應用關係模型：

消息中心應用：統一通用消息的業務處理應用，如短信發送、郵件發送、員工服務號推送等消息的處理問卷應用：負責員工調查問卷的分發，在該例子中只是一個簡單的消息發送測試應用common：存放各應用通用類，如短信消息類(SmsMessage)、消息常量類config-server-properties：配置中心的配置存放目錄由於該項目主要用於演示一些MQ的功能與使用中的問題解決方式，所以編碼部分比較簡單。

應用編碼

通用模塊編碼(common)

主要存放各應用通用類(如實體、常量、配置、功能等)。

MessageConstant：維護消息常量。

<code>public interface MessageConstant {

    interface System {
        String QUESTION = "QUESTION";
    }

    interface Topic {
        String SMS_TOPIC = "rocketmq.topic.sms";
        String SMS_TOPIC_TEMPLATE = "${rocketmq.topic.sms}";
        String MAIL_TOPIC = "rocketmq.topic.mail";
        String MAIL_TOPIC_TEMPLATE = "${rocketmq.topic.mail}";
    }

    interface Producer {
        String SMS_GROUP_TEMPLATE = "${rocketmq.producer.group.sms}";
        String MAIL_GROUP_TEMPLATE = "${rocketmq.producer.group.mail}"; 

    }

    interface Consumer {
        String SMS_GROUP_TEMPLATE = "${rocketmq.consumer.group.sms}";
        String MAIL_GROUP_TEMPLATE = "${rocketmq.consumer.group.mail}";
    }
}/<code>

BaseMessage：基礎消息類，所用的通用消息都需繼承此類方便統一信息的管理

<code>@Data
@Accessors(chain = true)
public abstract class BaseMessage implements Serializable {

    /**
     * 消息源系統:{@link io.wilson.common.message.constant.MessageConstant.System}
     */
    private String system;
}/<code>

SmsMessage：通用短信消息類，短信內容數據載體

<code>@EqualsAndHashCode(callSuper = true)
@Data
@Accessors(chain = true)
@ToString(callSuper = true)
public class SmsMessage extends BaseMessage {

    /**
     * 短信創建用戶
     */
    private String createUserId;
    /**
     * 接收短信用戶
     */
    private String toUserId;
    /**
     * 手機號碼
     */
    private String mobile; 

    /**
     * 短信內容
     */
    private String content;

}/<code>

消息中心應用(message-center)

消息中心在進行編碼之前，需確認消息中心該如何進行消息的處理。該項目所處的業務環境是各應用可能都需要發送一些短信消息、郵件、服務號消息等，相同消息的業務處理是一致的，所以消息中心對消息接收消費的主要流程如下：

• 保證消息冪等性(查詢數據庫使用已有消息記錄避免重複消費)
• 消息業務處理
• 消息日誌入庫

在該項目中，不同的消息類型存儲在不同的Mongodb collection(同Mysql table概念)，但共用一個消息日誌類MessageLog：

<code>@Data
@Accessors(chain = true)
public class MessageLog implements Serializable {
    private String msgId;
    /**
     * 發送方系統名稱 {@link io.wilson.common.message.constant.MessageConstant}
     */
    private String system;
    /** 

     * 消息對象json字符串
     */
    private String msgContent;
    /**
     * 業務執行結果
     */
    private Boolean success;
    private LocalDateTime createTime;
    private LocalDateTime updateTime;

    /**
     * 初始化消息記錄
     *
     * @param message       消息
     * @return
     */
    public static  MessageLog convertFromMessage(T message) {
        LocalDateTime now = LocalDateTime.now();
        return new MessageLog()
                .setSystem(message.getSystem())
                .setSuccess(false)
                .setCreateTime(now)
                .setUpdateTime(now);
    }
}/<code>

在該消費流程設計與開發編碼過程中個人考慮的核心點如下：

1. 如果使用普通消息類(如SmsMessage)作為db存儲的映射對象，會導致消息類摻雜不必要的屬性(如createTime、updateTime、success)，且作為一個通用的消息數據載體，普通消息類更適於作為一個VO而非DO使用，所以消息的處理結果、消息的創建更新時間這些作為原消息上的附加內容，更適合放到其它數據庫映射對象中維護，所以定義了MessageLog作為消息記錄的實體類
2. 既然是作為各應用都可使用的通用消息所以肯定都會有一定數據量，雖然映射實體都一樣，但存放到不同的collection可以提高操作的便捷性和獲得更好的性能，系統編碼可以更好地根據系統進行消息篩選
3. 在消息消費流程中，保證消息冪等性和消息日誌入庫這兩步只有數據庫名是不同的，所以可定義一個父Listener進行消息監聽消費的方法抽象，不同消息的業務處理交給不同的消息Service，同一類消息的消費可能會再細分調用不同的消息業務方法消費(如發送單條短信、批量發送短信)，所以可以對各service抽象出一個consume()方法根據參數調用具體的service業務方法進行消息消費

為了更好地展示消息中心類之間的關係，描繪以下類圖：

當一條短信消息發送到消息中心時，其消費流程如下：

• 消息業務處理編碼

BaseMessageService：消息業務消費抽象接口，抽象每個消費者(Listener)調用的業務消費方法

<code>public interface BaseMessageService {

    /**
     * 消費消息
     *
     * @param message         消息
     * @param consumeFunction 消費方法
     */
    default boolean consume(T message, Function consumeFunction) {
        return consumeFunction.apply(message);
    }
}/<code>

BaseMessageService：短信消息業務抽象接口

<code>@Service
public interface SmsMessageService extends BaseMessageService<smsmessage> {

    /**
     * 發送單條短信消息
     *
     * @param smsMessage
     * @return 業務處理結果
     */
    boolean sendSingle(SmsMessage smsMessage);

}/<smsmessage>/<code>

SmsMessageServiceImpl：短信消息業務實現類

<code>@Service
@Slf4j
public class SmsMessageServiceImpl implements SmsMessageService {

    @Override
    public boolean sendSingle(SmsMessage smsMessage) {
        // 短信業務操作結果 

        boolean isSuccess = true;
        /*
         * 短信業務操作並把操作結果設到isSuccess中
         */
        if (Objects.equals(smsMessage.getToUserId(), "Wilson")) {
            isSuccess = false;
            log.info("短信發送失敗,消息內容:{}", smsMessage);
        }
        return isSuccess;
    }
}/<code>

• 消息業務處理編碼

MessageLogConstant：維護MessageLog的相關常量(如不同消息的collection名)

<code>public interface MessageLogConstant {

    /**
     * 各消息日誌Mongo集合名
     */
    interface CollectionName {
        String SMS = "sms_message_log";
        String MAIL = "mail_message_log";
    }

}/<code>

AbstractMQStoreListener：保證消息冪等性、消息日誌入庫操作的抽象Listener類方法中

<code>

SmsMessageListener:短信消息監聽器(消費者)，如在消費過程中拋出異常，RocketMQ會以一定的時間間隔進行重新投遞消費

<code>@Slf4j
@Service
@ConditionalOnProperty(MessageConstant.Topic.SMS_TOPIC)
@RocketMQMessageListener(topic = MessageConstant.Topic.SMS_TOPIC_TEMPLATE, consumerGroup = MessageConstant.Consumer.SMS_GROUP_TEMPLATE)
public class SmsMessageListener extends AbstractMQStoreListener implements RocketMQListener<messageext> {
    @Resource
    private SmsMessageService smsMessageService;
    private static final String EXCEPTION_FORMAT = "短信消息消費失敗，消息內容：%s";

    @Override
    public void onMessage(MessageExt message) {
        String msgId = message.getMsgId();
        if (isConsumed(msgId)) {
            return;
        }
        SmsMessage smsMessage = JSONObject.parseObject(message.getBody(), SmsMessage.class);
        log.info("接收到短信消息{}：{}", msgId, smsMessage);
        /*if (Objects.equals(smsMessage.getToUserId(), "2020")) {
            log.error("消息{}消費失敗", message.getMsgId());
            // 拋出異常讓RocketMQ重新投遞消息重新消費
            throw new MQConsumeException(String.format(EXCEPTION_FORMAT, smsMessage));
        }*/
        boolean isSuccess = smsMessageService.consume(smsMessage, smsMessageService::sendSingle);
        if (!isSuccess) {
            log.info("短信消息業務操作失敗,消息id: {}", msgId);
        }
        // 保存消息消費記錄
        store(isSuccess, msgId, smsMessage);
    }

    @Override
    protected String collection() {
        return MessageLogConstant.CollectionName.SMS;
    }
}/<messageext>/<code>

MessageCenterApplication:主程序

<code>@SpringBootApplication
@EnableDiscoveryClient 

public class MessageCenterApplication {
    public static void main(String[] args) {
        SpringApplication.run(MessageCenterApplication.class, args);
    }
}/<code>

Spring Cloud配置文件bootstrap.yml

<code>eureka:
  client:
    service-url:
      defaultZone: http://localhost:8000/eureka
spring:
  cloud:
    config:
      discovery:
        enabled: true
        service-id: config-center
      #     資源文件名
      profile: dev
      name: rocketmq/<code>

SmsSendTest：單元測試類

<code>

• 配置中心(config-server)

主程序ConfigServerApplication

<code>@SpringBootApplication
@EnableDiscoveryClient
@EnableConfigServer
public class ConfigServerApplication {
   public static void main(String[] args) {
      SpringApplication.run(ConfigServerApplication.class, args);
   }
}/<code>

Spring Cloud配置文件bootstrap.yml:

<code>spring:
  cloud:
    config:
      server:
        git:
          uri: https://gitee.com/Wilson-He/rocketmq-message-center-demo.git
          username: Wilson-He
          force-pull: true
          password:
          # 配置文件在uri下的目錄
          search-paths: /config-server-properties
eureka:
  client:
    service-url:
      defaultZone: http://localhost:8000/eureka
/<code>

配置文件configs-server-properties/rocketmq-dev.properties:

<code>rocketmq.name-server=127.0.0.1:9876
rocketmq.topic.sms=sms-topic
rocketmq.producer.group.sms=sms-group
rocketmq.consumer.group.sms=sms-group
rocketmq.topic.mail=mail-topic
rocketmq.producer.group.mail=mail-group
rocketmq.consumer.group.mail=mail-group/<code>

運行流程

1. 運行RocketMQ name-server與broker,如mqnamesrv -n 127.0.0.1:9876,mqbroker -n 127.0.0.1:9876
2. 運行eureka應用
3. 運行配置中心config-server
4. 運行消息中心message-center
5. 運行message-center單元測試類(SmsSendTest)或運行question-app訪問localhost:8080/question/toUser?userId=xxx進行消費測試，消息中心控制檯打印出日誌信息與Mongo sms_message_log成功新增了數據即項目搭建完成

(待)擴展點：

1. RocketMQ的發送者應用可在配置文件中設置rocketmq.producer.retry-times-when-send-failed/retry-times-when-send-async-failed屬性配置rocketmq同步/異步發送消息失敗後的重試次數，不設置則默認都為2
2. 當業務執行操作結果失敗時仍然入庫的原因是有時業務執行過程中可能會包含調用第三方的操作，當第三方報錯時會導致業務操作結果失敗，而第三方的操作是不可控的，所以先把報錯結果保存便於追溯，且有業務需要時也可通過定時任務查庫重新執行業務
3. 該例子中只用了一個消息配置文件，實際開發中消息配置需根據項目所需配置到對應的項目配置文件，如question-app的消息配置(如topc、producerGroup)應在其項目中的配置文件(如application.yml、apollo的namespace)中配置
4. 該項目中的NameServer、Broker並沒有集群部署，Broker集群部署後配置同步雙寫避免主機寫入後尚未同步到從機就宕機導致消息丟失的情況(有意向的自行百度：RocketMQ 同步雙寫)

末

該文章通過一個簡單的項目例子演示了使用Spring Boot RocketMQ處理MQ常見問題的一些方式：

• 消息重複消費問題可通過數據庫存儲來保證冪等性
• 若消息消費業務操作失敗時可通過Listener拋出異常讓RocketMQ重新投遞消息進行消費

該項目例子可通過個人github/gitee(github|gitee.com/wilson-he/rocketmq-message-center-demo)查看

閱讀更多 Wilson1995 的文章

關鍵字: MongoDB 消費 可以很