rocketmq实现

### RocketMQ 实现方式及原理

#### 1. 消息存储机制

RocketMQ 使用 `CommitLog` 文件来存储消息，这些文件是顺序写入的。这种设计使得日志追加操作非常高效，因为不需要频繁地移动磁盘读写头。为了提高性能并减少随机访问带来的开销，所有的消息都按照时间顺序被记录下来[^1]。

对于延迟消息，系统会在其写入过程中加入特定标志位以及额外的时间戳字段用于后续处理逻辑判断。当达到预定发送时刻时，再由调度器负责将其推送出去。

#### 2. 数据持久化策略

在数据持久化的层面,RocketMQ 提供了两种不同的刷盘模式——异步刷盘与同步刷盘：

- **异步刷盘**：这种方式下 Broker 不等待实际的数据已经保存到了物理介质上就会立即向生产者返回确认信息；虽然提高了吞吐量但也意味着存在一定的风险即如果此时服务器突然断电可能会丢失最近未完成刷写的少量消息。

- **同步刷盘**：相比之下更为安全可靠的选择,它确保每条消息都被确实无误地存入硬盘之后才告知客户端成功接收。尽管如此这般做法不可避免地会对整体效率造成一定负面影响因此通常仅限于那些对一致性有着极高要求的应用场景比如金融服务行业等[^3]。

#### 3. 订阅与消费模型

一旦有新的 Topic 被创建出来并且至少有一个 Producer 开始往里面投放内容以后任何注册成为 Consumer 的实例都能够监听到相应频道内的更新通知。不过有时候并不是所有发布的项目都会引起下游用户的兴趣所以允许通过设定筛选规则的方式让服务端只传递符合条件的信息给定目标群体从而减轻不必要的网络传输负担同时也简化了应用程序内部对于冗余事件流管理的工作量[^4]。

from rocketmq.client import PushConsumer

consumer = PushConsumer('your_group_name')
consumer.subscribe('TopicTest', 'TagA || TagB')  # 设置订阅过滤条件
def callback(msg):
    print(f"Received message: {msg.body.decode()}")
consumer.register_message_listener(callback)

consumer.start()