rocketMq立即消費與延時消費

pom依賴：

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.9.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot-starter</artifactId>
                <version>2.2.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-common</artifactId>
                <version>4.9.0</version>
            </dependency>

配置類：

/**
 * @author: datszhang
 * @Date: 2021/8/21 14:58
 * @Description:
 */
public class RocketMqConfig {

    /**
     * 支付成功訂單主題
     */
    public static final String PAYED_ORDER_TOPIC = "TRADE_PAYED_ORDER_TOPIC";
    public static final String PAYED_ORDER_GROUP = "TRADE_PAYED_ORDER_GROUP";
}

發送類：

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MqSender {

    @Autowired(required = false)
    private RocketMQTemplate rocketMqTemplate;

    private final static int DEFAULT_TIMEOUT = 3000;

    /**
     * 發送重試消息
     */
    public <T> void sendRetry(String topic, MqRetryMessage<T> mqRetryMessage){
        sendDelay(topic, mqRetryMessage, mqRetryMessage.getDelayLevel());
    }

    /**
     * 發送重試消息
     */
    public <T> void sendRetry(String topic, MqRetryMessage<T> mqRetryMessage, SendCallback sendCallback){
        sendDelay(topic, mqRetryMessage, sendCallback, mqRetryMessage.getDelayLevel());
    }

    /**
     * 延時消息
     */
    public <T> void sendDelay(String topic, T message, SendCallback sendCallback, MqDelayLevel delayLevel){
        rocketMqTemplate.asyncSend(topic, MqMessage.builder().body(message).build(), sendCallback, DEFAULT_TIMEOUT, delayLevel.getLevel());
    }

    /**
     * 實時消息
     */
    public <T> void send(String topic, T message, SendCallback sendCallback){
        rocketMqTemplate.asyncSend(topic, MqMessage.builder().body(message).build(), sendCallback, DEFAULT_TIMEOUT, 0);
    }


    /**
     * 延時消息
     */
    public <T> void sendDelay(String topic, T message, MqDelayLevel delayLevel){
        sendDelay(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("發送消息成功 {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("發送消息失敗", throwable);
            }
        }, delayLevel);
    }



    /**
     * 實時消息
     */
    public <T> void send(String topic, T message){
        send(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("發送消息成功 {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("發送消息失敗", throwable);
            }
        });
    }

    /**
     * 延時消息
     */
    public <T> void sendRocketDelay(String topic, T message,String from, MqDelayLevel delayLevel){
        sendDelay(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("{} 發送消息成功 {}",from, sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("{} 發送消息失敗",from, throwable);
            }
        }, delayLevel);
    }



    /**
     * 實時消息
     */
    public <T> void sendRocket(String topic, T message,String from){
        send(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("{} 發送消息成功 {}",from, sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("{} 發送消息失敗",from, throwable);
            }
        });
    }


}

延時級别：同原有的配置，隻是加了自己需要的時間

import lombok.AllArgsConstructor;
import lombok.Getter;

@AllArgsConstructor
@Getter
public enum MqDelayLevel {

    //1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d 1w 1m 1y
    ONE_SEC(1),
    FIVE_SEC(2),
    TEN_SEC(3),
    THIRTY_SEC(4),
    ONE_MIN(5),
    TWO_MIN(6),
    THREE_MIN(7),
    FOUR_MIN(8),
    FIVE_MIN(9),
    SIX_MIN(10),
    SEVEN_MIN(11),
    EIGHT_MIN(12),
    NINE_MIN(13),
    TEN_MIN(14),
    TWENTY_MIN(15),
    THIRTY_MIN(16),
    ONE_HOUR(17),
    TWO_HOUR(18),
    ONE_DAY(19),
    ONE_WEEK(20),
    ONE_MONTH(21),
    ONE_YEAR(22)
    ;

    private int level;
}

MQ消息封裝類：

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.NoArgsConstructor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;


@AllArgsConstructor
@NoArgsConstructor
@Builder
public class MqMessage <T> implements Message<T> {

    private T body;

    private MessageHeaders headers;

    @Override
    public T getPayload() {
        return body;
    }

    @Override
    public MessageHeaders getHeaders() {
        return headers;
    }
}

延時嘗試類：

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.weibu.cloud.common.utils.DateUtil;
import com.weibu.cloud.common.utils.uuid.UUID;
import lombok.Getter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.time.LocalDateTime;


@Getter
@ToString
@Slf4j
public class MqRetryMessage <T> implements Serializable {

    private static final long serialVersionUID = 408371795120089055L;
    /**
     * 預設延時級别
     */
    private final static MqDelayLevel[] DEFAULT_DELAY_LEVEL = {
            MqDelayLevel.ONE_MIN
            , MqDelayLevel.FIVE_MIN
            , MqDelayLevel.TEN_MIN
            , MqDelayLevel.THIRTY_MIN
            , MqDelayLevel.ONE_HOUR
            , MqDelayLevel.TWO_HOUR
    };

    /**
     * 重試ID
     */
    private final String retryId;

    /**
     * 目前重試次數
     */
    private final int retryTimes;
    /**
     * 重試開始時間
     */
    @JsonFormat(pattern = DateUtil.PATTERN)
    private final LocalDateTime startTime;
    /**
     * 最大重試次數
     */
    private final int maxAttempts;
    /**
     * 延時級别
     */
    private MqDelayLevel[] delayLevelList = DEFAULT_DELAY_LEVEL;
    /**
     * 消息體
     */
    private final T body;

    public MqRetryMessage(){
        this(null, 6);
    }

    /**
     * 構造函數
     */
    public MqRetryMessage(T body){
        this(body, 6);
    }

    public MqRetryMessage(T body, int maxAttempts){
        this(body, maxAttempts, 0, null, UUID.randomUUID().toString());
    }

