RocketMQ的使用

@Test
public void syncSendMessageTest() throws Exception {
  // 实例化消息生产者Producer
  DefaultMQProducer producer = new DefaultMQProducer("rocketmq-cluster", "sync_send_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 启动producer
  producer.start();

  for (int i = 0; i < 10; i++) {
    // 创建消息，并指定Topic，Tag和消息体
    Message message = new Message("sync_product", JSON.toJSONBytes("Hello RocketMQ Sync Send Message " + i));
    // 发送消息
    SendResult sendResult = producer.send(message);
    System.out.println("sendResult = " + JSON.toJSONString(sendResult));
  }

  TimeUnit.SECONDS.sleep(5);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}

@Test
public void asyncSendMessageTest() throws Exception {
  // 实例化消息生产者Producer
  DefaultMQProducer producer = new DefaultMQProducer("rocketmq-cluster", "async_send_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 启动producer
  producer.start();

  for (int i = 0; i < 10; i++) {
    // 创建消息，并指定Topic，Tag和消息体
    Message message = new Message("async_product", JSON.toJSONBytes("Hello RocketMQ Async Send Message " + i));
    // 发送消息
    producer.send(message, new SendCallback() {

      // 成功处理
      @Override
      public void onSuccess(SendResult sendResult) {
        System.out.println("act=onSuccess sendResult = " + JSON.toJSONString(sendResult));
      }

      // 异常处理
      @Override
      public void onException(Throwable throwable) {
        System.out.println("act=onException throwable = " + JSON.toJSONString(throwable));
      }
    });

  }

  TimeUnit.SECONDS.sleep(5);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}

@Test
public void onewaySendMessageTest() throws Exception {
  // 实例化消息生产者Producer
  DefaultMQProducer producer = new DefaultMQProducer("rocketmq-cluster", "oneway_send_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 启动producer
  producer.start();

  for (int i = 0; i < 10; i++) {
    // 创建消息，并指定Topic，Tag和消息体
    Message message = new Message("oneway_product", JSON.toJSONBytes("Hello RocketMQ Oneway Send Message " + i));
    // 发送消息， 不关心发送结果直接返回
    producer.sendOneway(message);
  }

  TimeUnit.SECONDS.sleep(5);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}

@Test
public void batchSendMessageTest() throws Exception {
  // 实例化消息生产者Producer
  DefaultMQProducer producer = new DefaultMQProducer("rocketmq-cluster", "batch_send_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 启动producer
  producer.start();

  // 构建消息
  List<Message> messages = new ArrayList<>();
  messages.add(new Message("batch_topic", JSON.toJSONBytes("Batch Message 0")));
  messages.add(new Message("batch_topic", JSON.toJSONBytes("Batch Message 1")));
  messages.add(new Message("batch_topic", JSON.toJSONBytes("Batch Message 2")));

  // 发送消息， 不关心发送结果直接返回
  SendResult sendResult = producer.send(messages);
  System.out.println("sendResult = " + JSON.toJSONString(sendResult));

  TimeUnit.SECONDS.sleep(5);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}

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;
  }
}

@Test
public void clusterConsumerTest() throws Exception {
  // 实例化消息生产者,指定组名
  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocketmq-cluster","clusterConsumerGroup");
  // 设置NameServer的地址
  consumer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 订阅Topic
  consumer.subscribe("sync_product", "*");
  // 集群模式消费
  consumer.setMessageModel(MessageModel.CLUSTERING);
  // 从队列尾部消费
  consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  // 注册回调函数，处理消息
  consumer.registerMessageListener(new MessageListenerConcurrently() {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                    ConsumeConcurrentlyContext context) {
      for (MessageExt msg : msgs) {
        System.out.printf("topic=%s tag=%s msgId=%s msgKey=%s msgBody=%s \n",
                          msg.getTopic(), msg.getTags(), msg.getMsgId(), msg.getKeys(), 
                          JSON.parseObject(msg.getBody(), String.class));
      }
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
  });
  //启动消息者
  consumer.start();
  TimeUnit.SECONDS.sleep(1000);
}

@Test
public void broadcastConsumerTest() throws Exception {
  // 实例化消息生产者,指定组名
  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocketmq-cluster","broadcastConsumerGroup");
  // 设置NameServer的地址
  consumer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 订阅Topic
  consumer.subscribe("sync_product", "*");
  // 广播模式消费
  consumer.setMessageModel(MessageModel.BROADCASTING);
  // 从队列尾部消费
  consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  // 注册回调函数，处理消息
  consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                    ConsumeConcurrentlyContext context) {
      for (MessageExt msg : msgs) {
        System.out.printf("topic=%s tag=%s msgId=%s msgKey=%s msgBody=%s \n",
                          msg.getTopic(), msg.getTags(), msg.getMsgId(), msg.getKeys(), 
                          JSON.parseObject(msg.getBody(), String.class));
      }
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
  });
  //启动消息者
  consumer.start();
  TimeUnit.SECONDS.sleep(1000);
}

@Test
public void sendOrderMessageTest() throws Exception {
  // 实例化消息生产者Producer
  DefaultMQProducer producer = new DefaultMQProducer("rocketmq-cluster", "order_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 启动producer
  producer.start();

  for (int i = 0; i < 10; i++) {
    // 创建消息，并指定Topic，Tag和消息体
    Message message = new Message("order_product", JSON.toJSONBytes("order Message " + i));
    // 发送消息
    SendResult sendResult = producer.send(message,new MessageQueueSelector() {

      @Override
      public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Integer argInt = (Integer) arg;
        return mqs.get(argInt % mqs.size());
      }
    }, 101);
    System.out.println("sendResult = " + JSON.toJSONString(sendResult));
  }

  TimeUnit.SECONDS.sleep(10);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}


