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

rocketmq:
    consumer:
        group: springboot_consumer_group
        # 一次拉取消息最大值，注意是拉取消息的最大值而非消費(fèi)最大值
        pull-batch-size: 10
    name-server: 10.5.103.6:9876
    producer:
        # 發(fā)送同一類(lèi)消息的設(shè)置為同一個(gè)group，保證唯一
        group: springboot_producer_group
        # 發(fā)送消息超時(shí)時(shí)間，默認(rèn)3000
        sendMessageTimeout: 10000
        # 發(fā)送消息失敗重試次數(shù)，默認(rèn)2
        retryTimesWhenSendFailed: 2
        # 異步消息重試此處，默認(rèn)2
        retryTimesWhenSendAsyncFailed: 2
        # 消息最大長(zhǎng)度，默認(rèn)1024 * 1024 * 4(默認(rèn)4M)
        maxMessageSize: 4096
        # 壓縮消息閾值，默認(rèn)4k(1024 * 4)
        compressMessageBodyThreshold: 4096
        # 是否在內(nèi)部發(fā)送失敗時(shí)重試另一個(gè)broker，默認(rèn)false
        retryNextServer: false

@RestController
public class NormalProduceController {
  @Setter(onMethod_ = @Autowired)
  private RocketMQTemplate rocketmqTemplate;
  
  @GetMapping("/test")
  public SendResult test() {
    Message<String> msg = MessageBuilder.withPayload("Hello,RocketMQ").build();
    SendResult sendResult = rocketmqTemplate.send(topic, msg);
  }
}

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "your_topic_name", consumerGroup = "your_consumer_group_name")
public class MyConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        // 處理消息的邏輯
        System.out.println("Received message: " + message);
    }
}

@SpringBootApplication
public class RocketDemoApplication {

    public static void main(String[] args) {
        /*
         * 指定使用的日志框架，否則將會(huì)告警
         * RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.InternalThreadLocalMap).
         * RocketMQLog:WARN Please initialize the logger system properly.
         */
        System.setProperty("rocketmq.client.logUseSlf4j", "true");
      
        SpringApplication.run(RocketDemoApplication.class, args);
    }
}

@GetMapping("/test")
public void test(){
  //普通消息無(wú)返回值，只負(fù)責(zé)發(fā)送消息?不等待服務(wù)器回應(yīng)且沒(méi)有回調(diào)函數(shù)觸發(fā)。
  RocketMessage rocketMessage = RocketMessage.builder().
    id(1111L).
    message("hello,world")
    .localDate(LocalDate.now())
    .localDateTime(LocalDateTime.now())
    .build();
  rocketmqTemplate.convertAndSend(destination,rocketMessage);
}

@Component
@RocketMQMessageListener(consumerGroup = "springboot_consumer_group",topic = "consumer_topic")
public class RocketMQConsumer implements RocketMQListener<RocketMessage> {
    @Override
    public void onMessage(RocketMessage message) {
        System.out.println("消費(fèi)消息-" + message);
    }
}

@Configuration
public class RocketMQEnhanceConfig {

    /**
     * 解決RocketMQ Jackson不支持Java時(shí)間類(lèi)型配置
     * 源碼參考：{@link org.apache.rocketmq.spring.autoconfigure.MessageConverterConfiguration}
     */
    @Bean
    @Primary
    public RocketMQMessageConverter enhanceRocketMQMessageConverter(){
        RocketMQMessageConverter converter = new RocketMQMessageConverter();
        CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
        List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
        for (MessageConverter messageConverter : messageConverterList) {
            if(messageConverter instanceof MappingJackson2MessageConverter){
                MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
                ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
                objectMapper.registerModules(new JavaTimeModule());
            }
        }
        return converter;
    }
}

rocketmq:
	enhance:
  	# 啟動(dòng)隔離，用于激活配置類(lèi)EnvironmentIsolationConfig
  	# 啟動(dòng)后會(huì)自動(dòng)在topic上拼接激活的配置文件，達(dá)到自動(dòng)隔離的效果
  	enabledIsolation: true
  	# 隔離環(huán)境名稱(chēng)，拼接到topic后，topic_dev，默認(rèn)空字符串
  	environment: dev

@Configuration
public class EnvironmentIsolationConfig implements BeanPostProcessor {
  	@Value("${rocketmq.enhance.enabledIsolation:true}")
    private boolean enabledIsolation;
    @Value("${rocketmq.enhance.environment:''}")
    private String environmentName;
  
    /**
     * 在裝載Bean之前實(shí)現(xiàn)參數(shù)修改
     */
    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if(bean instanceof DefaultRocketMQListenerContainer){

            DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;
					  //拼接Topic
            if(enabledIsolation && StringUtils.hasText(environmentName)){
                container.setTopic(String.join("_", container.getTopic(),environmentName));
            }
            return container;
        }
        return bean;
    }
}

