• springboot+rocketmq（6）:实现消息过滤

    消息过滤是指消费者一端在消费消息时，对消息进行选择性过滤，只消费符合过滤条件的消息。

    一、RocketMQ消息过滤机制

    RocketMQ的消息过滤机制大致分为两种：标签过滤和类过滤。其中标签过滤又分为Tag过滤和SQL92过滤。

    类过滤是指支持自定义个一个过滤逻辑，消费时传到Broker，Broker根据自定义的逻辑进行消息过滤，生产中用的相对少，暂不讨论。

    在这里插入图片描述

    二、几种过滤机制介绍

    1.Tag过滤

    说明：消息发送时可以指定该消息的Tag号，消息存储到CommitLog后会异步转发构建ConsumeQueue，ConsumeQueue条目中包含消息Tag的hashcode值，消费者消费消息会从broker拉取消息，拉取时指定消息必须带有目标Tag值，那么遍历ConsumeQueue条目时会判断条目中Tag的hashcode值是否等于目标Tag的hashcode值，如果相等，说明可能是消费者感兴趣的消息。但是，因为比较的是Tag的hashcode值，所以实际的Tag可能不一样，所以消息拉取到消费者端后还要比较Tag值，才能保证是消费者正真感兴趣的消息。

    优点：仅仅根据ConsumeQueue就可以进行过滤，过滤完之后才去CommitLog拉取实际的消息内容返回给消费者，效率比较高。

    缺点：消费者端需要做进一步的比对才可以确定是否是感兴趣的消息。

    2.Sql92过滤(属性表达式)

    说明：同样在消息发送时指定消息的扩展属性，消息存储到CommitLog，消息拉取时，根据Topic从ConsumeQueue中取出消息的偏移量之后，还需要根据偏移量去CommitLog中遍历，判断某个消息的属性是否匹配属性表达式，匹配才会返回消费者。

    优点：可以设置多个自定义属性值，过滤可以做到多样化。并且属性值直接比较可以确保返回的消息都是消费者感兴趣的消息。

    缺点：仍然需要遍历Topic下的所有消息，判断是否匹配属性表达式，有一定的开销。

    3.类过滤

    说明：在Broker端可以运行一个或者多个过滤服务器FilterServer，消费者可以自定义消息过滤实现类并将其上传至FilterServer，那么消费者的拉取命令会中间经过一层FilterServer，Broker返回的消息先在FilterServer端做一次过滤之后再返回消费者。

    优点：过滤逻辑完全由消费者把控，想怎么过滤怎么过滤，只要在java类里面能实现出来就行。

    缺点：需要运行java类，过滤类太多的情况下消耗Broker性能。

    三、标签过滤应用

    1.创建Springboot项目，添加rockermq 依赖

    <!--rocketMq依赖-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.1</version>
</dependency>


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    2.配置rocketmq

    # 端口
server:
  port: 8083

# 配置 rocketmq
rocketmq:
  name-server: 127.0.0.1:9876
  #生产者
  producer:
    #生产者组名，规定在一个应用里面必须唯一
    group: group1
    #消息发送的超时时间 默认3000ms
    send-message-timeout: 3000
    #消息达到4096字节的时候，消息就会被压缩。默认 4096
    compress-message-body-threshold: 4096
    #最大的消息限制，默认为128K
    max-message-size: 4194304
    #同步消息发送失败重试次数
    retry-times-when-send-failed: 3
    #在内部发送失败时是否重试其他代理，这个参数在有多个broker时才生效
    retry-next-server: true
    #异步消息发送失败重试的次数
    retry-times-when-send-async-failed: 3

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    3 根据tag过滤消息

    tag过滤是最常用的方式。

    消息发送端只能设置一个tag，消息接收端可以设置多个tag。

    接收消息端通过’||‘设置多个tag,比如：tag1||tag2||tag3||…

    3.1 生产者

    生产端发送三个消息，tag分别是 tag1,tag2,tag3

    package com.example.springbootrocketfilter.controller;

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 消息过滤
 * @author qzz
 */
@RestController
public class RocketMQFilterCnntroller {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送带tag消息，测试根据 tag过滤消息
     */
    @RequestMapping("/testMessageSendByTag")
    public void sendMessageByTag(){
        //构造消息1
        Message msg1 = MessageBuilder.withPayload("rocket 消息过滤测试---tag1").build();
        //构造消息2
        Message msg2 = MessageBuilder.withPayload("rocket 消息过滤测试---tag2").build();
        //构造消息3
        Message msg3 = MessageBuilder.withPayload("rocket 消息过滤测试---tag3").build();

        //发送消息
        //参数一：topic   如果想添加tag,可以使用"topic:tag"的写法(消息发送端在设置消息目的地时通过topic与tag拼接方式)
        //参数二：消息内容
        rocketMQTemplate.syncSend("test-rocketmq-filter:tag1",msg1);
        rocketMQTemplate.syncSend("test-rocketmq-filter:tag2",msg2);
        rocketMQTemplate.syncSend("test-rocketmq-filter:tag3",msg3);
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41

    注意：
    ①消息发送端在设置消息目的地时通过topic与tag拼接方式
    ②消息发送端只能设置一个tag

    3.2 消费监听

    package com.example.springbootrocketfilter.config;

import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

/**
 * 消费消息---消息过滤
 * 配置RocketMQ监听
 *
 * selectorType：指明了消息选择通过tag的方法，默认值SelectorType.TAG
 * selectorExpression：指明能够接收哪些tag,多个tag通过 || 或的方法， 默认值*，代表全部
 * messageModel:指明了消息消费的模式，默认值为MessageModel.CLUSTERING（每条消息只能有一个消费者进行消费）；
 *                                      MessageModel.BROADCASTING(广播消息，所有订阅者都能收到消息)
 *
 * @author qzz
 */
@Service
@RocketMQMessageListener(consumerGroup = "test-filter",topic = "test-rocketmq-filter"
        ,messageModel = MessageModel.CLUSTERING
        ,selectorExpression = "tag1||tag2",selectorType = SelectorType.TAG)
public class RocketMQFilterListener implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt message) {