@Test
public void orderConsumerTest3() throws Exception {
  // 实例化消息生产者,指定组名
  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocketmq-cluster", "orderConsumerGroup");
  // 设置NameServer的地址
  consumer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 订阅Topic
  consumer.subscribe("order_product", "*");
  // 集群模式消费
  consumer.setMessageModel(MessageModel.CLUSTERING);
  // 从队列尾部消费
  consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  // 注册回调函数，处理消息
  consumer.registerMessageListener(new MessageListenerOrderly() {

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
      for (MessageExt msg : msgs) {
        System.out.printf("topic=%s tag=%s msgId=%s msgKey=%s msgBody=%s \n",
                          msg.getTopic(), msg.getTags(), msg.getMsgId(), msg.getKeys(),
                          JSON.parseObject(msg.getBody(), String.class));
      }
      return ConsumeOrderlyStatus.SUCCESS;
    }

  });
  //启动消息者
  consumer.start();
  TimeUnit.SECONDS.sleep(1000);
}

@Test
public void testDelayMessage() throws Exception {
  // 实例化消息生产者Producer
  DefaultMQProducer producer = new DefaultMQProducer("rocketmq-cluster", "delay_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  // 启动producer
  producer.start();

  for (int i = 0; i < 10; i++) {
    // 创建消息，并指定Topic，Tag和消息体
    Message message = new Message("delay_product", JSON.toJSONBytes("Hello RocketMQ Delay Send Message " + i));
    // 设置消息延迟等级
    message.setDelayTimeLevel(4);
    // 发送消息
    SendResult sendResult = producer.send(message);
    System.out.println("sendResult = " + JSON.toJSONString(sendResult));
  }

  TimeUnit.SECONDS.sleep(10);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}

// 利用Tag进行过滤，接收所有Tag消息为 *
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

// a取自消息的属性，即Message.putUserProperty("a", String.valueOf(num))
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");

@Test
public void sendTransactionMessageTest() throws Exception {
  // 实例化消息生产者Producer
  TransactionMQProducer producer = new TransactionMQProducer("rocketmq-cluster", "transaction_message_producer_group");
  // 设置NameServer的地址
  producer.setNamesrvAddr("192.168.22.162:9876;192.168.22.163:9876");
  producer.setTransactionListener(new OrderTransactionListenerImpl());
  // 启动producer
  producer.start();

  for (int i = 0; i < 10; i++) {
    // 创建消息，并指定Topic，Tag和消息体
    Message message = new Message("transaction_product", JSON.toJSONBytes("Order  Transaction Message"));
    // 发送消息
    SendResult sendResult = producer.sendMessageInTransaction(message, null);
    System.out.println("sendResult = " + JSON.toJSONString(sendResult));
  }

  TimeUnit.SECONDS.sleep(1000);
  // 如果不再发送消息，关闭Producer实例。
  producer.shutdown();
}

public class OrderTransactionListenerImpl implements TransactionListener {

  // 这里不应该是本地缓存，应该使用redis等
  private static ConcurrentHashMap<String, Integer> countCheckNumMap = new ConcurrentHashMap<>();

  // 最大检查次数
  private static final int MAX_CHECK = 5;

  @Override
  public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    try {
      System.out.println("transactionId = " + msg.getTransactionId());
      //执行本地事务
      System.out.println("执行本地事务并保存TransactionId到一张表中");
      // 模拟执行过程中发送异常
      boolean exeException = RandomUtil.randomBoolean();
      if (exeException) {
        System.out.println("transactionId = " + msg.getTransactionId() + " 执行过程中遇到了异常");
        throw new RuntimeException("执行过程中遇到了异常");
      }
      // 模拟执行过程宕机，回滚
      boolean unknowException = RandomUtil.randomBoolean();
      if (unknowException) {
        System.out.println("transactionId = " + msg.getTransactionId() + " 执行过程中遇到了中断执行");
        return LocalTransactionState.UNKNOW;
      }
    } catch (Throwable e) {
      return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    System.out.println("transactionId = " + msg.getTransactionId() + " 提交");
    return LocalTransactionState.COMMIT_MESSAGE;
  }

  @Override
  public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    System.out.println("checkLocalTransaction transactionId = " + msg.getTransactionId());
    // 从数据库中查找是否有此transactionId，有返回commit，没有返回unknow，示例代码用随机了
    boolean findTransaction = RandomUtil.randomBoolean();
    if (findTransaction) {
      System.out.println("提交 transactionId = " + msg.getTransactionId());
      return LocalTransactionState.COMMIT_MESSAGE;
    }
    return rollbackOrUnknow(msg.getTransactionId());
  }

  private LocalTransactionState rollbackOrUnknow(String transactionId) {
    Integer countCheckNum = countCheckNumMap.get(transactionId);

    if (countCheckNum != null && countCheckNum > MAX_CHECK) {
      System.out.println("回滚 transactionId = " + transactionId);
      return LocalTransactionState.ROLLBACK_MESSAGE;
    }

    if (countCheckNum == null) {
      countCheckNumMap.put(transactionId, 1);
    }

    System.out.println("未知 transactionId = " + transactionId);
    return LocalTransactionState.UNKNOW;
  }
}