目录

前言

代码实现

1、引入依赖

2、异步消息与异步延迟消息生产者工具类-RocketMqUtil

3、MQ常量配置：消息主题、消息标签、消息分组

4、调用MQ

5、MQ监听类

Topic表示一类消息的集合，每个主题包含若干个消息，每个消息只属于一个主题，是RocketMQ进行消息订阅的基本单位。topic:message=1:n;  message:topic=1:1

一个生产者可以同时发送多种topic消息，而一个消费者只对某种特定的Topic感兴趣，即只可以订阅和消费一种Topic的消息。producer:topic=1:n;  consumer:topic=1:1

(一) RocketMQ异步消息_不如自成宇宙-CSDN博客

RocketMQ官网：Quick Start - Apache RocketMQ

下载安装：windows下RocketMQ下载安装教程 - darendu - 博客园

分布式事务消息：SpringCloud Alibaba微服务实战三十二 - 集成RocketMQ实现分布式事务_飘渺Jam的博客-CSDN博客

MQ选型：关于ActiveMQ、RocketMQ、RabbitMQ、Kafka一些总结和区别 - Alano的自嘲 - 博客园

​

代码实现

1、引入依赖

        
        
            org.apache.rocketmq
            rocketmq-spring-boot-starter
        

2、异步消息与异步延迟消息生产者工具类-RocketMqUtil

package com.xx.xx.mq.util;

import lombok.RequiredArgsConstructor;
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.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.io.Serializable;


@Component
@RequiredArgsConstructor
@Slf4j
public class RocketMqUtil {

    private final RocketMQTemplate rocketMQTemplate;


    
    public  void asyncSendDelay(String topic, String tag, T msg, int delayLevel) {
        Message message = MessageBuilder.withPayload(msg).build();
        log.info("发送的Topic={},tag={},内容={}", topic, tag, msg);
        rocketMQTemplate.asyncSend(topic + ":" + tag, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送延时消息成功的Topic={},tag={},内容={}", topic, tag, msg);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("发送延时消息失败的Topic={},tag={},内容={}", topic, tag, msg);
            }
        }, 3000L, delayLevel);
    }


    
    public  void asyncSend(String topic, String tag, T msg) {
        log.info("发送的Topic={},tag={},内容={}", topic, tag, msg);
        rocketMQTemplate.asyncSend(topic + ":" + tag, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送成功的Topic={},tag={},内容={}", topic, tag, msg);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("发送失败的Topic={},tag={},内容={}", topic, tag, msg);
            }
        });
    }


    
    public void convertAndSend(String topic, String tag, Object msg) {
        log.info("发送的Topic={},tag={},内容={}", topic, tag, msg);
        rocketMQTemplate.convertAndSend(topic + ":" + tag, msg);
    }


}

3、MQ常量配置：消息主题、消息标签、消息分组

package com.xx.xx.common.constant;


public final class MqConstants {

    
    public static final String TEST_TOPIC = "test_topic";

    
    public static final String TEST_TAG = "test_tag";
    
    public static final String TEST_GROUP = "test_group";


    
    public static final String ORDER_TOPIC = "order_topic";

    
    public static final String ORDER_CANCEL_TAG = "order_cancel_tag";
    
    public static final String ORDER_CANCEL_GROUP = "order_cancel_group";

}

4、调用MQ

        //发送延迟队列，1小时未支付自动取消
        rocketMqUtil.asyncSendDelay(MqConstants.ORDER_TOPIC, MqConstants.ORDER_CANCEL_TAG, order.getId(), 9);

5、MQ监听类

package com.xx.xx.listener;

import com.xx.xx.common.constant.MqConstants;
import com.xx.order.service.OrderService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;


@Component
@RocketMQMessageListener(topic = MqConstants.ORDER_TOPIC, selectorexpression = MqConstants.ORDER_CANCEL_TAG, consumerGroup = MqConstants.ORDER_CANCEL_GROUP)
@Slf4j
@RequiredArgsConstructor
public class OrderCancelMqListener implements RocketMQListener {

    private final OrderService orderService;

    @Override
    public void onMessage(Long id) {
        log.info("下单后延迟1小时未支付订单将自动取消id：{}", id);
        try {
            orderService.autoCancel(id);
        } catch (Exception e) {
            log.error("取消订单异常", e);
        }
    }

}