1.背景

在第二章：《RocketMq詳解：二、SpringBoot集成RocketMq》中我們已經(jīng)實現(xiàn)了消費基本生產(chǎn)和消費的實現(xiàn)，但是在真實的開發(fā)環(huán)境中如果按照這種方式去實現(xiàn)，冗余代碼較多，且通過實現(xiàn)RocketMQListener中onMessage的方法去完成消息消費無返回結(jié)果，在后期的流程中不易維護，因此，本章將對這些問題進行二次改造和優(yōu)化。

為了防止新同學(xué)從頭開始學(xué)，本章將如何配置和實現(xiàn)簡單在復(fù)述一下，至于具體怎么安裝RocketMq，本文提供兩種安裝方法：

• 《MacOS環(huán)境下RocketMQ安裝及部署 RocketMQ Dashboard 可視化》
• 《docker安裝rocketMq》

2.通用方法改造

2.1添加maven依賴

		<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version></dependency>

2.2 RocketMq基礎(chǔ)配置

接著，在application.yml或application.properties文件中配置RocketMQ的相關(guān)參數(shù)，如NameServer地址、生產(chǎn)者組、消費者組等：

rocketmq:name-server: 127.0.0.1:9876# 生產(chǎn)者producer:group: boot_group_1# 消息發(fā)送超時時間send-message-timeout: 3000# 消息最大長度4Mmax-message-size: 4096# 消息發(fā)送失敗重試次數(shù)retry-times-when-send-failed: 3# 異步消息發(fā)送失敗重試次數(shù)retry-times-when-send-async-failed: 2# 消費者consumer:group: boot_group_1# 每次提取的最大消息數(shù)pull-batch-size: 5

上面的配置如果是在分布式環(huán)境下也可以配置在Apollo或nacos等配置中心里進行動態(tài)配置

2.3 配置類

在配置類中主要定義兩個Bean的加載，即RocketMQTemplate和DefaultMQProducer，主要是提供消息發(fā)送的能力，即生產(chǎn)消息；

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @author ninesun* @ClassName RocketMqConfig* @description: 消息中間件配置類* @date 2024年05月19日* @version: 1.0*/
@Configuration
public class RocketMqConfig {@Value("${rocketmq.name-server}")private String nameServer;@Value("${rocketmq.producer.group}")private String producerGroup;@Value("${rocketmq.producer.send-message-timeout}")private Integer sendMsgTimeout;@Value("${rocketmq.producer.max-message-size}")private Integer maxMessageSize;@Value("${rocketmq.producer.retry-times-when-send-failed}")private Integer retryTimesWhenSendFailed;@Value("${rocketmq.producer.retry-times-when-send-async-failed}")private Integer retryTimesWhenSendAsyncFailed;@Beanpublic RocketMQTemplate rocketMqTemplate() {RocketMQTemplate rocketMqTemplate = new RocketMQTemplate();rocketMqTemplate.setProducer(defaultMqProducer());return rocketMqTemplate;}@Beanpublic DefaultMQProducer defaultMqProducer() {DefaultMQProducer producer = new DefaultMQProducer();producer.setNamesrvAddr(this.nameServer);producer.setProducerGroup(this.producerGroup);producer.setSendMsgTimeout(this.sendMsgTimeout);producer.setMaxMessageSize(this.maxMessageSize);producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);producer.setRetryTimesWhenSendAsyncFailed(this.retryTimesWhenSendAsyncFailed);return producer;}
}

在編寫消費者和生產(chǎn)者之前，我們先統(tǒng)一一下消息傳輸?shù)膶ο?#xff0c;以及消費的結(jié)果

2.5 消息傳輸?shù)膶ο蠛徒Y(jié)果

• 基本傳輸對象

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class MessageDTO<T> {private T data;private Integer delayTime;@NotBlank(message = "Topic must not be blank")private String topic;@NotBlank(message = "Key must not be blank")private String key;private List<String> tags;private String messageType;
}

• 消費結(jié)果枚舉

public enum MsgRetryStatus {SUCCEED(-1), FAILURE(0), RETRY(0), RETRY_1S(1), RETRY_5S(2),RETRY_10S(3), RETRY_30S(4), RETRY_1M(5), RETRY_2M(6), RETRY_3M(7),RETRY_4M(8), RETRY_5M(9), RETRY_6M(10), RETRY_7M(11), RETRY_8M(12),RETRY_9M(13), RETRY_10M(14), RETRY_20M(15), RETRY_30M(16), RETRY_1H(17), RETRY_2H(18),RETRY_1D(19), RETRY_3D(20),  RETRY_7D(21), RETRY_14D(22), RETRY_21D(23),  RETRY_28D(24),RETRY_35D(25);int level;MsgRetryStatus(int level) {this.level = level;}public int getLevel() {return level;}
}

