RocketMQ的消息类型及简单代码示例

爱被打了一巴掌 2022-11-09 11:20 165阅读 0赞

## 1. MQ简介 ##

#### 1.1 简介 ####

**MQ（Message Queue）消息队列**，是基础数据结构中“**先进先出**”的一种数据结构。一般用来解决应用解耦，异步消息，流量削峰等问题，实现高性能，高可用，可伸缩和最终一致性架构。

把要传输的数据（消息）放在队列中，用队列机制来实现消息传递——生产者产生消息并把消息放入队列，然后由消费者去处理。消费者可以到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给其推送消息。

#### 1.2 作用 ####

1.  **异步**：主业务执行结束后从属业务通过MQ，异步执行，减低业务的响应时间，提高用户体验。
2.  **解耦**：一个业务需要多个模块共同实现，或者一条消息有多个系统需要对应处理，只需要主业务完成以后，发送一条MQ，其余模块消费MQ消息，即可实现业务，降低模块之间的耦合。
3.  **削峰**：高并发情况下，业务异步处理，提供高峰期业务处理能力，避免系统瘫痪。

#### 1.3 缺点 ####

1.  系统可用性降低：如何保证MQ的高可用
2.  系统复杂度提高：如何保证消费不会丢失？不会被重复调用？怎么保证消息的顺序性等问题
3.  消息一致性问题：如何保证主业务和从属业务一致性的处理

## 2. 常见MQ应用的比较 ##

<table> 
 <thead> 
  <tr> 
   <th>—</th> 
   <th>优点</th> 
   <th>缺点</th> 
   <th>使用场景</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>RabbitMQ</td> 
   <td>消息可靠性高、功能全面</td> 
   <td>吞吐量低、消息积累对性能影响较大</td> 
   <td>小规模场景</td> 
  </tr> 
  <tr> 
   <td>RocketMQ</td> 
   <td>高吞吐、高性能、功能全面</td> 
   <td>开源版（阉割版）、只支持Java</td> 
   <td>大部分场景</td> 
  </tr> 
  <tr> 
   <td>Kafka</td> 
   <td>高吞吐、高性能、高可用</td> 
   <td>会丢数据、功能比较单一</td> 
   <td>日志分析、大数据采集分析</td> 
  </tr> 
 </tbody> 
</table>

## 3. RocketMQ的编程模型 ##

#### 3.1 生产者 ####

1.  创建消息生产者producer，并指定生产者组名
2.  指定Name server地址
3.  启动producer
4.  创建消息对象，指定主题Topic、Tag和消息体
5.  发送消息
6.  关闭生产者producer

#### 3.2 消费者 ####

1.  创建消费者Consumer，指定消费者组名
2.  指定Name server地址
3.  订阅主题Topic和Tag
4.  设置回调函数，处理消息
5.  启动消费者consumer

## 4. RocketMQ的消息类型 ##

#### 4.1 基本类型（同步发送、异步发送、单向发送） ####

消费者：

