package com.codecoord.rocketmq.domain;

import lombok.Data;

import java.time.LocalDateTime;
import java.util.UUID;

/**
 * 基礎(chǔ)消息實(shí)體，包含基礎(chǔ)的消息
 * 根據(jù)自己的業(yè)務(wù)消息設(shè)置更多的字段
 *
 * @author tianxincoord@163.com
 * @since 2022/6/16
 */
@Data
public abstract class BaseMqMessage {
    /**
     * 業(yè)務(wù)鍵，用于RocketMQ控制臺查看消費(fèi)情況
     */
    protected String key;
    /**
     * 發(fā)送消息來源，用于排查問題
     */
    protected String source = "";
    /**
     * 發(fā)送時間
     */
    protected LocalDateTime sendTime = LocalDateTime.now();
    /**
     * 跟蹤id，用于slf4j等日志記錄跟蹤id，方便查詢業(yè)務(wù)鏈
     */
    protected String traceId = UUID.randomUUID().toString();
    /**
     * 重試次數(shù)，用于判斷重試次數(shù)，超過重試次數(shù)發(fā)送異常警告
     */
    protected Integer retryTimes = 0;
}

package com.codecoord.rocketmq.template;

import com.alibaba.fastjson.JSONObject;
import com.codecoord.rocketmq.constant.RocketMqSysConstant;
import com.codecoord.rocketmq.domain.BaseMqMessage;
import com.codecoord.rocketmq.util.JsonUtil;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * RocketMQ模板類
 *
 * @author tianxincoord@163.com
 * @since 2022/4/15
 */
@Component
public class RocketMqTemplate {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqTemplate.class);
    @Resource(name = "rocketMQTemplate")
    private RocketMQTemplate template;

    /**
     * 獲取模板，如果封裝的方法不夠提供原生的使用方式
     */
    public RocketMQTemplate getTemplate() {
        return template;
    }

    /**
     * 構(gòu)建目的地
     */
    public String buildDestination(String topic, String tag) {
        return topic + RocketMqSysConstant.DELIMITER + tag;
    }

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

    public <T extends BaseMqMessage> 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采集
        LOGGER.info("[{}]同步消息[{}]發(fā)送結(jié)果[{}]", destination, JsonUtil.toJson(message), JSONObject.toJSON(sendResult));
        return sendResult;
    }

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

    public <T extends BaseMqMessage> 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);
        LOGGER.info("[{}]延遲等級[{}]消息[{}]發(fā)送結(jié)果[{}]", destination, delayLevel, JsonUtil.toJson(message), JsonUtil.toJson(sendResult));
        return sendResult;
    }
}

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;

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

    /**
     * 入?yún)⒅恍枰獋魅胧悄膫€訂單號和業(yè)務(wù)體消息即可，其他操作根據(jù)需要處理
     * 這樣對于調(diào)用者而言，可以更加簡化調(diào)用
     */
    public SendResult sendOrderPaid(@NotNull String orderId, String body) {
        RocketMqEntityMessage message = new RocketMqEntityMessage();
        message.setKey(orderId);
        message.setSource("訂單支付");
        message.setMessage(body);
        // 這兩個字段只是為了測試
        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)聽器，封裝了所有公共處理業(yè)務(wù)，如
 * 1、基礎(chǔ)日志記錄
 * 2、異常處理
 * 3、消息重試
 * 4、警告通知
 * 5、....
 *
 * @author tianxincoord@163.com
 * @since 2022/4/17
 */
public abstract class BaseMqMessageListener<T extends BaseMqMessage> {
    /**
     * 這里的日志記錄器是哪個子類的就會被哪個子類的類進(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;

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

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

    /**
     * 消費(fèi)異常時是否拋出異常
     *
     * @return true: 拋出異常，false：消費(fèi)異常(如果沒有開啟重試則消息會被自動ack)
     */
    protected abstract boolean isThrowException();

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

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

