RocketMQ复杂过滤尝试

🕗 发布于 2024-07-06 18:36 rocketmq

需求

消息实体，根据实体中的一个字段，决定推给多个业务系统。

例：一个点位信息Bean，这个点位信息，设备、能源、安全都有用，那么点位信息表中有适用模块标识。

点位新增需要通知所有勾选业务系统 tag - add

点位编辑需要新增勾选业务系统标识 tag - add ，移除勾选 tag - delete ，不变 tag - update

点位删除通知所有勾选的系统 tag - delete

分析

1、MQ不支持同个消息，一下子发送到不同的topic。发多次也可以实现，但是多少有点抵触

2、那么发送到同一个topic下，让各个业务系统来取，那么必定需要去过滤，不然拿到不属于本业务系统的点位信息了，仅仅靠tag明显是不够的，服务端过滤可以采用SQL92方式

3、那么我随之就想到也可以在各个业务系统中过滤了，不是本业务系统的标识，直接返回。不执行相关逻辑。

本人更倾向于第二种实现

编码

尝试一：SQL92过滤实现

再broker.conf 新增配置 enablePropertyFilter=true

生产者

public class AddProducer {
 
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("a-group");
        producer.setNamesrvAddr("192.168.0.211:9876");
        producer.start();

        Point point0 = new Point(0, "point0");
        Point point1 = new Point(1, "point1");
        Point point2 = new Point(2, "point2");
        Point point3 = new Point(3, "point3");
        Point point4 = new Point(4, "point4");
        Point point5 = new Point(5, "point5");
        Point point6 = new Point(6, "point6");
        Point point7 = new Point(7, "point7");
        Point point8 = new Point(8, "point8");
        Point point9 = new Point(9, "point9");

        ArrayList<Point> list = new ArrayList<>();
        list.add(point0);
        list.add(point1);
        list.add(point2);
        list.add(point3);
        list.add(point4);
        list.add(point5);
        list.add(point6);
        list.add(point7);
        list.add(point8);
        list.add(point9);
 
 
 
 
        try {
 
            for (Point bean : list) {
                Message msg = new Message("UNIFIED_POINT", "add", JSONUtil.toJsonStr(bean).getBytes(RemotingHelper.DEFAULT_CHARSET));
                msg.putUserProperty("testValue",String.valueOf(bean.getId()));
                producer.send(msg,new CustomMessageQueueSelector(),bean.getId(),new CustomSendCallback());
                System.out.println(bean.getId() + " Continue execution ");
            }

            Thread.sleep(20000);

            for (Point bean : list) {
                Message msg = new Message("UNIFIED_POINT", "update", JSONUtil.toJsonStr(bean).getBytes(RemotingHelper.DEFAULT_CHARSET));
                msg.putUserProperty("testValue",String.valueOf(bean.getId()));
                producer.send(msg,new CustomMessageQueueSelector(),bean.getId(),new CustomSendCallback());
                System.out.println(bean.getId() + " Continue execution ");
            }
 
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消费者1 - 模拟业务系统1

    public static void main(String[] args) throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("calc_rules_point");
        // 指定Namesrv地址
        consumer.setNamesrvAddr("192.168.0.211:9876");
        // 订阅主题和标签
        consumer.subscribe("UNIFIED_POINT", MessageSelector.bySql("(testValue is not null and testValue between 0 and 3)"));
        // 设置Consumer第一次启动是从哪个位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setSuspendCurrentQueueTimeMillis(2000);
 
        // 注册消息监听器
        consumer.registerMessageListener(new CustomMessageListenerOrderly());
 
        // 启动消费者
        consumer.start();
        System.out.printf("Group1ConsumerTagA Started.%n");
    }

消费者2 - 业务系统2

public class consumer2 {
 
    public static void main(String[] args) throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("a_point");
        // 指定Namesrv地址
        consumer.setNamesrvAddr("192.168.0.211:9876");
        // 订阅主题和标签
        consumer.subscribe("UNIFIED_POINT", MessageSelector.bySql("(testValue is not null and testValue between 4 and 6)"));
        // 设置Consumer第一次启动是从哪个位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setSuspendCurrentQueueTimeMillis(2000);
 
        // 注册消息监听器
        consumer.registerMessageListener(new CustomMessageListenerOrderly());
 