/**
 * 消息實(shí)體，所有消息都需要繼承此類(lèi)
 * 公眾號(hào)：JAVA日知錄
 */
@Data
public abstract class BaseMessage {
    /**
     * 業(yè)務(wù)鍵，用于RocketMQ控制臺(tái)查看消費(fèi)情況
     */
    protected String key;
    /**
     * 發(fā)送消息來(lái)源，用于排查問(wèn)題
     */
    protected String source = "";

    /**
     * 發(fā)送時(shí)間
     */
    protected LocalDateTime sendTime = LocalDateTime.now();

    /**
     * 重試次數(shù)，用于判斷重試次數(shù)，超過(guò)重試次數(shù)發(fā)送異常警告
     */
    protected Integer retryTimes = 0;
}

@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class RocketMQEnhanceTemplate {
    private final RocketMQTemplate template;

    @Resource
    private RocketEnhanceProperties rocketEnhanceProperties;

    public RocketMQTemplate getTemplate() {
        return template;
    }

    /**
     * 根據(jù)系統(tǒng)上下文自動(dòng)構(gòu)建隔離后的topic
     * 構(gòu)建目的地
     */
    public String buildDestination(String topic, String tag) {
        topic = reBuildTopic(topic);
        return topic + ":" + tag;
    }

    /**
     * 根據(jù)環(huán)境重新隔離topic
     * @param topic 原始topic
     */
    private String reBuildTopic(String topic) {
        if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.hasText(rocketEnhanceProperties.getEnvironment())){
            return topic +"_" + rocketEnhanceProperties.getEnvironment();
        }
        return topic;
    }

    /**
     * 發(fā)送同步消息
     */
    public <T extends BaseMessage> SendResult send(String topic, String tag, T message) {
        // 注意分隔符
        return send(buildDestination(topic,tag), message);
    }


    public <T extends BaseMessage> SendResult send(String destination, T message) {
        // 設(shè)置業(yè)務(wù)鍵，此處根據(jù)公共的參數(shù)進(jìn)行處理
        // 更多的其它基礎(chǔ)業(yè)務(wù)處理...
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        SendResult sendResult = template.syncSend(destination, sendMessage);
        // 此處為了方便查看給日志轉(zhuǎn)了json，根據(jù)選擇選擇日志記錄方式，例如ELK采集
        log.info("[{}]同步消息[{}]發(fā)送結(jié)果[{}]", destination, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
        return sendResult;
    }

    /**
     * 發(fā)送延遲消息
     */
    public <T extends BaseMessage> SendResult send(String topic, String tag, T message, int delayLevel) {
        return send(buildDestination(topic,tag), message, delayLevel);
    }

    public <T extends BaseMessage> SendResult send(String destination, T message, int delayLevel) {
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
        log.info("[{}]延遲等級(jí)[{}]消息[{}]發(fā)送結(jié)果[{}]", destination, delayLevel, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
        return sendResult;
    }
}

@Slf4j
public abstract class EnhanceMessageHandler<T extends BaseMessage> {
    /**
     * 默認(rèn)重試次數(shù)
     */
    private static final int MAX_RETRY_TIMES = 3;

    /**
     * 延時(shí)等級(jí)
     */
    private static final int DELAY_LEVEL = EnhanceMessageConstant.FIVE_SECOND;


    @Resource
    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;

    /**
     * 消息處理
     *
     * @param message 待處理消息
     * @throws Exception 消費(fèi)異常
     */
    protected abstract void handleMessage(T message) throws Exception;

    /**
     * 超過(guò)重試次數(shù)消息，需要啟用isRetry
     *
     * @param message 待處理消息
     */
    protected abstract void handleMaxRetriesExceeded(T message);


    /**
     * 是否需要根據(jù)業(yè)務(wù)規(guī)則過(guò)濾消息，去重邏輯可以在此處處理
     * @param message 待處理消息
     * @return true: 本次消息被過(guò)濾，false：不過(guò)濾
     */
    protected boolean filter(T message) {
        return false;
    }

    /**
     * 是否異常時(shí)重復(fù)發(fā)送
     *
     * @return true: 消息重試，false：不重試
     */
    protected abstract boolean isRetry();

    /**
     * 消費(fèi)異常時(shí)是否拋出異常
     * 返回true，則由rocketmq機(jī)制自動(dòng)重試
     * false：消費(fèi)異常(如果沒(méi)有開(kāi)啟重試則消息會(huì)被自動(dòng)ack)
     */
    protected abstract boolean throwException();

    /**
     * 最大重試次數(shù)
     *
     * @return 最大重試次數(shù)，默認(rèn)5次
     */
    protected int getMaxRetryTimes() {
        return MAX_RETRY_TIMES;
    }

    /**
     * isRetry開(kāi)啟時(shí)，重新入隊(duì)延遲時(shí)間
     * @return -1：立即入隊(duì)重試
     */
    protected int getDelayLevel() {
        return DELAY_LEVEL;
    }

    /**
     * 使用模板模式構(gòu)建消息消費(fèi)框架，可自由擴(kuò)展或刪減
     */
    public void dispatchMessage(T message) {
        // 基礎(chǔ)日志記錄被父類(lèi)處理了
        log.info("消費(fèi)者收到消息[{}]", JSONObject.toJSON(message));

        if (filter(message)) {
            log.info("消息id{}不滿(mǎn)足消費(fèi)條件，已過(guò)濾。",message.getKey());
            return;
        }
        // 超過(guò)最大重試次數(shù)時(shí)調(diào)用子類(lèi)方法處理
        if (message.getRetryTimes() > getMaxRetryTimes()) {
            handleMaxRetriesExceeded(message);
            return;
        }
        try {
            long now = System.currentTimeMillis();
            handleMessage(message);
            long costTime = System.currentTimeMillis() - now;
            log.info("消息{}消費(fèi)成功，耗時(shí)[{}ms]", message.getKey(),costTime);
        } catch (Exception e) {
            log.error("消息{}消費(fèi)異常", message.getKey(),e);
            // 是捕獲異常還是拋出，由子類(lèi)決定
            if (throwException()) {
                //拋出異常，由DefaultMessageListenerConcurrently類(lèi)處理
                throw new RuntimeException(e);
            }
            //此時(shí)如果不開(kāi)啟重試機(jī)制，則默認(rèn)ACK了
            if (isRetry()) {
                handleRetry(message);
            }
        }
    }

    protected void handleRetry(T message) {
        // 獲取子類(lèi)RocketMQMessageListener注解拿到topic和tag
        RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
        if (annotation == null) {
            return;
        }
        //重新構(gòu)建消息體
        String messageSource = message.getSource();
        if(!messageSource.startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
            message.setSource(EnhanceMessageConstant.RETRY_PREFIX + messageSource);
        }
        message.setRetryTimes(message.getRetryTimes() + 1);

        SendResult sendResult;

        try {
            // 如果消息發(fā)送不成功，則再次重新發(fā)送，如果發(fā)送異常則拋出由MQ再次處理(異常時(shí)不走延遲消息)
            sendResult = rocketMQEnhanceTemplate.send(annotation.topic(), annotation.selectorExpression(), message, getDelayLevel());
        } catch (Exception ex) {
            // 此處捕獲之后，相當(dāng)于此條消息被消息完成然后重新發(fā)送新的消息
            //由生產(chǎn)者直接發(fā)送
            throw new RuntimeException(ex);
        }
        // 發(fā)送失敗的處理就是不進(jìn)行ACK，由RocketMQ重試
        if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
            throw new RuntimeException("重試消息發(fā)送失敗");
        }

    }
}

