public class Producer {

    public static void main(String[] args) throws Exception {

        //1.创建生产者组
        DefaultMQProducer producer = new DefaultMQProducer("producer-hello");

        //2.设置NameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");

        //3.启动producer实例
        producer.start();

        //4.创建消息
        Message message = new Message("log-topic", "info-tag", "这是一个info信息".getBytes(RemotingHelper.DEFAULT_CHARSET));

        //5.发送消息
        SendResult result = producer.send(message);

        //6.关闭producer实例
        System.out.println("发送完毕，结果: "+result);
    }
}

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-hello");

        //2.设置NameServer地址
        consumer.setNamesrvAddr("127.0.0.1:9876");

        //3.订阅topic，指定tag标签
        consumer.subscribe("log-topic","info-tag");

        //4.注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently(){

            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s 接收到新的消息:  %n", Thread.currentThread().getName());
                msgs.stream().forEach(messageExt -> {
                    String body = null;
                    try {
                        body = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                    System.out.println(body);
                });
			   //失败消费，稍后尝试消费,会进行多次重试
                //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //5.启动消费者
        consumer.start();

        System.out.println("消费者启动...");

    }
}

public class Producer {

    //演示消息同步发送
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //生产者
        DefaultMQProducer producer = new DefaultMQProducer("syn-producerGroup");
        //设置name server地址
        producer.setNamesrvAddr("127.0.0.1:9876");
//设置队列数量为2，默认为4，根据情况设置
        producer.setDefaultTopicQueueNums(2); 
        //启动
        producer.start();

        for (int i = 0 ; i < 16 ; i++){

            Message message = new Message();
            //消息主题
            message.setTopic("syn-topic");
            //消息标签
            message.setTags("sms");
            //添加内容
            message.setBody((i+"我是消息").getBytes(CharsetUtil.UTF_8));
            //执行发送
            SendResult result = producer.send(message);
            //打印结果
            System.out.println(result);
        }

        producer.shutdown();
    } }	

. . . 省略. . .
producer.send(
        //创建消息对象
        new Message("asyn-topic", "sms", "我是消息".getBytes(CharsetUtil.UTF_8)),
        //添加发送回调
        new SendCallback() {

            //发送成功结果处理
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }
            //发送异常结果处理
            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送异常："+throwable.getMessage());
            }
        }
);

... 省略...
Message message = new Message("asyn-topic", "sms", "我是消息".getBytes(CharsetUtil.UTF_8));
producer.sendOneway(message);

producer.setDefaultTopicQueueNums(1);

//最大线程1个
defaultMQPushConsumer.setConsumeThreadMax(1);
defaultMQPushConsumer.setConsumeThreadMin(1);
//同时只拉取一个消息
defaultMQPushConsumer.setPullBatchSize(1);

...省略...
defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                list.forEach(message->{
                    System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));
                });

                return ConsumeOrderlyStatus.SUCCESS;
            }


        });
...省略...

//演示消息同步发送
public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
    //生产者
    DefaultMQProducer producer = new DefaultMQProducer("syn-producerGroup");

    //设置name server地址
    producer.setNamesrvAddr("127.0.0.1:9876");
   
    //发送消息超时时间
    producer.setSendMsgTimeout(1000);
    //启动
    producer.start();

    for (long i = 0 ; i < 4 ; i++){


        Order order = new Order(i,"订单"+i,"创建");

        //添加内容
        byte[] bytes = (JSON.toJSONString(order)).getBytes(CharsetUtil.UTF_8);

        Message message = new Message("topic-order","product-order",bytes);
        message.setKeys("key-"+i);

        //执行发送
        SendResult result = producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long id = (Long) arg;
                //使用取模算法确定id存放到哪个队列
                int index =(int) (id % mqs.size());
                //index就是要存放的队列的索引
                return mqs.get(index);
            }
