【RocketMQ】基本使用：Java操作RocketMQ（rocketmq-client）

短命女 2022-12-22 08:35 231阅读 0赞

## 1.引入依赖 ##

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

## 2.Producer ##

public class MyProducer {
        
    
        public static void main(String[] args) throws Exception{
        
            // 构造Producer时，必须指定groupId
            DefaultMQProducer producer = new DefaultMQProducer("my_producer_group");
            // 只用namesrv的地址就行，它会从namesrv上拿到broker的地址和topic信息
            producer.setNamesrvAddr("43.107.136.120:9876"); 
            producer.start();
    
            int num = 0;
            while (num < 20) {
        
                num++;
                 /**
                  * rocketmq封装了Message
         	      * String topic, 
         	      * String tags, 标签（分类）---> 筛选
         	      * byte[] body
         	      */
                Message message = new Message("my_test_topic", "", ("hello rocketmq:" + num).getBytes());
                // 发送消息，拿到返回SendResult
                SendResult result = producer.send(message); 
                System.out.println(result);
            }
        }
    }

启动并发送消成功后，返回的SendResult如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

SendResult中，有一个sendStatus状态，表示消息的发送状态。一共有四种状态

*  FLUSH\_DISK\_TIMEOUT ： 表示没有在规定时间内完成刷盘（需要Broker 的刷盘策Ill创立设置成 SYNC\_FLUSH 才会报这个错误）
 *  FLUSH\_SLAVE\_TIMEOUT ：表示在主备方式下，并且Broker 被设置成SYNC\_MASTER 方式，没有在设定时间内完成主从同步。
 *  SLAVE\_NOT\_AVAILABLE ： 这个状态产生的场景和FLUSH\_SLAVE\_TIMEOUT 类似， 表示在主备方式下，并且Broker 被设置成SYNC\_MASTER ，但是没有找到被配置成Slave 的Broker 。
 *  SEND OK ：表示发送成功，发送成功的具体含义，比如消息是否已经被存储到磁盘？消息是否被同步到了Slave 上？消息在Slave 上是否被写入磁盘？需要结合所配置的刷盘策略、主从策略来定。这个状态还可以简单理解为，没有发生上面列出的三个问题状态就是SEND OK

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

## 3.Consumer ##

public class MyConsumer {
        
    
        public static void main(String[] args) throws MQClientException {
        
            // 构造Consumer时，必须指定groupId
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_group");
            consumer.setNamesrvAddr("39.105.136.112:9876"); // nameServer地址,用于获取broker、topic信息
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 指定订阅的主题与tag，通过tag可以定制性消费（*表示全部tag）
            consumer.subscribe("my_test_topic", "*"); 
    		
            // 异步消费
            consumer.registerMessageListener(new MessageListenerConcurrently() {
        
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
        
                    System.out.println("Receive Message:" + msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 签收
                }
            });
            consumer.start();
        }
    }

收到消息的内容：

![在这里插入图片描述][2020111815355513.png_pic_center]

consumer Group：位于同一个consumer Group中的consumer实例

*  和producer Group中的各个produer实例承担的角色类似
 *  同一个group中可以配置多个consumer，可以提高消费端的并发**消费能力**以及容灾
 *  和kafka一样，多个consumer会对消息做**负载均衡**，意味着同一个topic下的不同messageQueue会分发给同一个group中的不同consumer。
 *  同时，如果我们希望消息能够达到**广播**的目的，那么只需要把consumer加入到不同的group就行。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/824507fbc0df4fbfb9db0ce01bcd0966.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/5590e912df964daa85dac1308e17dc34.png
[2020111815355513.png_pic_center]: /images/20221120/f386b7cc986c4365ba76d7663cf4d77e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/23abba8800264d349cdf2b595b347930.png