@Configuration
@EnableConfigurationProperties(RocketEnhanceProperties.class)
public class RocketMQEnhanceAutoConfiguration {

    /**
     * 注入增強(qiáng)的RocketMQEnhanceTemplate
     */
    @Bean
    public RocketMQEnhanceTemplate rocketMQEnhanceTemplate(RocketMQTemplate rocketMQTemplate){
        return new RocketMQEnhanceTemplate(rocketMQTemplate);
    }

    /**
     * 解決RocketMQ Jackson不支持Java時(shí)間類(lèi)型配置
     * 源碼參考：{@link org.apache.rocketmq.spring.autoconfigure.MessageConverterConfiguration}
     */
    @Bean
    @Primary
    public RocketMQMessageConverter enhanceRocketMQMessageConverter(){
        RocketMQMessageConverter converter = new RocketMQMessageConverter();
        CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
        List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
        for (MessageConverter messageConverter : messageConverterList) {
            if(messageConverter instanceof MappingJackson2MessageConverter){
                MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
                ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
                objectMapper.registerModules(new JavaTimeModule());
            }
        }
        return converter;
    }


    /**
     * 環(huán)境隔離配置
     */
    @Bean
    @ConditionalOnProperty(name="rocketmq.enhance.enabledIsolation", havingValue="true")
    public EnvironmentIsolationConfig environmentSetup(RocketEnhanceProperties rocketEnhanceProperties){
        return new EnvironmentIsolationConfig(rocketEnhanceProperties);
    }

}