        // 启动消费者
        consumer.start();
        System.out.printf("Group1ConsumerTagA Started.%n");
    }
}

测试结果

消费者1

消费者2

sql92 确实可以实现我的需求。

那么我把testValue换成各业务系统唯一标识，逗号拼接

把生产者调整一下

msg.putUserProperty("testValue","aaaa,bbbb,cccc");

客户端试了 in 不行

consumer.subscribe("UNIFIED_POINT", MessageSelector.bySql("(testValue is not null and 'aaaa' in (testValue))"));

consumer.subscribe("UNIFIED_POINT", MessageSelector.bySql("(testValue is not null and testValue LIKE '%aaaa%')"));

不行，只支持这些关键字

in 关键字只能变量in 特定字符集合这样表达。

思路转变一下，也不是说不能实现

msg.putUserProperty("aaaa","aaaa");
msg.putUserProperty("bbbb","bbbb");
msg.putUserProperty("cccc","cccc");

生产者 setProperty 每个业务系统一个区分开来。

消费者即可实现

consumer.subscribe("UNIFIED_POINT", MessageSelector.bySql("aaaa is not null"));

尝试二：消费端逻辑处理

修改生产者

测试的点位信息加上适用系统标识（逗号分隔）

public class AddProducer {
 
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("a-group");
        producer.setNamesrvAddr("192.168.0.211:9876");
        producer.start();

        Point point1 = new Point(1, "point1","1,2");
        Point point2 = new Point(2, "point2","1,2");
        Point point3 = new Point(3, "point3","1,4");
        Point point4 = new Point(4, "point4","1,4");
        Point point5 = new Point(5, "point5","1,5");

        ArrayList<Point> list = new ArrayList<>();
        list.add(point1);
        list.add(point2);
        list.add(point3);
        list.add(point4);
        list.add(point5);

 
 
 
        try {
 
            for (Point bean : list) {
                Message msg = new Message("UNIFIED_POINT", "add", JSONUtil.toJsonStr(bean).getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg,new CustomMessageQueueSelector(),bean.getId(),new CustomSendCallback());
                System.out.println(bean.getId() + " Continue execution ");
            }

            Thread.sleep(20000);

            for (Point bean : list) {
                Message msg = new Message("UNIFIED_POINT", "update", JSONUtil.toJsonStr(bean).getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg,new CustomMessageQueueSelector(),bean.getId(),new CustomSendCallback());
                System.out.println(bean.getId() + " Continue execution ");
            }
 
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

模拟消费者一：业务系统标识为1

public class consumer {
 
    public static void main(String[] args) throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("calc_rules_point");
        // 指定Namesrv地址
        consumer.setNamesrvAddr("192.168.0.211:9876");
        // 订阅主题和标签
        consumer.subscribe("UNIFIED_POINT", "*");
        // 设置Consumer第一次启动是从哪个位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setSuspendCurrentQueueTimeMillis(2000);
 
        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                try {
                    for (MessageExt msg : list) {
                        String data = new String(msg.getBody());
                        Point p = JSONUtil.toBean(data, Point.class);

                        List<String> ids = Stream.of(p.getApplyModuleIds().split(","))
                                .map(String::trim)
                                .collect(Collectors.toList());
                        if(!ids.contains("1")){
                            //非本业务系统 直接返回
                            return ConsumeOrderlyStatus.SUCCESS;
                        }
                        if(msg.getTags().equals("add")){
                            System.out.println("新增消费：" + p + msg.getQueueId());
                        }else if(msg.getTags().equals("update")){
                            System.out.println("修改消费：" + p + msg.getQueueId());
                        }

                    }

                    return ConsumeOrderlyStatus.SUCCESS;
                } catch (Exception e) {
                    MessageExt msg = list.get(0);
                    log.error("consumer news error " + new String(msg.getBody()));
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
        });
 
        // 启动消费者
        consumer.start();
        System.out.printf("Group1ConsumerTagA Started.%n");
    }
}

模拟消费者二：业务系统标识2

public class consumer2 {
 