package com.example.mq.synchronous;
    
    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.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    /** * 描述: * 消费者 * * @author XueGuCheng * @create 2021-03-14 21:27 */
    public class Consumer { 
    
        public static void main(String[] args) throws MQClientException { 
            // 1.创建消费者Consumer，指定消费者组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
            // 2.指定Name server地址
            consumer.setNamesrvAddr("127.0.0.1:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    
            // 3.订阅主题Topic和Tag
            consumer.subscribe("TopicTest","*");
            // 4.设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) { 
                    System.out.printf("%s 收到的消息: %s %n", Thread.currentThread().getName(), list);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 5.启动消费者consumer
            consumer.start();
            System.out.printf("消费者开始运行.%n");
    
    
        }
    
    }

##### 4.1.1 同步发送 #####

**原理：**

同步发送是指消息发送方发出一条消息后，会在收到服务端返回响应之后才发下一条消息的通讯方式

应用场景：如重要通知邮件、报名短信通知等。

生产者：

package com.example.mq.synchronous;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    
    
    /** * 描述: * 同步发送消息 * * @author XueGuCheng * @create 2021-03-14 21:09 */
    public class SynchronousProducer { 
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            for (int i = 0; i < 10; i++) { 
                Message message = new Message("TopicTest", "Tag_Synchronous", "ID100",  "Synchronous".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 5.同步发送消息，消息会发给集群中的一个Broker节点
                SendResult sendResult = producer.send(message);
            }
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    }

##### 4.1.2 异步发送 #####

**原理：**

异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。RocketMQ异步发送，需要实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务端响应即可发送第二条消息。发送方通过回调接口接收服务端响应，并处理响应结果。

应用场景：一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如，视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

生产者：

package com.example.mq.Async;
    
    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 org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    /** * 描述: * 异步发送 * * @author XueGuCheng * @create 2021-03-14 21:56 */
    public class AsyncProducer { 
    
        //异步发送，需要引入一个countDownLatch，来保证所有Producer发送消息的回调方法都执行完了再停止Producer服务
        static final CountDownLatch countDownLatch = new CountDownLatch(10);
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            for (int i = 0; i < 10; i++) { 
                final int index = i;
                Message message = new Message("TopicTest", "Tag_Async", "ID100",  "Async".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 5.异步发送消息，消息会发给集群中的一个Broker节点
                producer.send(message, new SendCallback() { 
                    // 执行成功的回调
                    @Override
                    public void onSuccess(SendResult sendResult) { 
                        countDownLatch.countDown();
                        System.out.printf("%-10d true %s %n", index, sendResult.getMsgId());
                    }
                    // 执行失败的回调
                    @Override
                    public void onException(Throwable throwable) { 
                        countDownLatch.countDown();
                        System.out.printf("%-10d flase %s %n", index, throwable);
                        throwable.printStackTrace();
                    }
                });
            }
            System.out.println("消息发送完成");
            countDownLatch.await(5, TimeUnit.SECONDS);
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    
    }

##### 4.1.3 单向发送 #####

原理：

单向（Oneway）发送特点为只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。

应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。

生产者：

package com.example.mq.ida;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    /** * 描述: * 单向发送 * * @author XueGuCheng * @create 2021-03-14 22:08 */
    public class IDAProducer { 
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            for (int i = 0; i < 10; i++) { 
                Message message = new Message("TopicTest", "Tag_IDA", "ID100",  "IDA".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 5.单向发送消息，消息会发给集群中的一个Broker节点
                producer.sendOneway(message);
            }
            Thread.sleep(5000);
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    
    
    }

##### 4.1.4 总结 #####

<table> 
 <thead> 
  <tr> 
   <th>发送方式</th> 
   <th>发送 TPS</th> 
   <th>发送结果反馈</th> 
   <th>可靠性</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>同步发送</td> 
   <td>快</td> 
   <td>有</td> 
   <td>不丢失</td> 
  </tr> 
  <tr> 
   <td>异步发送</td> 
   <td>快</td> 
   <td>有</td> 
   <td>不丢失</td> 
  </tr> 
  <tr> 
   <td>单向发送</td> 
   <td>最快</td> 
   <td>无</td> 
   <td>可能丢失</td> 
  </tr> 
 </tbody> 
</table>

#### 4.2 顺序消息 ####

**原理：**

发送者端：  
在默认情况下，消息发送者会采取Round Robin轮询方式把消息发送到不同的MessageQueue(分区队列)，而消费者消费的时候也从多个MessageQueue上拉取消息，这种情况下消息是不能保证顺序的。而只有当一组有序的消息发送到同一个MessageQueue上时，才能利用MessageQueue先进先出的特性保证这一组消息有序。

消费者端：  
消费者会从多个消息队列上去拿消息。这时虽然每个消息队列上的消息是有序的，但是多个队列之间的消息仍然是乱序的。消费者端要保证消息有序，就需要按队列一个一个来取消息，即取完一个队列的消息后，再去取下一个队列的消息。

而给consumer注入的MessageListenerOrderly对象，在RocketMQ内部就会通过锁队列的方式保证消息是一个一个队列来取的。MessageListenerConcurrently这个消息监听器则不会锁队列，每次都是从多个Message中取一批数据（默认不超过32条）。因此也无法保证消息有序。

生产者：

package com.example.mq.order;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageQueue;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.util.List;
    
    /** * 描述: * 顺序消息——生产者 * * @author XueGuCheng * @create 2021-03-14 22:21 */
    public class OrderProducer { 
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            // 顺序消息：保证第二个循坏内的消息的顺序性，不保证第一个循坏的顺序性。即只保证局部有序，不保证全局有序
            for (int i = 0; i < 10; i++) { 
                int orderId = i;
                for (int i1 = 0; i1 < 5; i1++) { 
                    Message message = new Message("TopicTest", "Tag_Order" + orderId, "ID100" + orderId, ("Order" + i1).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // 5.同步发送消息，消息会发给集群中的一个Broker节点
                    SendResult sendResult = producer.send(message, new MessageQueueSelector() { 
                        @Override
                        public MessageQueue select(List<MessageQueue> list, Message message, Object o) { 
                            Integer id = (Integer) o;
                            int index = id % list.size();
                            return list.get(index);
                        }
                    }, orderId);
    
                    System.out.printf("%s%n", sendResult);
                }
            }
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    
    
    }

消费者：

package com.example.mq.order;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.*;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    /** * 描述: * 消费者 * * @author XueGuCheng * @create 2021-03-14 21:27 */
    public class Consumer { 
    
        public static void main(String[] args) throws MQClientException { 
            // 1.创建消费者Consumer，指定消费者组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
            // 2.指定Name server地址
            consumer.setNamesrvAddr("127.0.0.1:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    
            // 3.订阅主题Topic和Tag
            consumer.subscribe("TopicTest","*");
            // 4.设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerOrderly() { 
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) { 
                    context.setAutoCommit(true);
                    for (MessageExt message : list) { 
                        System.out.println("收到消息内容 "+new String(message.getBody()));
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            // 5.启动消费者consumer
            consumer.start();
            System.out.printf("消费者开始运行.%n");
    
    
        }
    
    }

#### 4.3 广播消息 ####

原理：

在集群状态(MessageModel.CLUSTERING)下，每一条消息只会被同一个消费者组中的一个实例消费到(这跟kafka和rabbitMQ的集群模式是一样的)。而广播模式则是把消息发给了所有订阅了对应主题的消费者，而不管消费者是不是同一个消费者组。

消费者：

package com.example.mq.broadcast;
    
    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.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.util.List;
    
    /** * 描述: * 广播模式 * * @author XueGuCheng * @create 2021-03-14 22:47 */
    public class PushConsumer { 
    
        public static void main(String[] args) throws MQClientException { 
            // 1.创建消费者Consumer，指定消费者组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
            // 2.指定Name server地址
            consumer.setNamesrvAddr("127.0.0.1:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    
            consumer.setMessageModel(MessageModel.BROADCASTING);
            
            // 3.订阅主题Topic和Tag
            consumer.subscribe("TopicTest","*");
            // 4.设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) { 
                    System.out.printf("%s 收到的消息: %s %n", Thread.currentThread().getName(), list);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 5.启动消费者consumer
            consumer.start();
            System.out.printf("消费者开始运行.%n");
        }
    }

#### 4.4 延迟消息 ####

延迟级别：

1到18分别对应messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

生产者：

package com.example.mq.scheduled;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    /** * 描述: * 延时消息 * * @author XueGuCheng * @create 2021-03-14 22:52 */
    public class ScheduledProducer { 
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            for (int i = 0; i < 10; i++) { 
                Message message = new Message("TopicTest", "Tag_Synchronous", "ID100",  "Synchronous".getBytes(RemotingHelper.DEFAULT_CHARSET));
                //设置延时级别
                message.setDelayTimeLevel(3);
                // 5.发送消息，消息会发给集群中的一个Broker节点
                producer.send(message);
            }
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    
    }

#### 4.5 批量消息 ####

批量消息是指将多条消息合并成一个批量消息，一次发送出去。这样的好处是可以减少网络IO，提升吞吐量。

注意：推荐一个批次消息的大小不要超过1MB，实际最大的限制是4194304字节，大概4MB。而且批量消息的使用是有一定限制的，这些消息应该有相同的Topic，相同的waitStoreMsgOK。而且不能是延迟消息、事务消息等。

生产者：

package com.example.mq.batch;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /** * 描述: * 批量消息 * * @author XueGuCheng * @create 2021-03-14 23:04 */
    public class BatchProducer { 
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            List<Message> messages = new ArrayList<>(16);
            for (int i = 0; i <  1000; i++) { 
                messages.add(new Message("TopicTest", "Tag_Batch", "ID100" + i,  "Batch".getBytes()));
            }
            // 5.发送消息，消息会发给集群中的一个Broker节点
            producer.send(messages);
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    
    }

#### 4.6 过滤消息 ####

即，使用Message的Tag属性来简单快速的过滤信息。

原理：

在消费者端使用MessageSelector.bySql(String sql)返回的一个MessageSelector。这里面的sql语句是按照SQL92标准来执行的。sql中可以使用的参数有默认的TAGS和一个在生产者中加入的a属性。

SQL92语法：

*  数值比较，比如：>，>=，<，<=，BETWEEN，=
 *  字符比较，比如：=，<>，IN
 *  IS NULL 或者 IS NOT NULL
 *  逻辑符号 AND，OR，NOT

常量支持类型为：

*  数值，比如：123，111222
 *  字符，比如：‘abc’，必须用单引号包裹起来
 *  NULL，特殊的常量
 *  布尔值，TRUE 或 FALSE

**注意：只有推模式的消费者可以使用SQL过滤。拉模式是用不了的。**

生产者：

package com.example.mq.filter;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    /** * 描述: * 过滤模式 * * @author XueGuCheng * @create 2021-03-14 23:18 */
    public class FilterProducer { 
    
        public static void main(String[] args) throws Exception { 
            // 1.创建消息生产者producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
            // 2.指定Name server地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动producer
            producer.start();
    
            String[] tags = new String[] { "TagA", "TagB", "TagC"};
    
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            for (int i = 0; i < 10; i++) { 
                Message message = new Message("TopicTest", tags[i % tags.length], "ID100",  "Synchronous".getBytes(RemotingHelper.DEFAULT_CHARSET));
                message.putUserProperty("a",String.valueOf(i));
                // 5.发送消息，消息会发给集群中的一个Broker节点
                SendResult sendResult = producer.send(message);
                System.out.printf("%s%n", sendResult);
            }
            // 6.关闭生产者producer
            producer.shutdown();
    
        }
    
    }

消费者：

package com.example.mq.filter;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.MessageSelector;
    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.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    /** * 描述: * 过滤模式 * * @author XueGuCheng * @create 2021-03-14 23:21 */
    public class FilterConsumer { 
    
        public static void main(String[] args) throws MQClientException { 
            // 1.创建消费者Consumer，指定消费者组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
            // 2.指定Name server地址
            consumer.setNamesrvAddr("127.0.0.1:9876");
            // 3.设置过滤主题Topic和Tag（TAGS不为空，且是A或者B，并且a属性不为空，且在0和3之间）
            consumer.subscribe("TopicTest", MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
                    "and (a is not null and a between 0 and 3)"));
            //consumer.subscribe("TagFilterTest", "TagA || TagC");
            // 4.设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) { 
                    System.out.printf("%s 收到的消息: %s %n", Thread.currentThread().getName(), list);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 5.启动消费者consumer
            consumer.start();
            System.out.printf("消费者开始运行.%n");
        }
    
    }

## 5. ACL权限控制 ##

#### 5.1 简介 ####

ACL权限控制，提供Topic资源级别的用户访问控制。

即，可以在Client客户端通过 RPCHook注入AccessKey和SecretKey签名，同时将对应的权限控制属性（包括Topic访问权限、IP白名单和AccessKey和SecretKey签名等）配置在$ROCKETMQ\_HOME/conf/plain\_acl.yml的配置文件中。

Broker端会对AccessKey所拥有的权限进行校验，校验不过，则抛出异常

#### 5.2 maven坐标 ####

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

#### 5.3 配置信息 ####

plan\_acl.yml的配置信息

#全局白名单，不受ACL控制，通常需要将主从架构中的所有节点加进来 
    globalWhiteRemoteAddresses:
    - 127.0.0.* 
    
    accounts: 
    #第一个账户 
    - accessKey: RocketMQ 
      secretKey: 12345678 
      whiteRemoteAddress: 
      admin: false 
      defaultTopicPerm: DENY #默认Topic访问策略是拒绝 
      defaultGroupPerm: SUB #默认Group访问策略是只允许订阅 
      topicPerms: 
      - topicA=DENY #topicA拒绝 
      - topicB=PUB|SUB #topicB允许发布和订阅消息
      - topicC=SUB #topicC只允许订阅 
      groupPerms: 
      - groupA=DENY 
      - groupB=PUB|SUB 
      - groupC=SUB 
    #第二个账户 
    - accessKey: rocketmq2 
      secretKey: 12345678 
      ......