当前位置：首页 > news >正文

RocketMQ: 消息过滤，通信组件，服务发现

news 2025/7/2 1:52:48

消息过滤

1 ) 简单消息过滤

/*** 订阅指定topic下tags分别等于 TagA 或 TagC 或 TagD
*/consumer.subscribe("TopicTest1", "TagA || TagC || TagD");

如以上代码所示，简单消息过滤通过指定多个 Tag 来过滤消息，过滤的动作在服务器进行

2 ) 高级消息过滤

1. Broker 所在的机器会启劢多个 FilterServer 过滤迕程
1. Consumer 启劢后，会吐 FilterServer 上传一个过滤的 Java 类
1. Consumer 从 FilterServer 拉消息，FilterServer 将请求转发给 Broker，FilterServer 从 Broker 收到消息后，按照 Consumer 上传的 Java 过滤程序做过滤，过滤完成后返回给 Consumer
总结
- 使用 CPU 资源来换取网卡流量资源
- FilterServer 与 Broker 部署在同一台机器，数据通过本地回环通信，不走网卡
- 一台 Broker 部署多个 FilterServer，充分利用 CPU 资源，因为单个 JVM 难以全面利用高配的物理机 CPU 资源
- 因为过滤代码使用 Java 语言来编写，应用几乎可以做任意形式的服务器端消息过滤，例如通过 Message Header 进行过滤，甚至可以按照 Message Body 进行过滤
- 使用 Java 语言进行作为过滤表达式是一个双刃剑，方便了应用的过滤操作，但是带来了服务器端的安全风险。需要应用来保证过滤代码安全，例如在过滤程序里尽可能不做申请大内存，创建线程等操作。避免 Broker 服务器发生资源泄漏

通信组件

RocketMQ 通信组件使用了 Netty-4.0.9.Final，在之上做了简单的协议封装

1 ）网络协议

在这里插入图片描述

大端 4 个字节整数，等于 2、3、4 长度总和
大端 4 个字节整数，等于 3 的长度
使用 json 序列化数据
应用自定义二进制序列化数据

Header 格式

{"code": 0,"language": "JAVA","version": 0,"opaque": 0,"flag": 1,"remark": "hello, I am respponse /127.0.0.1:27603","extFields": {"count": "0","messageTitle": "HelloMessageTitle"}
}

Header 字段名	类型	Request	Response
code	整数	请求操作代码，请求接收方根据不同的代码做不同的操作	应答结果代码，0 表示成功，非 0 表示各种错误代码
language	字符串	请求发起方实现语言，默认JAVA	应答接收方实现语言
version	整数	请求发起方程序版本	应答接收方程序版本
opaque	整数	请求发起方在同一连接上不同的请求标识代码，多线程连接复用使用	应答方不做修改，直接返回
flag	整数	通信局的标志位	通信局的标志位
remark	字符串	传输自定义文本信息	错误详细描述信息
extFields	HashMap<String,String>	请求自定义字段	应答自定义字段