学习RocketMQ

​

​

​

​

​

​

​

 public class SyncProducer {
     public static void main(String[] args) throws Exception {
         // 实例化消息生产者Producer
         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
         // 设置NameServer的地址
         producer.setNamesrvAddr("localhost:9876");
         // 启动Producer实例
         producer.start();
         for (int i = 0; i < 100; i++) {
             // 创建消息，并指定Topic，Tag和消息体
             Message msg = new Message("TopicTest" /* Topic */,
             "TagA" /* Tag */,
             ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
             );
             // 发送消息到一个Broker
             SendResult sendResult = producer.send(msg);
             // 通过sendResult返回消息是否成功送达
             System.out.printf("%s%n", sendResult);
         }
         // 如果不再发送消息，关闭Producer实例。
         producer.shutdown();
     }
 }

 public class AsyncProducer {
     public static void main(String[] args) throws Exception {
         // 实例化消息生产者Producer
         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
         // 设置NameServer的地址
         producer.setNamesrvAddr("localhost:9876");
         // 启动Producer实例
         producer.start();
         producer.setRetryTimesWhenSendAsyncFailed(0);
         for (int i = 0; i < 100; i++) {
                 final int index = i;
                 // 创建消息，并指定Topic，Tag和消息体
                 Message msg = new Message("TopicTest",
                     "TagA",
                     "OrderID188",
                     "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                 // SendCallback接收异步返回结果的回调
                 producer.send(msg, new SendCallback() {
                     @Override
                     public void onSuccess(SendResult sendResult) {
                         System.out.printf("%-10d OK %s %n", index,
                             sendResult.getMsgId());
                     }
                     @Override
                     public void onException(Throwable e) {
                       System.out.printf("%-10d Exception %s %n", index, e);
                       e.printStackTrace();
                     }
                 });
         }
         // 如果不再发送消息，关闭Producer实例。
         producer.shutdown();
     }
 }

 public class OnewayProducer {
     public static void main(String[] args) throws Exception{
         // 实例化消息生产者Producer
         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
         // 设置NameServer的地址
         producer.setNamesrvAddr("localhost:9876");
         // 启动Producer实例
         producer.start();
         for (int i = 0; i < 100; i++) {
             // 创建消息，并指定Topic，Tag和消息体
             Message msg = new Message("TopicTest" /* Topic */,
                 "TagA" /* Tag */,
                 ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
             );
             // 发送单向消息，没有任何返回结果
             producer.sendOneway(msg);
 ​
         }
         // 如果不再发送消息，关闭Producer实例。
         producer.shutdown();
     }
 }

 public static void main(String[] args) throws Exception {
     // 实例化消息生产者,指定组名
     DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
     // 指定Namesrv地址信息.
     consumer.setNamesrvAddr("localhost:9876");
     // 订阅Topic
     consumer.subscribe("Test", "*");
     //负载均衡模式消费
     consumer.setMessageModel(MessageModel.CLUSTERING);
     // 注册回调函数，处理消息
     consumer.registerMessageListener(new MessageListenerConcurrently() {
         @Override
         public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                         ConsumeConcurrentlyContext context) {
             System.out.printf("%s Receive New Messages: %s %n", 
                               Thread.currentThread().getName(), msgs);
             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
         }
     });
     //启动消息者
     consumer.start();
     System.out.printf("Consumer Started.%n");
 }

 public static void main(String[] args) throws Exception {
     // 实例化消息生产者,指定组名
     DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
     // 指定Namesrv地址信息.
     consumer.setNamesrvAddr("localhost:9876");
     // 订阅Topic
     consumer.subscribe("Test", "*");
     //广播模式消费
     consumer.setMessageModel(MessageModel.BROADCASTING);
     // 注册回调函数，处理消息
     consumer.registerMessageListener(new MessageListenerConcurrently() {
         @Override
         public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                         ConsumeConcurrentlyContext context) {
             System.out.printf("%s Receive New Messages: %s %n", 
                               Thread.currentThread().getName(), msgs);
             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
         }
     });
     //启动消息者
     consumer.start();
     System.out.printf("Consumer Started.%n");
 }

 /**
 * Producer，发送顺序消息
 */
 public class Producer {
 ​
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
 ​
        producer.setNamesrvAddr("127.0.0.1:9876");
 ​
        producer.start();
 ​
        String[] tags = new String[]{"TagA", "TagC", "TagD"};
 ​
        // 订单列表
        List<OrderStep> orderList = new Producer().buildOrders();
 ​
        Date date = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String dateStr = sdf.format(date);
        for (int i = 0; i < 10; i++) {
            // 加个时间前缀
            String body = dateStr + " Hello RocketMQ " + orderList.get(i);
            Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());
 ​
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long id = (Long) arg;  //根据订单id选择发送queue
                    long index = id % mqs.size();
                    return mqs.get((int) index);
                }
            }, orderList.get(i).getOrderId());//订单id
 ​
            System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
                sendResult.getSendStatus(),
                sendResult.getMessageQueue().getQueueId(),
                body));
        }
 ​
        producer.shutdown();
    }
 ​
    /**
     * 订单的步骤
     */
    private static class OrderStep {
        private long orderId;
        private String desc;
 ​
        public long getOrderId() {
            return orderId;
        }
 ​
        public void setOrderId(long orderId) {
            this.orderId = orderId;
        }
 ​
        public String getDesc() {
            return desc;
        }
 ​
        public void setDesc(String desc) {
            this.desc = desc;
        }
 ​
        @Override
        public String toString() {
            return "OrderStep{" +
                "orderId=" + orderId +
                ", desc='" + desc + '\'' +
                '}';
        }
    }
 ​
    /**
     * 生成模拟订单数据
     */
    private List<OrderStep> buildOrders() {
        List<OrderStep> orderList = new ArrayList<OrderStep>();
 ​
        OrderStep orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111065L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103117235L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111065L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103117235L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111065L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("推送");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103117235L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);
 ​
        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);
 ​
        return orderList;
    }
 }