public class EnvironmentIsolationConfig implements BeanPostProcessor {
    private RocketEnhanceProperties rocketEnhanceProperties;

    public EnvironmentIsolationConfig(RocketEnhanceProperties rocketEnhanceProperties) {
        this.rocketEnhanceProperties = rocketEnhanceProperties;
    }


    /**
     * 在裝載Bean之前實(shí)現(xiàn)參數(shù)修改
     */
    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if(bean instanceof DefaultRocketMQListenerContainer){

            DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;

            if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.hasText(rocketEnhanceProperties.getEnvironment())){
                container.setTopic(String.join("_", container.getTopic(),rocketEnhanceProperties.getEnvironment()));
            }
            return container;
        }
        return bean;
    }
}

@ConfigurationProperties(prefix = "rocketmq.enhance")
@Data
public class RocketEnhanceProperties {

    private boolean enabledIsolation;

    private String environment;
}

 <dependency>
   <groupId>com.jianzh5</groupId>
   <artifactId>cloud-rocket-starter</artifactId>
</dependency>

rocketmq:
	...
	enhance:
		# 啟動(dòng)隔離，用于激活配置類(lèi)EnvironmentIsolationConfig
  	# 啟動(dòng)后會(huì)自動(dòng)在topic上拼接激活的配置文件，達(dá)到自動(dòng)隔離的效果
  	enabledIsolation: true
    # 隔離環(huán)境名稱(chēng)，拼接到topic后，topic_dev，默認(rèn)空字符串
    environment: dev

@RestController
@RequestMapping("enhance")
@Slf4j
public class EnhanceProduceController {

    //注入增強(qiáng)后的模板，可以自動(dòng)實(shí)現(xiàn)環(huán)境隔離，日志記錄
    @Setter(onMethod_ = @Autowired)
    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;

    private static final String topic = "rocket_enhance";
    private static final String tag = "member";

    /**
     * 發(fā)送實(shí)體消息
     */
    @GetMapping("/member")
    public SendResult member() {
        String key = UUID.randomUUID().toString();
        MemberMessage message = new MemberMessage();
        // 設(shè)置業(yè)務(wù)key
        message.setKey(key);
        // 設(shè)置消息來(lái)源，便于查詢(xún)
        message.setSource("MEMBER");
        // 業(yè)務(wù)消息內(nèi)容
        message.setUserName("Java日知錄");
        message.setBirthday(LocalDate.now());

        return rocketMQEnhanceTemplate.send(topic, tag, message);
    }
}

@Slf4j
@Component
@RocketMQMessageListener(
        consumerGroup = "enhance_consumer_group",
        topic = "rocket_enhance",
        selectorExpression = "*",
        consumeThreadMax = 5 //默認(rèn)是64個(gè)線(xiàn)程并發(fā)消息，配置 consumeThreadMax 參數(shù)指定并發(fā)消費(fèi)線(xiàn)程數(shù)，避免太大導(dǎo)致資源不夠
)
public class EnhanceMemberMessageListener extends EnhanceMessageHandler<MemberMessage> implements RocketMQListener<MemberMessage> {

    @Override
    protected void handleMessage(MemberMessage message) throws Exception {
        // 此時(shí)這里才是最終的業(yè)務(wù)處理，代碼只需要處理資源類(lèi)關(guān)閉異常，其他的可以交給父類(lèi)重試
        System.out.println("業(yè)務(wù)消息處理:"+message.getUserName());
    }

    @Override
    protected void handleMaxRetriesExceeded(MemberMessage message) {
        // 當(dāng)超過(guò)指定重試次數(shù)消息時(shí)此處方法會(huì)被調(diào)用
        // 生產(chǎn)中可以進(jìn)行回退或其他業(yè)務(wù)操作
        log.error("消息消費(fèi)失敗，請(qǐng)執(zhí)行后續(xù)處理");
    }


    /**
     * 是否執(zhí)行重試機(jī)制
     */
    @Override
    protected boolean isRetry() {
        return true;
    }

    @Override
    protected boolean throwException() {
        // 是否拋出異常，false搭配retry自行處理異常
        return false;
    }
  
    @Override
    protected boolean filter() {
        // 消息過(guò)濾
        return false;
    }

    /**
     * 監(jiān)聽(tīng)消費(fèi)消息，不需要執(zhí)行業(yè)務(wù)處理，委派給父類(lèi)做基礎(chǔ)操作，父類(lèi)做完基礎(chǔ)操作后會(huì)調(diào)用子類(lèi)的實(shí)際處理類(lèi)型
     */
    @Override
    public void onMessage(MemberMessage memberMessage) {
        super.dispatchMessage(memberMessage);
    }
}