//把订单ID作为参数，作为选择器的基础数据
        },order.getId());

        System.out.println(result);

        //====================================================================
        order.setStatus("支付");
        //添加内容
        bytes = (JSON.toJSONString(order)).getBytes(CharsetUtil.UTF_8);
        message = new Message("topic-order","product-order",bytes);
        message.setKeys("key-"+i);

        //执行发送
        result = producer.send(message,new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long id = (Long) arg;
                //使用取模算法确定id存放到哪个队列
                int index =(int) (id % mqs.size());
                //index就是要存放的队列的索引
                return mqs.get(index);
            }
        },order.getId());

        System.out.println(result);

        //====================================================================
        order.setStatus("发货");
        //添加内容
        bytes = (JSON.toJSONString(order)).getBytes(CharsetUtil.UTF_8);
        message = new Message("topic-order","product-order",bytes);
        message.setKeys("key-"+i);

        //执行发送
        result = producer.send(message,new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long id = (Long) arg;
                //使用取模算法确定id存放到哪个队列
                int index =(int) (id % mqs.size());
                //index就是要存放的队列的索引
                return mqs.get(index);
            }
        },order.getId());

        System.out.println(result);

        //打印结果
    }

    producer.shutdown();
}

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("syn-consumerGroup");
        //设置name server 地址
        defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        //从开始位置消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅
        defaultMQPushConsumer.subscribe("topic-order","product-order");

        defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                list.forEach(message->{
                    System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));
                });

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        defaultMQPushConsumer.start();
    }
}

public class Producer {

    //演示消息同步发送
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //生产者
        DefaultMQProducer producer = new DefaultMQProducer("syn-producerGroup-delay");

        //设置name server地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //启动
        producer.start();

        for (long i = 0 ; i < 4 ; i++){


            Order order = new Order(i,"订单"+i,"创建");

            //添加内容
            byte[] bytes = (JSON.toJSONString(order)).getBytes(CharsetUtil.UTF_8);

            Message message = new Message("topic-order-delay","product-order-delay",bytes);
		  //延迟级别 3，代表 10s延迟
            message.setDelayTimeLevel(3);
            message.setKeys("key-"+i);

          
            //执行发送
            SendResult result = producer.send(message);
            System.out.println("发送时间:"+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));

            System.out.println(result);
        }

        producer.shutdown();
    }
}

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("syn-consumerGroup-delay");
        //设置name server 地址
        defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        //从开始位置消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //订阅
        defaultMQPushConsumer.subscribe("topic-order-delay","product-order-delay");

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                list.forEach(message->{
                    System.out.println("消费时间:"+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                    System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));
                });

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        defaultMQPushConsumer.start();
    }
}

public class MyTransactionCheckListener implements TransactionListener {

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        //执行业务，保存本地事务

        //保存成功
        return LocalTransactionState.COMMIT_MESSAGE ; //ROLLBACK_MESSAGE;    //未知
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        //这里查询本地事务状态
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

public class TransationSender {

    public static void main(String[] args) throws MQClientException {
		//使用事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("tran-product-group");

        producer.setNamesrvAddr("127.0.0.1:9876");

        //线程池底层是使用新开线程去发布消息到MQ
        ExecutorService excutorService = Executors.newFixedThreadPool(20);

        producer.setExecutorService(excutorService);
		//指定事务监听器
        producer.setTransactionListener(new MyTransactionCheckListener());

        //设置事务消息监听
        producer.start();

        for(int i = 0 ; i < 10 ; i++){
            String orderId = UUID.randomUUID().toString();
            String tags = "Tag";
            Message message = new Message("topic-tran", "tag", orderId, ("下单:"+i).getBytes(CharsetUtil.UTF_8));
			//发送事务消息
            TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
            System.out.println(transactionSendResult);

        }
        producer.shutdown();
    }
}

public class TransationConsumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("trans-consumer-group");
        //设置name server 地址
        defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");

        //订阅
        defaultMQPushConsumer.subscribe("topic-tran", "tag");

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                list.forEach(message->{
                    System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));
                });

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        defaultMQPushConsumer.start();
    }
}

