RocketMQ入门案例一：消息生产与消费

冷不防 2023-10-03 13:19 11阅读 0赞

### 1、版本介绍 ###

RocketMQ4.8.0

**maven依赖：**

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

### 2、发送消息的方式 ###

一般的MQ发送消息的方式无外乎：同步发送、异步发送，而RocketMQ除此之还有一中发送消息的方式：单向发送（只管发送消息、不需要broker回复确认）。  
在同步和异步发送消息时，我们可以选择批量发送消息（Message list）。

#### 1）同步发送消息 ####

生产者发送消息过程中进入到同步等待状态；它可以保证消息投递一定到达。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    
    import java.nio.charset.StandardCharsets;
    
    /**
     * RocketMQ生产者--同步生产
     */
    public class Producer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer("saint-study");
            // 设置nameserver地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.start();
    
            // topic 和body
            Message msg = new Message("saint-study-topic", "study002".getBytes(StandardCharsets.UTF_8));
            // 同步获取生产消息结果
            SendResult send = producer.send(msg);
            System.out.println("sendResult: " + send);
    
            // 关闭生产者
            producer.shutdown();
            System.out.println("已经停机");
        }
    }

#### 2）异步发送消息 ####

异步非阻塞的发送消息，采用事件监听的方式等待broker返回的ACK确认。代码中的onSuccess()和onException()方法可以理解为broker的ACK回传，success -->1，exception --> 0

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    
    import java.nio.charset.StandardCharsets;
    
    /**
     * 异步发送
     */
    public class ProducerAsync {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer("saint-study");
            // 设置nameserver地址
            producer.setNamesrvAddr("106.10.100.100:9876");
            producer.start();
    
            // topic 和body
            Message msg = new Message("saint-study-topic", "original-debug".getBytes(StandardCharsets.UTF_8));
    
    
            producer.send(msg, new SendCallback() {
        
                /**
                 * 消息发送成功
                 */
                @Override
                public void onSuccess(SendResult sendResult) {
        
                    System.out.println("消息发送成功");
                    System.out.println("sendResult  " + sendResult);
                }
    
                /**
                 * 发送消息出现异常
                 */
                @Override
                public void onException(Throwable throwable) {
        
                    // 发生异常时，尝试重试，或 日志记录
                    System.out.println("发送异常");
                }
            });
        }
    }

#### 3）单向发送 ####

只发送消息，不等待broker响应。  
这种方式最显著的特点：发送消息耗时非常短，通常在微妙级别。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.message.Message;
    
    import java.nio.charset.StandardCharsets;
    
    /**
     * 单向发送
     */
    public class ProducerSingleDirec {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer("saint-study");
            // 设置nameserver地址
            producer.setNamesrvAddr("106.10.100.100:9876");
            producer.start();
    
            // topic 和body
            Message msg = new Message("saint-study-topic", "study004-async".getBytes(StandardCharsets.UTF_8));
    
            // 单向发送。只管发送消息，不需要broker回复ACK确认
            producer.sendOneway(msg);
        }
    }

#### 4）批量发送消息 ####

只需要将Message改为List Message发送即可。  
优点：批量发送会减少网络传输开销，进而加快响应时长。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    
    import java.nio.charset.StandardCharsets;
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * RocketMQ生产者
     */
    public class Producer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer("saint-study");
            // 设置nameserver地址
            producer.setNamesrvAddr("106.15.139.143:9876");
            producer.start();
    
            // topic 和body
            Message msg = new Message("saint-study-topic", "study002".getBytes(StandardCharsets.UTF_8));
            Message msg2 = new Message("saint-study-topic", "study003".getBytes(StandardCharsets.UTF_8));
    
            List<Message> list = new ArrayList<>();
            list.add(msg);
            list.add(msg2);
            // 同步发送 -- 批量消息发送
            SendResult send = producer.send(list);
    
            System.out.println("sendResult: " + send);
    
            // 关闭生产者
            producer.shutdown();
            System.out.println("已经停机");
        }
    }

### 3、消息消费的方式 ###

#### 1）消费传播方式 ####

RocketMQ的消息传播方式有两种，分别为：集群消息（默认）、广播消息。  
至于一条消息是广播消息还是集群消息由consumer客户端决定。

**设置方式如下：**

// 方式一：广播模式，每个consumer都可以消费到消息
    consumer.setMessageModel(MessageModel.BROADCASTING);
    // 方式二： 集群模式，只有一个consumer能消费到消息。
    consumer.setMessageModel(MessageModel.CLUSTERING);

##### 1. 集群消息 #####

集群化部署消费者，一条消息只能由consumer group中的一个消费者消费。

**消息分摊机制**：相同Consumer Group的每个consumer实例评分分摊消息。

**特点：**

*  每条消息只能被处理一次，broker只会把消息发送到消费者集群中的一个consumer实例。
 *  在消息重投时，不能保证路由到同一台consumer实例上。
 *  消费状态由broker维护。

###### 2. 广播消息 ######

相同的Consumer Group中的每个实例都会接收到全量的消息。

**特点:**

*  消费进度由consumer维护。
 *  保证每个consumer实例消费一次消息。
 *  消费失败的消息不会重投。

**注：一个topic的consumer group中的消费者不允许即是广播模式 又是集群模式，否者会出现消息错乱问题。**

#### 2）实例 ####

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.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    
    import java.util.List;
    
    /**
     * 消息消费者
     */
    public class Consumer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("study-consumer");
            consumer.setNamesrvAddr("106.10.100.100:9876");
    
            // 每个consumer 只关注一个topic
            // topic , 过滤器 * 表示不过滤
            consumer.subscribe("saint-study-topic", "*");
            consumer.setConsumeTimeout(20L);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            // 消费方式
            consumer.setMessageModel(MessageModel.CLUSTERING);
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
        
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        
                    for (MessageExt msg : msgs) {
        
                        System.out.println(new String(msg.getBody()));
                    }
                    // 默认情况下 这条消息只会被一个consumer消费到，因为是点对点消费的。
                    // ack机制
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            consumer.start();
            System.out.println("Consumer start。。。。。。");
        }
    }

附：[apache rocketmq demo文档][apache rocketmq demo]

[apache rocketmq demo]: https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md