package com.codecoord.rocketmq.template;

import com.codecoord.rocketmq.constant.RocketMqBizConstant;
import com.codecoord.rocketmq.domain.RocketMqEntityMessage;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.validation.constraints.NotNull;
import java.time.LocalDate;
import java.time.LocalDateTime;

/**
 * 訂單類(lèi)發(fā)送消息模板工具類(lèi)
 *
 * @author tianxincode@163.com
 * @since 2022/6/16
 */
@Component
public class OrderMessageTemplate extends RocketMqTemplate {
    /// 如果不采用繼承也可以直接注入使用
    /* @Resource
    private RocketMqTemplate rocketMqTemplate; */

    /**
     * 入?yún)⒅恍枰獋魅胧悄膫€(gè)訂單號(hào)和業(yè)務(wù)體消息即可，其他操作根據(jù)需要處理
     * 這樣對(duì)于調(diào)用者而言，可以更加簡(jiǎn)化調(diào)用
     */
    public SendResult sendOrderPaid(@NotNull String orderId, String body) {
        RocketMqEntityMessage message = new RocketMqEntityMessage();
        message.setKey(orderId);
        message.setSource("訂單支付");
        message.setMessage(body);
        // 這兩個(gè)字段只是為了測(cè)試
        message.setBirthday(LocalDate.now());
        message.setTradeTime(LocalDateTime.now());
        return send(RocketMqBizConstant.SOURCE_TOPIC, RocketMqBizConstant.ORDER_PAID_TAG, message);
    }
}

package com.codecoord.rocketmq.listener;

import com.codecoord.rocketmq.constant.RocketMqSysConstant;
import com.codecoord.rocketmq.domain.BaseMqMessage;
import com.codecoord.rocketmq.template.RocketMqTemplate;
import com.codecoord.rocketmq.util.JsonUtil;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import javax.annotation.Resource;
import java.time.Instant;
import java.util.Objects;

/**
 * 抽象消息監(jiān)聽(tīng)器，封裝了所有公共處理業(yè)務(wù)，如
 * 1、基礎(chǔ)日志記錄
 * 2、異常處理
 * 3、消息重試
 * 4、警告通知
 * 5、....
 *
 * @author tianxincoord@163.com
 * @since 2022/4/17
 */
public abstract class BaseMqMessageListener<T extends BaseMqMessage> {
    /**
     * 這里的日志記錄器是哪個(gè)子類(lèi)的就會(huì)被哪個(gè)子類(lèi)的類(lèi)進(jìn)行初始化
     */
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Resource
    private RocketMqTemplate rocketMqTemplate;

    /**
     * 消息者名稱
     *
     * @return 消費(fèi)者名稱
     */
    protected abstract String consumerName();

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

    /**
     * 超過(guò)重試次數(shù)消息，需要啟用isRetry
     *
     * @param message 待處理消息
     */
    protected abstract void overMaxRetryTimesMessage(T message);
    /**
     * 是否過(guò)濾消息，例如某些
     *
     * @param message 待處理消息
     * @return true: 本次消息被過(guò)濾，false：不過(guò)濾
     */
    protected boolean isFilter(T message) {
        return false;
    }

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

    /**
     * 消費(fèi)異常時(shí)是否拋出異常
     *
     * @return true: 拋出異常，false：消費(fèi)異常(如果沒(méi)有開(kāi)啟重試則消息會(huì)被自動(dòng)ack)
     */
    protected abstract boolean isThrowException();

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

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

    /**
     * 由父類(lèi)來(lái)完成基礎(chǔ)的日志和調(diào)配，下面的只是提供一個(gè)思路
     */
    public void dispatchMessage(T message) {
        MDC.put(RocketMqSysConstant.TRACE_ID, message.getTraceId());
        // 基礎(chǔ)日志記錄被父類(lèi)處理了
        logger.info("[{}]消費(fèi)者收到消息[{}]", consumerName(), JsonUtil.toJson(message));
        if (isFilter(message)) {
            logger.info("消息不滿足消費(fèi)條件，已過(guò)濾");
            return;
        }
        // 超過(guò)最大重試次數(shù)時(shí)調(diào)用子類(lèi)方法處理
        if (message.getRetryTimes() > maxRetryTimes()) {
            overMaxRetryTimesMessage(message);
            return;
        }
        try {
            long start = Instant.now().toEpochMilli();
            handleMessage(message);
            long end = Instant.now().toEpochMilli();
            logger.info("消息消費(fèi)成功，耗時(shí)[{}ms]", (end - start));
        } catch (Exception e) {
            logger.error("消息消費(fèi)異常", e);
            // 是捕獲異常還是拋出，由子類(lèi)決定
            if (isThrowException()) {
                throw new RuntimeException(e);
            }
            if (isRetry()) {
                // 獲取子類(lèi)RocketMQMessageListener注解拿到topic和tag
                RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
                if (Objects.nonNull(annotation)) {
                    message.setSource(message.getSource() + "消息重試");
                    message.setRetryTimes(message.getRetryTimes() + 1);
                    SendResult sendResult;
                    try {
                        // 如果消息發(fā)送不成功，則再次重新發(fā)送，如果發(fā)送異常則拋出由MQ再次處理(異常時(shí)不走延遲消息)
                        // 此處捕獲之后，相當(dāng)于此條消息被消息完成然后重新發(fā)送新的消息
                        sendResult = rocketMqTemplate.send(annotation.topic(), annotation.selectorExpression(), message, retryDelayLevel());
                    } catch (Exception ex) {
                        throw new RuntimeException(ex);
                    }
                    // 發(fā)送失敗的處理就是不進(jìn)行ACK，由RocketMQ重試
                    if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                        throw new RuntimeException("重試消息發(fā)送失敗");
                    }
                }
            }
        }
    }
}