/**
 * 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
 */
 public class ConsumerInOrder {
 ​
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new 
            DefaultMQPushConsumer("please_rename_unique_group_name_3");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 ​
        consumer.subscribe("TopicTest", "TagA || TagC || TagD");
 ​
        consumer.registerMessageListener(new MessageListenerOrderly() {
 ​
            Random random = new Random();
 ​
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                for (MessageExt msg : msgs) {
                    // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));
                }
 ​
                try {
                    //模拟业务逻辑处理中...
                    TimeUnit.SECONDS.sleep(random.nextInt(10));
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
 ​
        consumer.start();
 ​
        System.out.println("Consumer Started.");
    }
 }

 public class ScheduledMessageConsumer {
    public static void main(String[] args) throws Exception {
       // 实例化消费者
       DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
       // 订阅Topics
       consumer.subscribe("TestTopic", "*");
       // 注册消息监听者
       consumer.registerMessageListener(new MessageListenerConcurrently() {
           @Override
           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
               for (MessageExt message : messages) {
                   // Print approximate delay time period
                   System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
               }
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           }
       });
       // 启动消费者
       consumer.start();
   }
 }

 public class ScheduledMessageProducer {
    public static void main(String[] args) throws Exception {
       // 实例化一个生产者来产生延时消息
       DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
       // 启动生产者
       producer.start();
       int totalMessagesToSend = 100;
       for (int i = 0; i < totalMessagesToSend; i++) {
           Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
           // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
           message.setDelayTimeLevel(3);
           // 发送消息
           producer.send(message);
       }
        // 关闭生产者
       producer.shutdown();
   }
 }

 // org/apache/rocketmq/store/config/MessageStoreConfig.java
 private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

 String topic = "BatchTest";
 List<Message> messages = new ArrayList<>();
 messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
 messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
 messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
 try {
    producer.send(messages);
 } catch (Exception e) {
    e.printStackTrace();
    //处理error
 }

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 nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            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字节
            if (tmpSize > SIZE_LIMIT) {
                //单个消息超过了最大的限制
                //忽略,否则会阻塞分裂的进程
                if (nextIndex - currIndex == 0) {
                   //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
                   nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
 ​
        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
 }
 //把大的消息分裂成若干个小的消息
 ListSplitter splitter = new ListSplitter(messages);
 while (splitter.hasNext()) {
   try {
       List<Message>  listItem = splitter.next();
       producer.send(listItem);
   } catch (Exception e) {
       e.printStackTrace();
       //处理error
   }
 }

 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
 consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

 public void subscribe(finalString topic, final MessageSelector messageSelector)

 DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
 producer.start();
 Message msg = new Message("TopicTest",
    tag,
    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
 );
 // 设置一些属性
 msg.putUserProperty("a", String.valueOf(i));
 SendResult sendResult = producer.send(msg);
 ​
 producer.shutdown();

 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
 // 只有订阅的消息有这个属性a, a >=0 and a <= 3
 consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
 consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
 });
 consumer.start();

 public class Producer {
     public static void main(String[] args) throws MQClientException, InterruptedException {
         //创建事务监听器
         TransactionListener transactionListener = new TransactionListenerImpl();
         //创建消息生产者
         TransactionMQProducer producer = new TransactionMQProducer("group6");
         producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
         //生产者这是监听器
         producer.setTransactionListener(transactionListener);
         //启动消息生产者
         producer.start();
         String[] tags = new String[]{"TagA", "TagB", "TagC"};
         for (int i = 0; i < 3; i++) {
             try {
                 Message msg = new Message("TransactionTopic", tags[i % tags.length], "KEY" + i,
                         ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                 SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                 System.out.printf("%s%n", sendResult);
                 TimeUnit.SECONDS.sleep(1);
             } catch (MQClientException | UnsupportedEncodingException e) {
                 e.printStackTrace();
             }
         }
         //producer.shutdown();
     }
 }

 public class TransactionListenerImpl implements TransactionListener {
 ​
     @Override
     public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
         System.out.println("执行本地事务");
         if (StringUtils.equals("TagA", msg.getTags())) {
             return LocalTransactionState.COMMIT_MESSAGE;
         } else if (StringUtils.equals("TagB", msg.getTags())) {
             return LocalTransactionState.ROLLBACK_MESSAGE;
         } else {
             return LocalTransactionState.UNKNOW;
         }
 ​
     }
 ​
     @Override
     public LocalTransactionState checkLocalTransaction(MessageExt msg) {
         System.out.println("MQ检查消息Tag【"+msg.getTags()+"】的本地事务执行结果");
         return LocalTransactionState.COMMIT_MESSAGE;
     }
 }