//消息切割器，按照4M大小写个
public class ListSplitter implements Iterator<List<Message>> {

    private final int SIZE_LIMIT = 1024 * 1024 * 4;

    private final List<Message> messages;

    private int currIndex;

    public ListSplitter(List<Message> messages) { 
        this.messages = messages;
    }

    @Override public boolean hasNext() {
        return currIndex < messages.size(); 
    }

    @Override public List<Message> next() { 
        int startIndex = getStartIndex();
        int nextIndex = startIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex); 
            int tmpSize = calcMessageSize(message);
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break; 
            } else {
                totalSize += tmpSize; 
            }
        }
        List<Message> subList = messages.subList(startIndex, nextIndex); 
        currIndex = nextIndex;
        return subList;
    }
    private int getStartIndex() {
        Message currMessage = messages.get(currIndex); 
        int tmpSize = calcMessageSize(currMessage); 
        while(tmpSize > SIZE_LIMIT) {
            currIndex += 1;
            Message message = messages.get(currIndex);
            tmpSize = calcMessageSize(message);
        }
        return currIndex; 
    }
    private int calcMessageSize(Message message) {
        int tmpSize = message.getTopic().length() + message.getBody().length;
        Map<String, String> properties = message.getProperties();
        for (Map.Entry<String, String> entry : properties.entrySet()) {
            tmpSize += entry.getKey().length() + entry.getValue().length(); 
        }
        tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
        return tmpSize; 
    }
}

public class BatchProducer {

    //演示消息同步发送
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //生产者
        DefaultMQProducer producer = new DefaultMQProducer("batch-producerGroup");

        //设置name server地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //设置最大消息大小，默认4M
        producer.setMaxMessageSize(1024 * 1024 * 4);
        //启动
        producer.start();


        //===========准备消息==========================================================
        List<Message> messages = new ArrayList<>();

        for (long i = 0 ; i < 10000 ; i++){
            //添加内容
            byte[] bytes = ("批量消息".getBytes(CharsetUtil.UTF_8));
            Message message = new Message("topic-order-batch","product-order-batch",bytes);
            message.setKeys("key-"+i);
            messages.add(message);
        }
        //===========切割消息==========================================================

        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(messages);

        while (splitter.hasNext()) {
            try {
                List<Message>  listItem = splitter.next();
                //发送消息
                SendResult sendResult = producer.send(listItem);
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                //处理error
            }
        }

        producer.shutdown();
    }
}

public class BatchConsumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("batch-consumerGroup");
        //设置name server 地址
        defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        //从开始位置消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //批量拉取消息数量，默认32
        defaultMQPushConsumer.setPullBatchSize(32);
        //每次消费条数，默认1
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(10);

        //订阅
        defaultMQPushConsumer.subscribe("topic-order-batch","product-order-batch");

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                list.forEach(message->{
                    System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));
                });

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        defaultMQPushConsumer.start();
    }
}

public void subscribe(finalString topic, final MessageSelector messageSelector)

...省略...
            //添加内容
            byte[] bytes = (JSON.toJSONString("消息")).getBytes(CharsetUtil.UTF_8);

            Message message = new Message("topic-order-filter","product-order-filter",bytes);
            //添加一个用户属性，用来作为过滤条件
            message.putUserProperty("sex",(i % 2)+"");
            message.setKeys("key-"+i);
            //执行发送
            SendResult result = producer.send(message);
...省略...

...省略...
//订阅
//defaultMQPushConsumer.subscribe("topic-order-filter","product-order-filter || tagb");

//只是消费 sex为 0的消息
defaultMQPushConsumer.subscribe("topic-order-filter", MessageSelector.bySql("sex = 0"));

defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        list.forEach(message->{
            System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));
        });

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
...省略...

// 设置重试发送的次数，默认为2次 
producer.setRetryTimesWhenSendFailed(3); 
// 设置发送超时时限为5s，默认3s 
producer.setSendMsgTimeout(5000); 