package com.codecoord.rocketmq.listener;

import com.codecoord.rocketmq.constant.RocketMqBizConstant;
import com.codecoord.rocketmq.domain.RocketMqEntityMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * 實(shí)體類(lèi)消費(fèi)監(jiān)聽(tīng)器，在實(shí)現(xiàn)RocketMQListener中間還加了一層BaseMqMessageListener來(lái)處理基礎(chǔ)業(yè)務(wù)消息
 *
 * @author tianxincoord@163.com
 * @since 2022/5/12
 */
@Slf4j
@Component
@RocketMQMessageListener(
        topic = RocketMqBizConstant.SOURCE_TOPIC,
        consumerGroup = RocketMqBizConstant.SOURCE_GROUP,
        selectorExpression = RocketMqBizConstant.SOURCE_TAG,
        // 指定消費(fèi)者線程數(shù)，默認(rèn)64，生產(chǎn)中請(qǐng)注意配置，避免過(guò)大或者過(guò)小
        consumeThreadMax = 5
)
public class RocketEntityMessageListener extends BaseMqMessageListener<RocketMqEntityMessage>
                                         implements RocketMQListener<RocketMqEntityMessage> {
    /**
     * 此處只是說(shuō)明封裝的思想，更多還是要根據(jù)業(yè)務(wù)操作決定
     * 內(nèi)功心法有了，無(wú)論什么招式都可以發(fā)揮最大威力
     */
    @Override
    protected String consumerName() {
        return "RocketMQ二次封裝消息消費(fèi)者";
    }

    @Override
    public void onMessage(RocketMqEntityMessage message) {
        // 注意，此時(shí)這里沒(méi)有直接處理業(yè)務(wù)，而是先委派給父類(lèi)做基礎(chǔ)操作，然后父類(lèi)做完基礎(chǔ)操作后會(huì)調(diào)用子類(lèi)的實(shí)際處理類(lèi)型
        super.dispatchMessage(message);
    }

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

    @Override
    protected void overMaxRetryTimesMessage(RocketMqEntityMessage message) {
        // 當(dāng)超過(guò)指定重試次數(shù)消息時(shí)此處方法會(huì)被調(diào)用
        // 生產(chǎn)中可以進(jìn)行回退或其他業(yè)務(wù)操作
    }

    @Override
    protected boolean isRetry() {
        return false;
    }

    @Override
    protected int maxRetryTimes() {
        // 指定需要的重試次數(shù)，超過(guò)重試次數(shù)overMaxRetryTimesMessage會(huì)被調(diào)用
        return 5;
    }

    @Override
    protected boolean isThrowException() {
        // 是否拋出異常，到消費(fèi)異常時(shí)是被父類(lèi)攔截處理還是直接拋出異常
        return false;
    }
}

package com.codecoord.rocketmq.listener;

import com.codecoord.rocketmq.constant.RocketMqBizConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * 廣播消息
 * 應(yīng)用場(chǎng)景：多租戶或者服務(wù)有內(nèi)部緩存需要刷新情況下如果需要刷新租戶信息或者緩存信息
 *      也就是需要所有服務(wù)節(jié)點(diǎn)都需要同事做某一件事情的時(shí)候
 * 此時(shí)可以借助廣播消息發(fā)送消息到所有節(jié)點(diǎn)刷新，無(wú)需一個(gè)節(jié)點(diǎn)一個(gè)節(jié)點(diǎn)的處理
 *
 * 特別說(shuō)明：廣播消息默認(rèn)會(huì)在家目錄下創(chuàng)建消費(fèi)進(jìn)度文件，會(huì)以www.tianxincoord.com:9876@www.tianxincoord.com:9876
 *      這種地址形式生成文件路徑，但是由于帶有:符號(hào)，windows下是不允許此符號(hào)作為文件夾名稱的
 *      所以如果rocketMQ的鏈接地址不是連接串(不帶有端口)可以取消下面的messageModel注釋
 *      否則啟動(dòng)的時(shí)候就會(huì)提示目標(biāo)卷或者路徑不存在，其實(shí)是因?yàn)檫@個(gè)問(wèn)題
 *
 * @author tianxincoord@163.com
 * @since 2022/5/12
 */
