五分钟带你玩转rocketMQ（七）吐血总结延时队列，批处理，条件过滤与日志配置

一.延时队列

定时队列 -它们要在规定的时间之后才能传递

1.修改调用类即可

      @RestController
      @RequestMapping("/test")
      public class TestControllor {
         private static final Logger logger = LoggerFactory.getLogger(TestControllor.class);
          /**
           * 使用RocketMq的生产者
           */
         @Resource(name = "customRocketMQProducer")
         private DefaultMQProducer defaultMQProducer;
         @RequestMapping("/send")
         public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
             for (int i = 0; i < 100; i++) {
                 final int index = i;
                  String msg = "demo msg test";
                  logger.info("开始发送消息：" + msg);
                  Message sendMsg = new Message("DemoTopic", "TagA", String.valueOf(i), msg.getBytes());
                 //预设值的延迟时间间隔为：1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h
                 //设置延时时间 3即为10s
                  sendMsg.setDelayTimeLevel(3);
                 //默认3秒超时
                  SendResult sendResult = defaultMQProducer.send(sendMsg);
                  logger.info("消息发送响应信息：" + sendResult.toString() + "当前为第" + i + "次");
              }
          }
      }
  
 

二.批处理

成批发送消息提高了传递小消息的性能。

使用限制

同一批的消息应该有：相同的主题，相同的waitStoreMsgOK，并且不支持调度。

此外，一批消息的总大小不应超过1MiB。

如何使用批处理

如果一次只发送不超过1MiB的消息，则很容易使用批处理：

      @RestController
      @RequestMapping("/test")
      public class TestControllor {
         private static final Logger logger = LoggerFactory.getLogger(TestControllor.class);
          /**
           * 使用RocketMq的生产者
           */
         @Resource(name = "customRocketMQProducer")
         private DefaultMQProducer defaultMQProducer;
         @RequestMapping("/send")
         public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
             //for (int i = 0; i < 100; i++) {
             //final int index = i;
              String msg = "demo msg test";
              logger.info("开始发送消息：" + msg);
              List<Message> messages = new ArrayList<>();
              messages.add(new Message("DemoTopic", "TagA", "OrderID001", "Hello world 0".getBytes()));
              messages.add(new Message("DemoTopic", "TagA", "OrderID002", "Hello world 1".getBytes()));
              messages.add(new Message("DemoTopic", "TagA", "OrderID003", "Hello world 2".getBytes()));
             //默认3秒超时
              SendResult sendResult = defaultMQProducer.send(messages);
          }
      }
  
 

条件过滤

在使用时需要开启条件过滤启动配置 否则会报错

The broker does not support consumer to filter message by SQL92

修改broker.conf文件 加入

enablePropertyFilter = true

 

同时启动broker 命令为

The broker does not support consumer to filter message by SQL92

 

筛选固定条件消息

      数值比较：>, >=, <, <=, BETWEEN, =;
      字符比较  =, <>, IN;
      IS NULL or IS NOT NULL;
      逻辑：AND, OR, NOT;
  
 

使用限制：

只有push consumer可以通过SQL92选择消息。接口是：

public void subscribe(final String topic, final MessageSelector messageSelector)

 

代码示例

调用方法

      @RestController
      @RequestMapping("/test")
      public class TestControllor {
         private static final Logger logger = LoggerFactory.getLogger(TestControllor.class);
          /**
           * 使用RocketMq的生产者
           */
         @Resource(name = "customRocketMQProducer")
         private DefaultMQProducer defaultMQProducer;
         @RequestMapping("/send")
         public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
             for (int i = 0; i < 100; i++) {
                 final int index = i;
                  String msg = "demo msg test";
                  logger.info("开始发送消息：" + msg);
                  Message sendMsg = new Message("DemoTopic",
                         "DemoTag",
                          ("Hello RocketMQ " + i).getBytes()
                  );
                 //消费者根据a进行过滤
                  sendMsg.putUserProperty("a", String.valueOf(i));
                  SendResult sendResult = defaultMQProducer.send(sendMsg);
                  logger.info("消息发送响应信息：" + sendResult.toString() + "当前为第" + i + "次");
              }
          }
      }
  
 

消费者

      @SpringBootConfiguration
      public class MQConsumerConfiguration {
         public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfiguration.class);
         @Value("${rocketmq.consumer.namesrvAddr}")
         private String namesrvAddr;
         @Value("${rocketmq.consumer.groupName}")
         private String groupName;
         @Value("${rocketmq.consumer.consumeThreadMin}")
         private int consumeThreadMin;
         @Value("${rocketmq.consumer.consumeThreadMax}")
         private int consumeThreadMax;
         @Value("${rocketmq.consumer.topics}")
         private String topics;
         @Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
         private int consumeMessageBatchMaxSize;
         @Autowired
         private MQConsumeMsgListenerProcessor mqMessageListenerProcessor;
         @Bean
         public DefaultMQPushConsumer testRocketMQConsumer() throws Exception {
             if (StringUtils.isEmpty(groupName)){
                 throw new Exception("groupName is null !!!");
              }
             if (StringUtils.isEmpty(namesrvAddr)){
                 throw new Exception("namesrvAddr is null !!!");
              }
             if(StringUtils.isEmpty(topics)){
                 throw new Exception("topics is null !!!");
              }
              DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
              consumer.setNamesrvAddr(namesrvAddr);
              consumer.setConsumeThreadMin(consumeThreadMin);
              consumer.setConsumeThreadMax(consumeThreadMax);
              consumer.registerMessageListener(mqMessageListenerProcessor);
             /**
       * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
       * 如果非第一次启动，那么按照上次消费的位置继续消费
       */
              consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
             /**
       * 设置消费模型，集群还是广播，默认为集群
       */
              consumer.setMessageModel(MessageModel.CLUSTERING);
             /**
       * 设置一次消费消息的条数，默认为1条
       */
              consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
             try {
                 //根据条件过滤消息
                  consumer.subscribe("DemoTopic", MessageSelector.bySql("a between 0 and 3"));
                  consumer.start();
                  LOGGER.info("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr);
              }catch (MQClientException e){
                  LOGGER.error("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr,e);
                 throw new Exception(e);
              }
             return consumer;
          }
      }
  
 

日志配置

      -Drocketmq.client.logRoot=E:\logs -Drocketmq.client.logLevel=ALL
      -Drocketmq.client.logRoot=E:\logs -Drocketmq.client.logLevel=ERROR
  
 

文章来源: baocl.blog.csdn.net，作者：小黄鸡1992，版权归原作者所有，如需转载，请联系作者。

原文链接：baocl.blog.csdn.net/article/details/103426858