RocketMQ基本知识

深藏阁楼爱情的钟 2021-11-26 10:12 281阅读 0赞

## RocketMQ ##

rocketmq对比kafaka，它支持预定消息、消息跟踪、广播消息等等。  
[更多参考][Link 1]

## 启动 ##

1.  启动nameserver

> nohup sh bin/mqnamesrv &
      > tail -f ~/logs/rocketmqlogs/namesrv.log
      The Name Server boot success...

1.  启动Broker

> nohup sh bin/mqbroker -n localhost:9876 &
     > tail -f ~/logs/rocketmqlogs/broker.log 
     The broker[%s, 172.30.30.233:10911] boot success...

## 关闭 ##

> sh bin/mqshutdown broker
    The mqbroker(36695) is running...
    Send shutdown request to mqbroker(36695) OK
    
    > sh bin/mqshutdown namesrv
    The mqnamesrv(36664) is running...
    Send shutdown request to mqnamesrv(36664) OK

## 发送消息 ##

可以发送的消息有三种：同步发送、异步发送、单向发送  
特殊的是单向发送这种模式，这种不等待服务器返回信息，速度快。

下面举一个例子，比如同步发送

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

public class SyncProducer {
        
        public static void main(String[] args) throws Exception {
        
            //Instantiate with a producer group name.
            DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
            // Specify name server addresses.
            producer.setNamesrvAddr("localhost:9876");
            //Launch the instance.
            producer.start();
            for (int i = 0; i < 100; i++) {
        
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                        i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                //Call send message to deliver message to one of brokers.
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            }
            //Shut down once the producer instance is not longer in use.
            producer.shutdown();
        }
    }

## 接受消息 ##

public class Consumer {
        
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
        
    
            // Instantiate with specified consumer group name.
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
             
            // Specify name server addresses.
            consumer.setNamesrvAddr("localhost:9876");
            
            // Subscribe one more more topics to consume.
            consumer.subscribe("TopicTest", "*");
            // Register callback to execute on arrival of messages fetched from brokers.
            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;
                }
            });
    
            //Launch the consumer instance.
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }

## 局部有序消息 ##

全局有序消息：由于一个 topic 只有一个 queue ，即使我们有多个 producer 实例和 consumer 实例也很难提高消息吞吐量。就好比过独木桥，大家只能一个挨着一个过去，效率低下。

局部有序消息：常见做法就是将 order id 进行处理，将 order id 相同的消息发送到 topicB 的同一个 queue，假设我们 topicB 有 2 个 queue，那么我们可以简单的对 id 取余，奇数的发往 queue0，偶数的发往 queue1，消费者按照 queue 去消费时，就能保证 queue0 里面的消息有序消费，queue1 里面的消息有序消费。

创建一个topic，默认有4个queue，我们这里保证了每一个queue是有序的。这种可以提高吞吐。

public class OrderedProducer {
        
        public static void main(String[] args) throws Exception {
        
            //Instantiate with a producer group name.
            MQProducer producer = new DefaultMQProducer("example_group_name");
            //Launch the instance.
            producer.start();
            String[] tags = new String[] {
        "TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 100; i++) {
        
                int orderId = i % 10;
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTestjjj", 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); // orderId会传递到上面的arg里面
    
                System.out.printf("%s%n", sendResult);
            }
            //server shutdown
            producer.shutdown();
        }
    }

处理消息接受

public class OrderedConsumer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTest", "TagA || TagC || TagD");
    
            consumer.registerMessageListener(new MessageListenerOrderly() {
        
    
                AtomicLong consumeTimes = new AtomicLong(0);
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                           ConsumeOrderlyContext context) {
        
                    context.setAutoCommit(false);
                    System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                    this.consumeTimes.incrementAndGet();
                    if ((this.consumeTimes.get() % 2) == 0) {
        
                        return ConsumeOrderlyStatus.SUCCESS;
                    } else if ((this.consumeTimes.get() % 3) == 0) {
        
                        return ConsumeOrderlyStatus.ROLLBACK;
                    } else if ((this.consumeTimes.get() % 4) == 0) {
        
                        return ConsumeOrderlyStatus.COMMIT;
                    } else if ((this.consumeTimes.get() % 5) == 0) {
        
                        context.setSuspendCurrentQueueTimeMillis(3000);
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
    
                }
            });
    
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }

处理消息接受和异常处理  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE5MTA3MDEx_size_16_color_FFFFFF_t_70]  
消息消费的顺序问题：

并发情况下，返回RECONSUME\_LATER，表示过一会儿再消费，先去消费其他消息  
有序情况下，返回SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT，表示等一会儿再消费，无法消费其他消息  
可以通过getReconsumeTimes获取重试次数  
参考https://www.cnblogs.com/happyflyingpig/p/8207003.html

## 广播模式 ##

广播正在向主题的所有订阅者发送消息。  
如果您希望所有订阅者都收到有关主题的消息，则广播是一个不错的选择。

public class BroadcastProducer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            producer.start();
    
            for (int i = 0; i < 100; i++){
        
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            }
            producer.shutdown();
        }
    }