//指定异步发送失败后不进行重试发送 
producer.setRetryTimesWhenSendAsyncFailed(0)

//同步发送消息，如果5秒内没有发送成功，则重试5次
DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer");
producer.setRetryTimesWhenSendFailed(5);
producer.send(msg,5000L);

//每隔100毫秒重试
consumer.setSuspendCurrentQueueTimeMillis(100); 

// 修改消费重试次数 
consumer.setMaxReconsumeTimes(10); 

//最大重试次数，默认16
defaultMQPushConsumer.setMaxReconsumeTimes(10);

defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for(MessageExt message : list){

            System.out.println(message+" ; "+new String(message.getBody(), CharsetUtil.UTF_8));

            if(message.getReconsumeTimes() > 3){
                //如果重试次数大于3，可以把消息持久化到数据库，然后另外使用一个定时任务去定时重试。
                //甚至进行人工干预

                //消息持久化到数据库，这里就没必要重试了，返回success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }

            //出现异常,进行重试
            try {
                throw new Exception("出现异常了...");
            } catch (Exception e) {
                e.printStackTrace();
                //稍后重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }

        };

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RocketMQApplication {
    public static void main(String[] args) {
        SpringApplication.run(RocketMQApplication.class, args);
    }
}

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.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class BaseRocketMQProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    /**
     * 发送普通同步消息
     */
    public SendResult sendMsg(String topic, String msgBody) {
        return rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 发送异步消息
     */
    public void sendAsyncMsg(String topic, String msgBody) {
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功：" + sendResult);
            }
            @Override
            public void onException(Throwable e) {
                System.out.println("异步消息发送失败：" + e.getMessage());
            }
        });
    }

    /**
     * 发送基础延时消息
     * 延时等级：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public SendResult sendDelayMsg(String topic, String msgBody, Integer delayLevel) {
        return rocketMQTemplate.syncSend(topic,
                MessageBuilder.withPayload(msgBody).build(),
                messageTimeOut,
                delayLevel);
    }

    /**
     * 发送带Tag的消息
     */
    public SendResult sendTagMsg(String topic, String tag, String msgBody) {
        String destination = topic + ":" + tag;
        return rocketMQTemplate.syncSend(destination, MessageBuilder.withPayload(msgBody).build());
    }
}

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;
import cn.hutool.core.util.CharsetUtil;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "base_topic",
        selectorExpression = "*",
        consumerGroup = "base_consumer_group"
)
public class BaseConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), CharsetUtil.UTF_8);
        log.info("基础消费者接收消息：{}", msg);
    }
}

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class OrderlyProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送顺序消息
     * @param topic 主题
     * @param msgBody 消息内容
     * @param orderId 顺序ID（相同ID的消息进入同一个队列，保证顺序）
     */
    public SendResult sendOrderlyMsg(String topic, String msgBody, Long orderId) {
        // syncSendOrderly ：发送顺序消息
        // 第三个参数：hashKey（根据hashKey路由到固定队列）
        return rocketMQTemplate.syncSendOrderly(
                topic,
                MessageBuilder.withPayload(msgBody).build(),
                orderId.toString()
        );
    }
}

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import cn.hutool.core.util.CharsetUtil;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "orderly_topic",
        consumerGroup = "orderly_consumer_group",
        consumeMode = ConsumeMode.ORDERLY, // 核心：开启顺序消费
        selectorExpression = "*"
)
public class OrderlyConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), CharsetUtil.UTF_8);
        log.info("顺序消费者接收消息：{}，队列ID：{}", msg, message.getQueueId());
    }
}

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class DelayProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送延时消息
     * 延时等级：1=1s,2=5s,3=10s,4=30s,5=1m,6=2m,7=3m,8=4m,9=5m,10=6m,11=7m,12=8m,13=9m,14=10m,15=20m,16=30m,17=1h,18=2h
     */
    public SendResult sendDelayMsg(String msg) {
        // 参数：topic, 消息体, 超时时间, 延时等级
        return rocketMQTemplate.syncSend(
                "delay_topic",
                MessageBuilder.withPayload(msg).build(),
                3000,
                5  // 延时1分钟
        );
    }
}

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;
import cn.hutool.core.util.CharsetUtil;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "delay_topic",
        consumerGroup = "delay_consumer_group"
)
public class DelayConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), CharsetUtil.UTF_8);
        log.info("延时消费者接收消息：{}", msg);
    }
}