        System.out.println("消费消息："+new String(message.getBody()));
        System.out.println("消费消息："+message.getProperties());
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34

    注意：
    发送消息时设置了tag为：tag1、tag2；那接收消息端只有设置了selectorExpression值中包含了tag1或者tag2的才能接收消息。

    ### 3.3 启动服务，进行测试

    在这里插入图片描述
    发现只消费了tag1和tag2的消息，tag3消息没有被消费。测试成功。

    4 sql92过滤方式

    基于 sql92 表达式消息过滤，其实是对消息的属性运用 sql 过滤表达式进行条件匹配。

    在RocketMQ定义的语法下，可以实现一些有趣的逻辑。如下：
    在这里插入图片描述
    RocketMQ的SQL表达式语法 只定义了一些基本的语法功能。

    • 数字比较，如>，>=，<，<=，BETWEEN，=；
    • 字符比较，如：=，<>，IN；
    • IS NULL or IS NOT NULL;
    • 逻辑运算符：AND, OR, NOT;

    常量类型:
        ①数值，如：123, 3.1415；
        ②字符, 如：‘abc’, 必须使用单引号;
        ③NULL，特殊常量
        ④Boolean, TRUE or FALSE;

    4.1 生产者

    package com.example.springbootrocketfilter.controller;

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;

/**
 * 消息过滤
 * @author qzz
 */
@RestController
public class RocketMQFilterCnntroller {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 测试根据 sql92过滤消息
     */
    @RequestMapping("/testMessageSendBySql")
    public void testMessageSendBySql(){
        //构造消息1
        Message msg1 = MessageBuilder.withPayload("rocket 消息过滤测试---sql92").build();

        //设置消息的属性（header）信息
        Map<String,Object> headers = new HashMap<>();
        headers.put("a",1);

        //发送消息
        //参数一：topic   如果想添加tag,可以使用"topic:tag"的写法
        //参数二：消息内容
        rocketMQTemplate.convertAndSend("test-rocketmq-sql",msg1,headers);
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42

    说明：传递了a参数为1进行条件判断

    4.2 消费监听

    package com.example.springbootrocketfilter.config;

import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

/**
 * 消费消息---消息过滤  sql
 * 配置RocketMQ监听
 *
 * selectorType：指明了消息过滤使用SQL92方式
 * selectorExpression：指明了只能接收消息属性（header）中a=1的消息， 默认值*，代表全部
 * messageModel:指明了消息消费的模式，默认值为MessageModel.CLUSTERING（每条消息只能有一个消费者进行消费）；
 *                                      MessageModel.BROADCASTING(广播消息，所有订阅者都能收到消息)
 *
 * @author qzz
 */
@Service
@RocketMQMessageListener(consumerGroup = "test-sql",topic = "test-rocketmq-sql"
        ,messageModel = MessageModel.CLUSTERING
        ,selectorExpression = "a=1",selectorType = SelectorType.SQL92)
public class RocketMQFilterSqlListener implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt message) {

        System.out.println("消费消息："+new String(message.getBody()));
        System.out.println("消费消息："+message.getProperties());
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34

    说明：

    • selectorType：指明了消息过滤使用SQL92方式
    • selectorExpression：指明了只能接收消息属性（header）中a=1的消息， 默认值*，代表全部
    • messageModel:指明了消息消费的模式，默认值为MessageModel.CLUSTERING（每条消息只能有一个消费者进行消费）； MessageModel.BROADCASTING(广播消息，所有订阅者都能收到消息)

    4.3 启动服务测试

    启动服务报错：The broker does not support consumer to filter message by SQL92
    在这里插入图片描述
    原因：默认情况下broke没有开启对SQL语法的支持，需要修改配置

    （1）打开rocketmq服务下的broker.conf文件，添加如下配置：
    在这里插入图片描述
    （2）重启broker服务即可

    测试结果：
    在这里插入图片描述
    修改发送者里面的 消息的属性（header）信息：
    在这里插入图片描述
    重启进行测试：
    在这里插入图片描述
    没有消费消息。
    说明：只有设置的消息属性（header）满足selectorExpression属性值，才能正常消费。

    注意：

    ①sql92过滤仅对push消费方式生效。

    ②要使用SQL92标准过滤消息，首先需要开启enablePropertyFilter，该属性默认为false。

    修改\conf\broker.conf配置文件，末尾处增加enablePropertyFilter=true
    重启Broker

    ③SQL92底层使用布隆过滤器进行过滤。

  • 相关阅读:
    webpack：详解cache模块常用配置
    Golang复习
    产品外观设计有哪些表面处理工艺
    浪潮服务器安装CentOS 7 教程
    微前端基础知识
    【FISCO BCOS】十九、区块链浏览器部署
    MySQ 内存使用率高
    L9110S电机驱动——让小车动起来
    遥控器红外解码数码管显示
    Kubernetes技术与架构-网络 3
  • 原文地址：https://blog.csdn.net/qq_26383975/article/details/125600949