public class BroadcastConsumer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //set to broadcast mode
            consumer.setMessageModel(MessageModel.BROADCASTING);
    
            consumer.subscribe("TopicTest", "TagA || TagC || TagD");
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
        
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
        
                    System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            consumer.start();
            System.out.printf("Broadcast Consumer Started.%n");
        }
    }

## 延迟消息 ##

预定消息与普通消息的不同之处在于，它们将在稍后提供的时间之前传送。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
     import org.apache.rocketmq.common.message.Message;
        
     public class ScheduledMessageProducer {
        
        
         public static void main(String[] args) throws Exception {
        
             // Instantiate a producer to send scheduled messages
             DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
             // Launch producer
             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();
         }
            
     }

## 批量发送 ##

批量发送应该需要保持同一个主题，大小不超过1mb。

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();
        //handle the error
    }

[详情参考][Link 2]

## 消息的tag过滤 ##

在大多数情况下，tag是一种简单实用的设计，用于选择所需的消息。  
例如：

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

消息的tag这可以有一个，这对于复杂的需求是没办法满足的，不过我们可以使用properties配合sql语法过滤需要的消息。  
[具体参考][Link 2]

## RocketMQ事务 ##

事务状态

三种状态：
    （1）TransactionStatus.CommitTransaction：提交事务，它意味着允许消费者使用此消息。
    （2）TransactionStatus.RollbackTransaction：回滚事务，表示该消息将被删除而不允许使用。
    （3）TransactionStatus.Unknown：中间状态，表示需要MQ检查以确定状态。

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import java.util.List;
    
    public class TransactionProducer {
        
        public static void main(String[] args) throws MQClientException, InterruptedException {
        
            TransactionListener transactionListener = new TransactionListenerImpl();
            TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
            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("TopicTest1234", 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();
        }
    }

import ...
       
       public 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) {
        
           	   //这里获取本地事务的业务ID，返回UNKON是为了
           	   //重试checkLocalTransaction方法
               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;
                   }
               }
               return LocalTransactionState.COMMIT_MESSAGE;
           }
       }

事务的处理参考  
[https://yq.aliyun.com/articles/684259][https_yq.aliyun.com_articles_684259]  
[http://rocketmq.apache.org/docs/transaction-example/][http_rocketmq.apache.org_docs_transaction-example]

## 消息日志配置 ##

log4j xml配置

logback配置

<appender name="mqAppender1" class="org.apache.rocketmq.logappender.logback.RocketmqLogbackAppender">
        <tag>yourTag</tag>
        <topic>yourLogTopic</topic>
        <producerGroup>yourLogGroup</producerGroup>
        <nameServerAddress>yourRocketmqNameserverAddress</nameServerAddress>
        <layout>
            <pattern>%date %p %t - %m%n</pattern>
        </layout>
    </appender>
    
    <appender name="mqAsyncAppender1" class="ch.qos.logback.classic.AsyncAppender">
        <queueSize>1024</queueSize>
        <discardingThreshold>80</discardingThreshold>
        <maxFlushTime>2000</maxFlushTime>
        <neverBlock>true</neverBlock>
        <appender-ref ref="mqAppender1"/>
    </appender>

[参考][Link 3]

## 其他 ##

[OpenMessaging][]

## 需要注意的问题 ##

*  消息的异常处理：发送端发送消息异常可以重新发送、接受端可以重新消费
 *  消息的重复  
    1、消费端处理消息的业务逻辑保持幂等性  
    2、保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现
 *  顺序问题  
    rocketmq有局部有序
 *  消息丢失  
    rocketmq特性

## 参考博客 ##

*  [RocketMQ的有序消息][RocketMQ]
 *  [深入解析中间件之-RocketMQ][-RocketMQ]
 *  [RocketMQ][RocketMQ 1]
 *  [https://blog.csdn.net/abennet/article/details/79402776][https_blog.csdn.net_abennet_article_details_79402776]

[Link 1]: https://rocketmq.apache.org/docs/motivation/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE5MTA3MDEx_size_16_color_FFFFFF_t_70]: /images/20211126/5a217b99ab6d4b018bb3a58af3e6c02e.png
[Link 2]: http://rocketmq.apache.org/docs/schedule-example/
[https_yq.aliyun.com_articles_684259]: https://yq.aliyun.com/articles/684259
[http_rocketmq.apache.org_docs_transaction-example]: http://rocketmq.apache.org/docs/transaction-example/
[Link 3]: http://rocketmq.apache.org/docs/logappender-example/
[OpenMessaging]: http://rocketmq.apache.org/docs/openmessaging-example/
[RocketMQ]: https://blog.csdn.net/alex_xfboy/article/details/84943022
[-RocketMQ]: http://zqhxuyuan.github.io/2017/10/18/Midd-RocketMQ/
[RocketMQ 1]: http://rocketmq.apache.org/
[https_blog.csdn.net_abennet_article_details_79402776]: https://blog.csdn.net/abennet/article/details/79402776