    public static void main(String[] args) throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("a_point");
        // 指定Namesrv地址
        consumer.setNamesrvAddr("192.168.0.211:9876");
        // 订阅主题和标签
        consumer.subscribe("UNIFIED_POINT", "*");
        // 设置Consumer第一次启动是从哪个位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setSuspendCurrentQueueTimeMillis(2000);
 
        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                try {
                    for (MessageExt msg : list) {
                        String data = new String(msg.getBody());
                        Point p = JSONUtil.toBean(data, Point.class);

                        List<String> ids = Stream.of(p.getApplyModuleIds().split(","))
                                .map(String::trim)
                                .collect(Collectors.toList());
                        if(!ids.contains("2")){
                            //非本业务系统 直接返回
                            return ConsumeOrderlyStatus.SUCCESS;
                        }
                        if(msg.getTags().equals("add")){
                            System.out.println("新增消费：" + p + msg.getQueueId());
                        }else if(msg.getTags().equals("update")){
                            System.out.println("修改消费：" + p + msg.getQueueId());
                        }

                    }

                    return ConsumeOrderlyStatus.SUCCESS;
                } catch (Exception e) {
                    MessageExt msg = list.get(0);
                    log.error("consumer news error " + new String(msg.getBody()));
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
        });
 
        // 启动消费者
        consumer.start();
        System.out.printf("Group1ConsumerTagA Started.%n");
    }
}

测试结果 - 消费者1

测试结果 - 消费者2

结论

SQL92方式：过滤在服务端，但是功能还是局限的。如果服务器性能好压力不大，且过滤方式能满足。个人任务还是可用的

消费端逻辑处理方式：略费带宽，服务器压力小。过滤在消费端

原文地址：https://blog.csdn.net/zjy660358/article/details/140158245

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

多个点安装到点b的距离从小到大排序
多个点安装到点b的距离从小到大排序
阅读更多2024-09-23
UE5学习笔记22-武器瞄准和武器自动开火
绘制武器准心，自动开火，瞄准武器时缩放视场角，复制运动的通知函数，瞄准时准心变色
阅读更多2024-09-23
面试题（二）
综上所述，选择最优的通信机制要考虑具体的应用场景、性能需求、同步要求和实现复杂度等因素。没有绝对的“最优”方案，只有适合的解决方案。在不同的应用场景中，通信机制的“最优”取决于具体需求和环境。（Int
阅读更多2024-09-23
2024/9/21 leetcode 21.合并两个有序链表 2.两数相加
将两个升序链表合并为一个新的链表并返回。新链表是通过拼接给定的两个链表的所有节点组成的。[][0]l2。
阅读更多2024-09-23
浅谈Linux中文件与目录的ACL
setfacl是 Linux 系统中用于设置文件和目录的 ACL（访问控制列表）的命令。它允许为特定的用户或用户组指定访问权限，提供比传统的文件权限机制更灵活的权限控制。下面是setfacl的常用选项
阅读更多2024-09-23
Maven-五、属性
使用maven中的属性可以来简化我们的配置工作。在pom文件中可以使用**标签设置属性，属性名自定义,然后可以在使用${属性名}**文件中引用自己定义的属性。使用属性进行配置，在后来修改相关配置时直接
阅读更多2024-09-23
FreeRTOS下UART的封装
FreeRTOS下UART的封装_哔哩哔哩_bilibili
阅读更多2024-09-23
find命令详解
find命令用于在指定的目录及其子目录中查找符合条件的文件和目录。它支持复杂的搜索条件组合，并能对搜索结果执行各种操作，是系统管理员和开发人员日常工作中不可或缺的工具。
阅读更多2024-09-23
网络层协议 —— IP协议
对于网络层IP协议的学习，我们不能局限于IP协议本身，而应通过学习IP协议建立对整个网络的宏观认识。不同主机进行网络通信时，需要能够找到对方，网络层的IP协议便提供了这种能力，主要依靠IP地址。网络的
阅读更多2024-09-23
C++ Linux IPC进程通信-消息队列MQ
相比于共享内存和管道,消息队列能够实现指定的消息格式和排序,能实现更复杂的通信。MQ的复杂度较高,维护成本增加.如果MQ挂了,主服务直接挂了.
阅读更多2024-09-23

RocketMQ复杂过滤尝试

需求

分析

编码

尝试一：SQL92过滤实现

尝试二：消费端逻辑处理

结论

相关文章