@Slf4j
@Component
@RequiredArgsConstructor
public class GeneralMessageDemoProduce {

    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 发送普通消息
     *
     * @param topic            消息发送主题，用于标识同一类业务逻辑的消息
     * @param tag              消息的过滤标签，消费者可通过Tag对消息进行过滤，仅接收指定标签的消息。
     * @param keys             消息索引键，可根据关键字精确查找某条消息
     * @param messageSendEvent 普通消息发送事件，自定义对象，最终都会序列化为字符串
     * @return 消息发送 RocketMQ 返回结果
     */
    public SendResult sendMessage(String topic, String tag, String keys, GeneralMessageEvent messageSendEvent) {
        SendResult sendResult;
        try {
            StringBuilder destinationBuilder = StrUtil.builder().append(topic);
            if (StrUtil.isNotBlank(tag)) {
                destinationBuilder.append(":").append(tag);
            }
            Message<?> message = MessageBuilder
                    .withPayload(messageSendEvent)
                    .setHeader(MessageConst.PROPERTY_KEYS, keys)
                    .setHeader(MessageConst.PROPERTY_TAGS, tag)
                    .build();
            sendResult = rocketMQTemplate.syncSend(
                    destinationBuilder.toString(),
                    message,
                    2000L
            );
            log.info("[普通消息] 消息发送结果：{}，消息ID：{}，消息Keys：{}", sendResult.getSendStatus(), sendResult.getMsgId(), keys);
        } catch (Throwable ex) {
            log.error("[普通消息] 消息发送失败，消息体：{}", JSON.toJSONString(messageSendEvent), ex);
            throw ex;
        }
        return sendResult;
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "rocketmq-demo_common-message_topic",
        selectorExpression = "general",
        consumerGroup = "rocketmq-demo_general-message_cg"
)
public class GeneralMessageDemoConsume implements RocketMQListener<GeneralMessageEvent> {

    @Override
    public void onMessage(GeneralMessageEvent message) {
        log.info("接到到RocketMQ消息，消息体：{}", JSON.toJSONString(message));
    }
}

 @PostMapping("/test/send/general-message")
    @Operation(summary = "发送RocketMQ普通消息")
    public String sendGeneralMessage() {
        String keys = UUID.randomUUID().toString();
        GeneralMessageEvent generalMessageEvent = GeneralMessageEvent.builder()
                .body("消息具体内容，可以是自定义对象，最终都会序列化为字符串。如果是取消订单，这里应该是订单ID或者相关联的信息")
                .keys(keys)
                .build();
        SendResult sendResult = generalMessageDemoProduce.sendMessage(
                "rocketmq-demo_common-message_topic",
                "general",
                keys,
                generalMessageEvent
        );
        return sendResult.getSendStatus().name();
    }