    private MqRetryMessage(T body, int maxAttempts, int retryTimes, MqDelayLevel[] delayLevelList, String retryId){
        this.retryTimes = retryTimes;
        this.maxAttempts = maxAttempts;
        this.startTime = LocalDateTime.now();
        this.body = body;
        this.retryId = retryId;
        this.setDelayLevelList(delayLevelList);
    }

    /**
     * 是否可以重試
     */
    public boolean canRetry(){
        //maxAttempt 為0時無限制
        if(maxAttempts == 0){
            return true;
        }
        return retryTimes + 1 <= maxAttempts;
    }

    /**
     * 擷取下一個重試消息
     */
    public MqRetryMessage <T> nextRetry(){
        if(!canRetry()){
            throw new IllegalStateException("重試次數已到達最大無法重試");
        }

        return new MqRetryMessage<>(body, maxAttempts, retryTimes + 1, delayLevelList, retryId);
    }

    /**
     * 擷取目前延時級别
     */
    @JsonIgnore
    public MqDelayLevel getDelayLevel(){
        int index = retryTimes;
        if(retryTimes + 1 > delayLevelList.length){
            index = delayLevelList.length - 1;
        }

        return delayLevelList[index];
    }

    /**
     * 設定DelayLevel
     */
    public void setDelayLevelList(MqDelayLevel[] delayLevelList) {
        if(delayLevelList != null && delayLevelList.length > 0) {
            this.delayLevelList = delayLevelList;
        }
    }
}

MQ消息生産者：

import com.alibaba.fastjson.JSONObject;
import com.weibu.cloud.common.constant.OrderEnum;
import com.weibu.cloud.common.feign.user.UserApi;
import com.weibu.cloud.common.mq.MqSender;
import com.weibu.cloud.common.mq.RocketMqConfig;
import com.weibu.cloud.common.utils.DateUtil;
import com.weibu.cloud.trade.dto.OrderStatisticalItem;
import com.weibu.cloud.trade.dto.UserBrief;
import com.weibu.cloud.trade.entity.WbOrder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;


@Slf4j
@Component
public class PayedOrderStatisticalProducer {
    @Autowired
    private MqSender mqSender;
    @Autowired
    private UserApi userApi;

    
    public void send(OrderStatisticalItem item){
        mqSender.send(RocketMqConfig.ORDER_STATISTICAL_TOPIC, item, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("statistical send success.sendResult:{}", JSONObject.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable e) {
                log.error("statistical send failed . :{}",e);
            }
        });
    }

    @Async
    public void sendList(List<WbOrder> list){
        log.info("PayedOrderStatisticalProducer.sendList -> list:{}",list);
        if (CollectionUtils.isEmpty(list)){
            return;
        }
        list.forEach(wb -> {
            OrderStatisticalItem item = entityToItem(wb);
            this.send(item);
        });
    }

    public OrderStatisticalItem entityToItem(WbOrder entity){
        OrderStatisticalItem item = new OrderStatisticalItem();
        item.setOrderType(queryEnumType(entity.getOrderType()));
        item.setPayedAmount(entity.getAmountRealPay());
        item.setPayedDate(DateUtil.localDateTimeTransLocalDate(entity.getCreateTime()));
        item.setPayFlag(true);
        List<UserBrief> list = new ArrayList<>(2);
        list.add(UserBrief.build(entity.getAgentId(),2));
        list.add(UserBrief.build(entity.getMerchantId(),3));
        item.setUserIds(list);
        return item;
    }

    private OrderEnum.OrderType queryEnumType(Integer orderType){
        switch (orderType){
            case 0:
                return OrderEnum.OrderType.ALL;
            case 1:
                return OrderEnum.OrderType.AGENT;
            case 2:
                return OrderEnum.OrderType.BIZ;
            case 3:
                return OrderEnum.OrderType.MALL;
        }
        return OrderEnum.OrderType.BIZ;
    }
}

MQ消息消費者：

import com.alibaba.fastjson.JSONObject;
import com.weibu.cloud.common.mq.RocketMqConfig;
import com.weibu.cloud.trade.dto.OrderStatisticalItem;
import com.weibu.cloud.trade.service.impl.AsyncStatisticalService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;


@Slf4j
@Component
@RocketMQMessageListener(
        topic = RocketMqConfig.ORDER_STATISTICAL_TOPIC,
        consumerGroup = RocketMqConfig.ORDER_STATISTICAL_GROUP,
        consumeMode = ConsumeMode.ORDERLY, //并行處理，預設并行；ORDERLY 按順序處理
        selectorExpression = "*")
public class PayedOrderStatisticalConsumer implements RocketMQListener<OrderStatisticalItem> {
    @Autowired
    private AsyncStatisticalService asyncStatisticalService;

    @Override
    public void onMessage(OrderStatisticalItem item) {
        log.info("PayedOrderStatisticalConsumer -> item:{}", JSONObject.toJSONString(item));
        statisticalOrder(item);
    }

    //重試機制
    
    //@Retryable(value = Throwable.class, backoff = @Backoff(value = 600000L, delay = 300000L, multiplier = 5, maxDelay = 3600000L))
    @Retryable(value = Throwable.class, backoff = @Backoff(delay = 60000L, multiplier = 2), recover = "liquidationNotifyFailed")
    public void statisticalOrder(OrderStatisticalItem item){
        log.info("statisticalOrder -> item:{}",JSONObject.toJSONString(item));
        asyncStatisticalService.updateUserOrderStatistical(item.getUserIds(),item.getPayedAmount(),item.getPayedDate(),item.getOrderType(),item.getPayFlag());
    }
}

rocketMq立即消費與延時消費

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method