rocketmq顺序消息、延时消息、事务消息等类型_mq延时消息 事务消息

maven依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.4</version>
</dependency>

一、普通消息

1、生产者：

// Initialize Consumer and set Consumer Goup Name
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("kingTestGroup");
   
    // Set the address of NameServer 
    consumer.setNamesrvAddr("192.168.124.16:9876");
    // Subscribe One or more of topics，and specify the tag filtering conditions, here specify * means receive all tag messages
    consumer.subscribe("kingTopic", "*");
    // Register a callback interface to handle messages received from the Broker
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
          System.out.println(new String(msg.getBody()));
        }
//        System.out.printf("%s Receive New Messages: %n",  msgs);
        // Return to the message consumption status, ConsumeConcurrentlyStatus.CONSUME_SUCCESS for successful consumption
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    // Start Consumer
    consumer.start();
    System.out.printf("Consumer Started.%n");

消费者：

// Initialize a producer and set the Producer group name
DefaultMQProducer producer = new DefaultMQProducer("kingTestGroup"); //（1）
// Set the address of NameServer
producer.setNamesrvAddr("192.168.124.16:9876");  //（2）
// Start Producer
producer.start();
for (int i = 0; i < 10; i++) {
  // Create a message and set the topic, tag, body and so on. The tag can be understood as a label to categorize the message, and RocketMQ can filter the tag on the consumer side.
  Message msg = new Message("kingTopic" /* Topic */,
    "TagA" /* Tag */,
    ("Hello king" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );   //（3）
  // Use the producer to send and wait for the result of sending synchronously
  SendResult sendResult = producer.send(msg);   //（4）
  System.out.printf("%s%n", sendResult);
}
// Close the producer once it is no longer in use
producer.shutdown();

二、顺序消息

通常会有这样的业务场景，一个订单的新建、扣钱、完成这三个消息必须要保证顺序消费，所以rocketmq提供了一种保证部分顺序的消息机制

生产者：

DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
            producer.start();
 
            String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 100; i++) {
                int orderId = i % 10;
                Message msg =
                    new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);
 
                System.out.printf("%s%n", sendResult);
            }
 
            producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }

消费者：

// Initialize Consumer and set Consumer Goup Name
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("kingTestGroup");
 
// Set the address of NameServer 
consumer.setNamesrvAddr("192.168.124.16:9876");
// Subscribe One or more of topics，and specify the tag filtering conditions, here specify * means receive all tag messages
consumer.subscribe("kingTopicOrder", "*");
// Register a callback interface to handle messages received from the Broker
consumer.registerMessageListener(new MessageListenerOrderly() {
  @Override
  public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
    for (MessageExt msg : list) {
      System.out.println(new String(msg.getBody()));
    }
    return ConsumeOrderlyStatus.SUCCESS;
  }
});
// Start Consumer
consumer.start();
System.out.printf("Consumer Started.%n");

三、延时消息

顾名思义，消息会延时发送，下图是延时的级别丢应的秒数，rocketmq不能延时指定的秒数。

 生产者：

DefaultMQProducer producer = new DefaultMQProducer("kingTestGroup"); //（1）
// Set the address of NameServer
producer.setNamesrvAddr("192.168.124.16:9876");  //（2）
producer.start();
 
int totalMessagesToSend = 100;
for (int i = 0; i < totalMessagesToSend; i++) {
  Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
  // This message will be delivered to consumer 10 seconds later.
  message.setDelayTimeLevel(3);
  // Send the message
  producer.send(message);
}
 
// Shutdown producer after use.
producer.shutdown();

四、批量发送消息

DefaultMQProducer producer = new DefaultMQProducer("kingTestGroup"); //（1）
// Set the address of NameServer
producer.setNamesrvAddr("192.168.124.16:9876");  //（2）
producer.start();
//If you just send messages of no more than 1MiB at a time, it is easy to use batch
//Messages of the same batch should have: same topic, same waitStoreMsgOK and no schedule support
String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "Tag", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "Tag", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "Tag", "OrderID003", "Hello world 2".getBytes()));
producer.send(messages);

五、事务消息

mq事务的作用就是保证本地事务与mq消息的一致性，换句话来说就是，保证本地事务执行完了，消息一定发送成功，本地事务失败了，发出去的消息要回滚，半消息的作用就是保证rocketmq是否正常。

public static void main(String[] args) throws MQClientException, InterruptedException {
    TransactionListener transactionListener = new TransactionListenerImpl();
    TransactionMQProducer producer = new TransactionMQProducer("kingTestGroup");
    producer.setNamesrvAddr("192.168.124.16:9876");  //（2）
    ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        }
    });
 
    producer.setExecutorService(executorService);
    producer.setTransactionListener(transactionListener);
    producer.start();
 
    String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
    for (int i = 0; i < 10; i++) {
        try {
            Message msg =
                new Message("TopicTest", 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);
 
            Thread.sleep(10);
        } catch (MQClientException | UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }
 
    for (int i = 0; i < 100000; i++) {
        Thread.sleep(1000);
    }
    producer.shutdown();
}
 
static class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
 
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
 
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        return LocalTransactionState.UNKNOW;
    }
 
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                default:
                    return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}