import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class TransactionProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送事务消息
     */
    public TransactionSendResult sendTransactionMsg(String msg) {
        return rocketMQTemplate.sendMessageInTransaction(
                "tx_producer_group",  // 事务生产者组
                "tx_topic",
                MessageBuilder.withPayload(msg).build(),
                null  // 本地事务参数
        );
    }
}

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RocketMQTransactionListener(producerGroup = "tx_producer_group")
public class TransactionListener implements RocketMQLocalTransactionListener {

    /**
     * 执行本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 1. 执行本地业务（数据库操作）
            log.info("执行本地事务成功");
            // 2. 提交事务，消息可被消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("本地事务执行失败", e);
            // 回滚消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 事务回查（未收到事务状态时触发）
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 查询本地事务状态
        log.info("执行事务回查，状态：提交");
        return RocketMQLocalTransactionState.COMMIT;
    }
}

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;
import cn.hutool.core.util.CharsetUtil;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "tx_topic",
        consumerGroup = "tx_consumer_group"
)
public class TransactionConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), CharsetUtil.UTF_8);
        log.info("事务消费者接收消息：{}", msg);
    }
}

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;
import cn.hutool.core.util.CharsetUtil;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "normal_topic",
        consumerGroup = "normal_consumer_group",
        maxReconsumeTimes = 3  // 最大重试次数，超过进入死信
)
public class NormalConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), CharsetUtil.UTF_8);
        log.info("开始消费消息：{}", msg);
        // 模拟消费失败，触发重试→死信
        throw new RuntimeException("消费异常，消息进入死信队列");
    }
}

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;
import cn.hutool.core.util.CharsetUtil;

@Slf4j
@Component
@RocketMQMessageListener(
        // 死信队列格式：%DLQ%消费者组名
        topic = "%DLQ%normal_consumer_group",
        consumerGroup = "dlq_consumer_group"
)
public class DLQConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), CharsetUtil.UTF_8);
        log.error("=====死信队列接收消息=====");
        log.error("死信消息内容：{}", msg);
        log.error("死信消息ID：{}", message.getMsgId());
        // 人工处理：重发、修复数据、记录日志等
    }
}

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class RocketMQTestController {

    @Autowired
    private BaseRocketMQProducer baseProducer;

    @Autowired
    private OrderlyProducer orderlyProducer;

    @Autowired
    private DelayProducer delayProducer;

    @Autowired
    private TransactionProducer transactionProducer;

    // 测试普通消息
    @GetMapping("/send/base")
    public SendResult baseMsg(String msg) {
        return baseProducer.sendMsg("base_topic", msg);
    }

    // 测试顺序消息
    @GetMapping("/send/orderly")
    public SendResult orderlyMsg(String msg, Long orderId) {
        return orderlyProducer.sendOrderlyMsg("orderly_topic", msg, orderId);
    }

    // 测试延时消息
    @GetMapping("/send/delay")
    public SendResult delayMsg(String msg) {
        return delayProducer.sendDelayMsg(msg);
    }

    // 测试事务消息
    @GetMapping("/send/transaction")
    public TransactionSendResult transactionMsg(String msg) {
        return transactionProducer.sendTransactionMsg(msg);
    }

    // 测试死信消息
    @GetMapping("/send/dlq")
    public SendResult dlqMsg(String msg) {
        return baseProducer.sendMsg("normal_topic", msg);
    }
}