    /**
     * 由父類來完成基礎(chǔ)的日志和調(diào)配，下面的只是提供一個思路
     */
    public void dispatchMessage(T message) {
        MDC.put(RocketMqSysConstant.TRACE_ID, message.getTraceId());
        // 基礎(chǔ)日志記錄被父類處理了
        logger.info("[{}]消費(fèi)者收到消息[{}]", consumerName(), JsonUtil.toJson(message));
        if (isFilter(message)) {
            logger.info("消息不滿足消費(fèi)條件，已過濾");
            return;
        }
        // 超過最大重試次數(shù)時調(diào)用子類方法處理
        if (message.getRetryTimes() > maxRetryTimes()) {
            overMaxRetryTimesMessage(message);
            return;
        }
        try {
            long start = Instant.now().toEpochMilli();
            handleMessage(message);
            long end = Instant.now().toEpochMilli();
            logger.info("消息消費(fèi)成功，耗時[{}ms]", (end - start));
        } catch (Exception e) {
            logger.error("消息消費(fèi)異常", e);
            // 是捕獲異常還是拋出，由子類決定
            if (isThrowException()) {
                throw new RuntimeException(e);
            }
            if (isRetry()) {
                // 獲取子類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再次處理(異常時不走延遲消息)
                        // 此處捕獲之后，相當(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í)體類消費(fèi)監(jiān)聽器，在實(shí)現(xiàn)RocketMQListener中間還加了一層BaseMqMessageListener來處理基礎(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)中請注意配置，避免過大或者過小
        consumeThreadMax = 5
)
public class RocketEntityMessageListener extends BaseMqMessageListener<RocketMqEntityMessage>
                                         implements RocketMQListener<RocketMqEntityMessage> {
    /**
     * 此處只是說明封裝的思想，更多還是要根據(jù)業(yè)務(wù)操作決定
     * 內(nèi)功心法有了，無論什么招式都可以發(fā)揮最大威力
     */
    @Override
    protected String consumerName() {
        return "RocketMQ二次封裝消息消費(fèi)者";
    }

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

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

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

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

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

    @Override
    protected boolean isThrowException() {
        // 是否拋出異常，到消費(fè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)用場景：多租戶或者服務(wù)有內(nèi)部緩存需要刷新情況下如果需要刷新租戶信息或者緩存信息
 *      也就是需要所有服務(wù)節(jié)點(diǎn)都需要同事做某一件事情的時候
 * 此時可以借助廣播消息發(fā)送消息到所有節(jié)點(diǎn)刷新，無需一個節(jié)點(diǎn)一個節(jié)點(diǎn)的處理
 *
 * 特別說明：廣播消息默認(rèn)會在家目錄下創(chuàng)建消費(fèi)進(jìn)度文件，會以www.tianxincoord.com:9876@www.tianxincoord.com:9876
 *      這種地址形式生成文件路徑，但是由于帶有:符號，windows下是不允許此符號作為文件夾名稱的
 *      所以如果rocketMQ的鏈接地址不是連接串(不帶有端口)可以取消下面的messageModel注釋
 *      否則啟動的時候就會提示目標(biāo)卷或者路徑不存在，其實(shí)是因?yà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;
    /**
     * 注入對應(yīng)業(yè)務(wù)的模板類
     */
    @Resource
    private OrderMessageTemplate orderMessageTemplate;

    /**
     * 通過實(shí)體類發(fā)送消息，發(fā)送注意事項(xiàng)請參考實(shí)體類
     * 說明：也可以在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è)置消息來源，便于查詢we年
        message.setSource("封裝測試");
        // 業(yè)務(wù)消息內(nèi)容
        message.setMessage("當(dāng)前消息發(fā)送時間為：" + LocalDateTime.now());
        // Java時間字段需要單獨(dú)處理，否則會序列化失敗
        message.setBirthday(LocalDate.now());
        message.setTradeTime(LocalDateTime.now());
        return rocketMqTemplate.send(RocketMqBizConstant.SOURCE_TOPIC, RocketMqBizConstant.SOURCE_TAG, message);
    }

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

    /**
     * 直接將對象進(jìn)行傳輸，也可以自己進(jìn)行json轉(zhuǎn)化后傳輸
     */
    @RequestMapping("/messageExt/message")
    public SendResult convertAndSend() {
        // 生產(chǎn)中不推薦使用jsonObject傳遞，不看發(fā)送者無法知道傳遞的消息包含什么信息
        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)來，否則就使用原生的消息發(fā)送
        return rocketMqTemplate.getTemplate().syncSend(destination, jsonObject);
    }
}