@Slf4j
@Component
@RocketMQMessageListener(
        topic = RocketMqBizConstant.SOURCE_TOPIC,
        consumerGroup = RocketMqBizConstant.SOURCE_BROADCASTING_GROUP,
        selectorExpression = RocketMqBizConstant.SOURCE_BROADCASTING_TAG
        // messageModel = MessageModel.BROADCASTING
)
public class RocketBroadcastingListener implements RocketMQListener<MessageExt> {

    /**
     * MessageExt：內(nèi)置的消息實(shí)體，生產(chǎn)中根據(jù)需要自己封裝實(shí)體
     */
    @Override
    public void onMessage(MessageExt message) {
        log.info("收到廣播消息【{}】", new String(message.getBody()));
    }
}

package com.codecoord.rocketmq.controller;

import com.alibaba.fastjson.JSONObject;
import com.codecoord.rocketmq.constant.RocketMqBizConstant;
import com.codecoord.rocketmq.domain.RocketMqEntityMessage;
import com.codecoord.rocketmq.template.OrderMessageTemplate;
import com.codecoord.rocketmq.template.RocketMqTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.UUID;

/**
 * 消息發(fā)送
 *
 * @author tianxin01@huice.com
 * @since 2022/6/16
 */
@RestController
@RequestMapping("/rocketmq")
@Slf4j
public class RocketMqController {
    /**
     * 注意此處注入的是封裝的RocketMqTemplate
     */
    @Resource
    private RocketMqTemplate rocketMqTemplate;
    /**
     * 注入對(duì)應(yīng)業(yè)務(wù)的模板類(lèi)
     */
    @Resource
    private OrderMessageTemplate orderMessageTemplate;

    /**
     * 通過(guò)實(shí)體類(lèi)發(fā)送消息，發(fā)送注意事項(xiàng)請(qǐng)參考實(shí)體類(lèi)
     * 說(shuō)明：也可以在RocketMqTemplate按照業(yè)務(wù)封裝發(fā)送方法，這樣只需要調(diào)用方法指定基礎(chǔ)業(yè)務(wù)消息接口
     */
    @RequestMapping("/entity/message")
    public Object sendMessage() {
        RocketMqEntityMessage message = new RocketMqEntityMessage();
        // 設(shè)置業(yè)務(wù)key
        message.setKey(UUID.randomUUID().toString());
        // 設(shè)置消息來(lái)源，便于查詢we年
        message.setSource("封裝測(cè)試");
        // 業(yè)務(wù)消息內(nèi)容
        message.setMessage("當(dāng)前消息發(fā)送時(shí)間為：" + LocalDateTime.now());
        // Java時(shí)間字段需要單獨(dú)處理，否則會(huì)序列化失敗
        message.setBirthday(LocalDate.now());
        message.setTradeTime(LocalDateTime.now());
        return rocketMqTemplate.send(RocketMqBizConstant.SOURCE_TOPIC, RocketMqBizConstant.SOURCE_TAG, message);
    }

    /**
     * 此時(shí)對(duì)于調(diào)用者而且，無(wú)需創(chuàng)建任何類(lèi)
     * 如果某天需要調(diào)整消息發(fā)送來(lái)源，如果不封裝，所有原來(lái)產(chǎn)生message的地方全部改
     * 如果封裝了，只需要改sendOrderPaid就可以切換
     */
    @RequestMapping("/order/paid")
    public Object sendOrderPaidMessage() {
        return orderMessageTemplate.sendOrderPaid(UUID.randomUUID().toString(), "客戶下單了...，快快備貨");
    }

    /**
     * 直接將對(duì)象進(jìn)行傳輸，也可以自己進(jìn)行json轉(zhuǎn)化后傳輸
     */
    @RequestMapping("/messageExt/message")
    public SendResult convertAndSend() {
        // 生產(chǎn)中不推薦使用jsonObject傳遞，不看發(fā)送者無(wú)法知道傳遞的消息包含什么信息
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("type", "messageExt");
        String destination = rocketMqTemplate.buildDestination(RocketMqBizConstant.SOURCE_TOPIC, RocketMqBizConstant.SOURCE_BROADCASTING_TAG);
        // 如果要走內(nèi)部方法發(fā)送則必須要按照標(biāo)準(zhǔn)來(lái)，否則就使用原生的消息發(fā)送
        return rocketMqTemplate.getTemplate().syncSend(destination, jsonObject);
    }
}