2.4 消息生產(chǎn)者

@Component
public class MessageProduct {@Resourceprivate RocketMQTemplate rocketMqTemplate;public SendResult SendMessage(MessageDTO data) {// 創(chuàng)建一個Message對象org.springframework.messaging.Message<?> message = MessageBuilder.withPayload(JSON.toJSONString(data)).setHeader(RocketMQHeaders.KEYS,data.getKey()).setHeader(RocketMQHeaders.TAGS,data.getTags()).build();return rocketMqTemplate.syncSendDelayTimeSeconds(data.getTopic(), message, data.getDelayTime());}
}

2.5 消息消費者

在原始的實現(xiàn)中，我們通過實現(xiàn)RocketMQListener接口中的onMessage方法來完成消息的消費。

然而，這種方式存在一個問題：如果消息消費失敗，我們無法獲取到返回結(jié)果，也不便于進行錯誤處理和重試。

為了優(yōu)化這個問題，我們可以使用@RocketMQMessageListener注解的returnTopic屬性來指定一個返回主題，當(dāng)消息消費失敗時，將消息發(fā)送到這個返回主題中。

同時，我們可以創(chuàng)建一個專門的處理失敗消息的消費者來處理這些返回的消息。

另外，我們還可以在onMessage方法中添加異常處理邏輯，以便在消費失敗時進行錯誤處理和記錄日志。

新增一個抽象類CommonConsumer去實現(xiàn)RocketMQListener，并提供一個有返回值的doConsumerProcess方法，去實現(xiàn)具體的消費邏輯，具體實現(xiàn)如下：

@Slf4j
@Component
public abstract class CommonConsumer implements RocketMQListener<MessageDTO> {public void onMessage(MessageDTO message) {log.info("收到延遲消息成功，消息體：{}", message);doConsumerProcess(message);}public abstract MsgRetryStatus doConsumerProcess(MessageDTO messageDTO);
}

以boot-mq-topic為例，我們實現(xiàn)具體的消費，新增一個對象BootMqConsumer繼承我們的CommonConsumer，來實現(xiàn)具體的消費邏輯

@Slf4j
@Component
public abstract class CommonConsumer implements RocketMQListener<MessageDTO> {public void onMessage(MessageDTO message) {try {// 處理消息的邏輯log.info("收到延遲消息成功，消息體：{}", message);MsgRetryStatus msgRetryStatus = doConsumerProcess(message);if (MsgRetryStatus.RETRY.equals(msgRetryStatus)|| MsgRetryStatus.FAILURE.equals(msgRetryStatus)) {//TODO 消費失敗或重試 則發(fā)送重試Topic}} catch (Exception e) {// 記錄錯誤日志e.printStackTrace();// 可以選擇將失敗消息發(fā)送到指定Topicthis.sendReturnMessgae();}}public abstract MsgRetryStatus doConsumerProcess(MessageDTO messageDTO);private void doRetrytConsumerProcess() {//TODO：待實現(xiàn)，后面補上}
}

2.6 功能測試

package com.example.demo.controller;import com.alibaba.fastjson.JSON;
import com.example.demo.annoation.Idempotent;
import com.example.demo.mq.producer.MessageProduct;
import com.example.demo.po.MessageDTO;
import com.example.demo.po.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.Collections;
import java.util.UUID;@RestController
@Slf4j
@Validated
public class TestController01 {@Resourceprivate RocketMQTemplate rocketMqTemplate;@Resourceprivate DefaultMQProducer defaultMqProducer;@Resourceprivate MessageProduct messageProduct;@GetMapping("/send/msg4")public String sendMsg4() {try {User user = User.builder().id(1).name("ninesun").build();MessageDTO<User> messageDTO = MessageDTO.<User>builder().data(user).delayTime(3).topic("boot-mq-topic").key(String.valueOf(UUID.randomUUID())).build();SendResult sendResult = messageProduct.SendMessage(messageDTO);log.info("msgId:{},sendStatus:{},data:{}", sendResult.getMsgId(), sendResult.getSendStatus(), JSON.toJSONString(messageDTO));} catch (Exception e) {e.printStackTrace();}return "OK";}
}

附：User對象

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class User {private Integer id;private String name;private String desc;
}

訪問：/send/msg4接口，結(jié)果如下：

至此，我們便可以在真實環(huán)境中很容易的去集成消息隊列實現(xiàn)功能的解耦，流量的削峰。