前五讲我们构建了一个高可用的消息队列——消息可以可靠地生产、存储、消费即使节点宕机也能自动恢复。但在很多业务场景中仅仅不丢消息还不够订单系统创建订单 → 支付 → 发货这些消息必须严格按顺序处理金融交易扣款和入账要么都成功要么都失败这一讲我们来实现顺序消息和事务消息这两个高级特性。一、顺序消息1.1 什么是顺序消息❌ 乱序消费普通消息 Producer: 订单创建 → 支付成功 → 订单取消 Consumer: 支付成功 → 订单创建 → 订单取消 ✗ 逻辑错误 ✅ 顺序消费顺序消息 Producer: 订单创建 → 支付成功 → 订单取消 Consumer: 订单创建 → 支付成功 → 订单取消 ✓ 正确顺序1.2 全局顺序 vs 分区顺序全局顺序 ┌─────────────────────────────────────────┐ │ Topic 只有一个分区 │ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │ Msg 1│ │ Msg 2│ │ Msg 3│ │ Msg 4│ │ │ └──────┘ └──────┘ └──────┘ └──────┘ │ │ 所有消息严格按 FIFO 顺序 │ └─────────────────────────────────────────┘ 分区顺序 ┌─────────────────────────────────────────┐ │ Topic 有多个分区每个分区内部有序 │ │ ┌──────────────────┐ │ │ │ Partition 0 │ │ │ │ 订单A: 创建→支付 │ │ │ └──────────────────┘ │ │ ┌──────────────────┐ │ │ │ Partition 1 │ │ │ │ 订单B: 创建→支付 │ │ │ └──────────────────┘ │ │ 相同 Key 的消息进入同一分区 │ └─────────────────────────────────────────┘1.3 顺序消息实现# mq/ordered/ordered_producer.py import threading import time import logging from typing import Dict, Optional, Callable from ..protocol.message import Message, MessageType from ..transport.server import TCPClient from ..producer.batch import ProducerRecord logger logging.getLogger(__name__) class OrderedProducer: 顺序消息生产者 保证相同 Key 的消息按发送顺序到达 Broker def __init__(self, brokers: list, max_inflight: int 1): Args: brokers: Broker 地址列表 max_inflight: 每个分区允许的未确认消息数顺序消息必须为1 self.brokers brokers self.max_inflight max_inflight # 每个分区的 inflight 计数 self.inflight_counts: Dict[str, Dict[int, int]] {} self.lock threading.Lock() # 等待队列按分区排队 self.wait_queues: Dict[str, Dict[int, list]] {} # 网络连接 self.client TCPClient(*brokers[0]) self.client.connect() def send_sync(self, topic: str, value: str, key: str ) - Optional[int]: 同步发送顺序消息 保证相同 key 的消息按顺序发送 # 计算分区使用 key 哈希确保相同 key 进入同一分区 partition self._hash_partition(key, 3) # 等待前面的消息完成 self._wait_for_completion(topic, partition) # 增加 inflight 计数 with self.lock: if topic not in self.inflight_counts: self.inflight_counts[topic] {} self.inflight_counts[topic][partition] \ self.inflight_counts[topic].get(partition, 0) 1 try: # 发送消息 import json payload json.dumps({ topic: topic, partition: partition, messages: [{key: key, value: value}] }) msg Message( msg_typeMessageType.PRODUCE_REQUEST, topictopic, valuepayload.encode() ) response self.client.send(msg) if response: result json.loads(response.value.decode()) if result.get(success): return result.get(offsets, [-1])[0] else: raise Exception(result.get(error, Unknown error)) finally: # 减少 inflight 计数 with self.lock: self.inflight_counts[topic][partition] - 1 # 唤醒等待的线程 if topic in self.wait_queues and partition in self.wait_queues[topic]: if self.wait_queues[topic][partition]: self.wait_queues[topic][partition].pop(0) return None def _wait_for_completion(self, topic: str, partition: int): 等待前面的消息完成 with self.lock: if topic not in self.wait_queues: self.wait_queues[topic] {} if partition not in self.wait_queues[topic]: self.wait_queues[topic][partition] [] current_count self.inflight_counts.get(topic, {}).get(partition, 0) if current_count self.max_inflight: # 加入等待队列 event threading.Event() self.wait_queues[topic][partition].append(event) # 释放锁等待被唤醒 self.lock.release() event.wait(timeout30) self.lock.acquire() def _hash_partition(self, key: str, num_partitions: int) - int: 哈希分区 if not key: return 0 return abs(hash(key)) % num_partitions def close(self): 关闭 self.client.disconnect() class OrderedConsumer: 顺序消息消费者 保证消息按顺序投递给业务处理器 def __init__(self, brokers: list, group_id: str): self.brokers brokers self.group_id group_id # 每个分区的处理队列 self.processing_queues: Dict[tuple, list] {} self.lock threading.Lock() # 业务处理器 self.handler: Optional[Callable] None # 网络连接 self.client TCPClient(*brokers[0]) self.client.connect() def process_ordered(self, messages: list): 顺序处理消息 确保同一分区的消息串行处理 for msg in messages: topic msg.get(topic, ) partition msg.get(partition, 0) key (topic, partition) with self.lock: if key not in self.processing_queues: self.processing_queues[key] [] self.processing_queues[key].append(msg) # 串行处理 self._process_next(topic, partition) def _process_next(self, topic: str, partition: int): 处理下一条消息 key (topic, partition) with self.lock: if key not in self.processing_queues or not self.processing_queues[key]: return msg self.processing_queues[key].pop(0) try: if self.handler: self.handler(msg) except Exception as e: logger.error(fProcess message error: {e}) # 失败时停止处理后续消息 return # 处理下一条 self._process_next(topic, partition)二、事务消息2.1 事务消息模型事务消息流程 1. Producer 发送 Half Message半消息 │ ▼ 2. Broker 存储 Half Message不可见 │ ▼ 3. Producer 执行本地事务 │ ├──▶ 事务成功 → Commit → 消息可见 │ └──▶ 事务失败 → Rollback → 消息删除 │ ▼ 4. Broker 回查如果 Producer 长时间未决 │ ├──▶ 回查成功 → Commit │ └──▶ 回查失败 → Rollback2.2 事务状态管理# mq/transaction/transaction.py import json import os import threading import time import logging from typing import Dict, Optional, Callable from enum import Enum, auto from dataclasses import dataclass, field logger logging.getLogger(__name__) class TransactionStatus(Enum): 事务状态 PREPARING auto() # 准备中Half Message 已存储 COMMITTED auto() # 已提交 ROLLBACK auto() # 已回滚 UNKNOWN auto() # 未知需要回查 dataclass class Transaction: 事务对象 transaction_id: str topic: str partition: int messages: list status: TransactionStatus TransactionStatus.PREPARING created_at: float 0.0 updated_at: float 0.0 check_times: int 0 def __post_init__(self): now time.time() if not self.created_at: self.created_at now if not self.updated_at: self.updated_at now class TransactionManager: 事务管理器 管理事务的生命周期 1. 创建 Half Message 2. 等待 Commit/Rollback 3. 超时回查 def __init__(self, storage, transaction_timeout_ms: int 60000, check_interval_ms: int 10000): self.storage storage self.transaction_timeout_ms transaction_timeout_ms self.check_interval_ms check_interval_ms # transaction_id - Transaction self.transactions: Dict[str, Transaction] {} self.lock threading.Lock() # 回查回调 self.check_callbacks: Dict[str, Callable] {} # 持久化 self.state_file ./transaction_state.json self._load_state() # 启动回查线程 self.running True self.check_thread threading.Thread(targetself._check_loop, daemonTrue) self.check_thread.start() def prepare(self, transaction_id: str, topic: str, partition: int, messages: list) - bool: 准备事务存储 Half Message Half Message 对消费者不可见 with self.lock: tx Transaction( transaction_idtransaction_id, topictopic, partitionpartition, messagesmessages, statusTransactionStatus.PREPARING ) self.transactions[transaction_id] tx # 存储 Half Message标记为不可见 for msg in messages: value msg.get(value, ).encode() key msg.get(key, ) # 在消息头中添加事务标记 tagged_value self._tag_half_message(value, transaction_id) self.storage.produce(topic, partition, tagged_value, key) self._save_state() logger.info(fTransaction prepared: {transaction_id}) return True def commit(self, transaction_id: str) - bool: 提交事务 使 Half Message 对消费者可见 with self.lock: tx self.transactions.get(transaction_id) if not tx or tx.status ! TransactionStatus.PREPARING: return False tx.status TransactionStatus.COMMITTED tx.updated_at time.time() # 更新消息标记为可见 self._make_messages_visible(tx) self._save_state() logger.info(fTransaction committed: {transaction_id}) return True def rollback(self, transaction_id: str) - bool: 回滚事务 删除 Half Message with self.lock: tx self.transactions.get(transaction_id) if not tx or tx.status ! TransactionStatus.PREPARING: return False tx.status TransactionStatus.ROLLBACK tx.updated_at time.time() # 删除 Half Message self._delete_half_messages(tx) self._save_state() logger.info(fTransaction rolled back: {transaction_id}) return True def register_check_callback(self, transaction_id: str, callback: Callable): 注册回查回调 self.check_callbacks[transaction_id] callback def _tag_half_message(self, value: bytes, transaction_id: str) - bytes: 标记 Half Message tag f__TRANSACTION__:{transaction_id}:.encode() return tag value def _make_messages_visible(self, tx: Transaction): 使消息可见 # 在实际系统中这里会更新索引标记 # 简化实现直接标记事务已完成 pass def _delete_half_messages(self, tx: Transaction): 删除 Half Message # 在实际系统中这里会删除对应的日志记录 pass def _check_loop(self): 回查循环 while self.running: time.sleep(self.check_interval_ms / 1000.0) with self.lock: now time.time() for tx_id, tx in list(self.transactions.items()): if tx.status ! TransactionStatus.PREPARING: continue elapsed (now - tx.created_at) * 1000 if elapsed self.transaction_timeout_ms: tx.check_times 1 # 调用回查回调 callback self.check_callbacks.get(tx_id) if callback: try: result callback(tx) if result: self.commit(tx_id) else: self.rollback(tx_id) except Exception as e: logger.error(fCheck callback error: {e}) if tx.check_times 3: # 多次回查失败回滚 self.rollback(tx_id) def _save_state(self): 持久化事务状态 data {} for tx_id, tx in self.transactions.items(): data[tx_id] { topic: tx.topic, partition: tx.partition, messages: tx.messages, status: tx.status.name, created_at: tx.created_at, updated_at: tx.updated_at, check_times: tx.check_times } with open(self.state_file, w) as f: json.dump(data, f, indent2) def _load_state(self): 加载持久化的事务状态 if not os.path.exists(self.state_file): return try: with open(self.state_file, r) as f: data json.load(f) for tx_id, info in data.items(): tx Transaction( transaction_idtx_id, topicinfo[topic], partitioninfo[partition], messagesinfo[messages], statusTransactionStatus[info[status]], created_atinfo[created_at], updated_atinfo[updated_at], check_timesinfo[check_times] ) self.transactions[tx_id] tx except Exception as e: logger.error(fLoad transaction state error: {e}) def stop(self): 停止 self.running False self._save_state()三、事务生产者3.1 事务消息生产者# mq/transaction/transaction_producer.py import uuid import logging import json from typing import Optional, Callable from ..protocol.message import Message, MessageType from ..transport.server import TCPClient from .transaction import TransactionManager logger logging.getLogger(__name__) class TransactionProducer: 事务消息生产者 使用两阶段提交保证本地事务和消息发送的原子性 def __init__(self, brokers: list, storage): self.brokers brokers self.storage storage # 事务管理器 self.tx_manager TransactionManager(storage) # 网络连接 self.client TCPClient(*brokers[0]) self.client.connect() def send_message_in_transaction(self, topic: str, value: str, key: str , local_executor: Callable None) - bool: 在事务中发送消息 Args: topic: 主题 value: 消息内容 key: 消息键 local_executor: 本地事务执行器返回 True/False Returns: 事务是否成功 transaction_id ftx-{uuid.uuid4().hex[:12]} # 1. 发送 Half Message half_success self._send_half_message( transaction_id, topic, 0, value, key ) if not half_success: logger.error(Failed to send half message) return False # 2. 执行本地事务 try: if local_executor: local_result local_executor() else: local_result True except Exception as e: logger.error(fLocal transaction failed: {e}) local_result False # 3. 根据本地事务结果 Commit 或 Rollback if local_result: return self.tx_manager.commit(transaction_id) else: self.tx_manager.rollback(transaction_id) return False def _send_half_message(self, transaction_id: str, topic: str, partition: int, value: str, key: str) - bool: 发送 Half Message # 存储 Half Message messages [{key: key, value: value}] return self.tx_manager.prepare( transaction_id, topic, partition, messages ) def close(self): 关闭 self.tx_manager.stop() self.client.disconnect() class TransactionConsumer: 事务消息消费者 能够识别并正确处理 Half Message def __init__(self, brokers: list, group_id: str): self.brokers brokers self.group_id group_id self.client TCPClient(*brokers[0]) self.client.connect() def consume(self, topic: str, partition: int 0) - list: 消费消息过滤掉未提交的 Half Message import json # 发送拉取请求 msg Message( msg_typeMessageType.FETCH_REQUEST, topictopic, valuejson.dumps({ topic: topic, partition: partition, offset: 0, max_bytes: 1048576 }).encode() ) response self.client.send(msg) if not response: return [] result json.loads(response.value.decode()) messages result.get(messages, []) # 过滤 Half Message visible_messages [] for m in messages: value m.get(value, ) if value.startswith(__TRANSACTION__:): # 这是 Half Message检查事务状态 parts value.split(:, 2) if len(parts) 3: tx_id parts[1] # 如果事务已提交提取真实消息 real_value parts[2] m[value] real_value visible_messages.append(m) # 注意这里简化处理实际应该查询事务状态 else: visible_messages.append(m) return visible_messages四、演示# examples/transaction_demo.py import time import logging import sys import os import threading import tempfile logging.basicConfig(levellogging.INFO) sys.path.insert(0, ..) from mq.broker.broker import Broker from mq.producer.producer import Producer from mq.consumer.consumer import Consumer from mq.ordered.ordered_producer import OrderedProducer, OrderedConsumer from mq.transaction.transaction import TransactionManager from mq.transaction.transaction_producer import TransactionProducer, TransactionConsumer def demo_ordered_messages(): 演示顺序消息 print( * 80) print( 顺序消息演示) print( * 80) broker Broker(host0.0.0.0, port29692, data_dir/tmp/mq_demo_ord) broker.create_topic(order-events, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 发送顺序消息 print(\n发送顺序消息同一订单的消息按顺序:) producer OrderedProducer(brokers[(localhost, 29692)]) # 模拟订单 A 的三个事件 order_a_events [ (order-A, 订单创建), (order-A, 支付成功), (order-A, 发货完成), ] order_b_events [ (order-B, 订单创建), (order-B, 支付成功), (order-B, 退款处理), ] for key, value in order_a_events order_b_events: offset producer.send_sync(order-events, value, keykey) print(f 发送: [{key}] {value} → offset{offset}) producer.close() # 消费消息观察顺序 print(\n消费消息同一订单的消息应该连续:) consumer Consumer( brokers[(localhost, 29692)], group_idorder-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(order-events) consumer.start() time.sleep(1) consumer.stop() print(f\n消费结果:) for msg in received: print(f {msg[value]}) broker.stop() def demo_transaction_messages(): 演示事务消息 print(\n * 80) print( 事务消息演示) print( * 80) broker Broker(host0.0.0.0, port29792, data_dir/tmp/mq_demo_tx) broker.create_topic(payment-events, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 创建事务生产者 print(\n事务消息示例转账) print(账户A: 余额 1000 元) print(账户B: 余额 500 元) print(转账金额: 200 元) accounts {A: 1000, B: 500} def transfer(from_acct: str, to_acct: str, amount: int) - bool: 本地转账事务 if accounts[from_acct] amount: print(f ❌ 余额不足: {accounts[from_acct]} {amount}) return False accounts[from_acct] - amount accounts[to_acct] amount print(f ✅ 转账成功: {from_acct}({accounts[from_acct]}) → f{to_acct}({accounts[to_acct]})) return True # 成功的转账 print(\n1. 成功转账:) tx_producer TransactionProducer( brokers[(localhost, 29792)], storagebroker.store ) success tx_producer.send_message_in_transaction( payment-events, 转账200元: A→B, keytransfer-001, local_executorlambda: transfer(A, B, 200) ) print(f 事务结果: {✅ 已提交 if success else ❌ 已回滚}) # 失败的转账 print(\n2. 失败转账余额不足:) success tx_producer.send_message_in_transaction( payment-events, 转账1000元: B→A, keytransfer-002, local_executorlambda: transfer(B, A, 1000) ) print(f 事务结果: {✅ 已提交 if success else ❌ 已回滚}) tx_producer.close() # 消费消息只看到成功的事务消息 print(\n3. 消费消息只显示已提交的事务:) tx_consumer TransactionConsumer( brokers[(localhost, 29792)], group_idpayment-group ) messages tx_consumer.consume(payment-events) for msg in messages: print(f {msg[value]}) print(f\n 最终余额: A{accounts[A]}, B{accounts[B]}) broker.stop() def demo_transaction_timeout(): 演示事务超时回查 print(\n * 80) print(⏱️ 事务超时回查演示) print( * 80) broker Broker(host0.0.0.0, port29892, data_dir/tmp/mq_demo_tx2) broker.create_topic(timeout-test, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 创建一个短超时的事务管理器 tx_manager TransactionManager( broker.store, transaction_timeout_ms3000, # 3秒超时 check_interval_ms1000 # 每秒检查 ) # 发送 Half Message 但不提交 print(\n发送 Half Message不提交等待超时:) tx_id test-timeout-tx tx_manager.prepare(tx_id, timeout-test, 0, [{key: timeout-key, value: 这条消息会被回滚}]) # 注册回查回调模拟网络中断一直返回 False tx_manager.register_check_callback(tx_id, lambda tx: False) print( 等待超时回查...) time.sleep(5) # 检查事务状态 tx tx_manager.transactions.get(tx_id) if tx: print(f 事务状态: {tx.status.name}) print(f 回查次数: {tx.check_times}) tx_manager.stop() broker.stop() if __name__ __main__: demo_ordered_messages() demo_transaction_messages() demo_transaction_timeout()五、测试# tests/test_transaction.py import unittest import time import threading import tempfile from mq.ordered.ordered_producer import OrderedProducer from mq.transaction.transaction import TransactionManager, TransactionStatus from mq.transaction.transaction_producer import TransactionProducer from mq.broker.broker import Broker class TestOrderedProducer(unittest.TestCase): 顺序消息测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.broker Broker(host0.0.0.0, port29992, data_dirself.tmpdir) self.broker.create_topic(test, partitions1) self.bt threading.Thread(targetself.broker.start, daemonTrue) self.bt.start() time.sleep(0.3) def tearDown(self): self.broker.stop() def test_ordered_send(self): 测试顺序发送 producer OrderedProducer(brokers[(localhost, 29992)]) # 发送相同 key 的消息 offsets [] for i in range(5): offset producer.send_sync(test, fmsg-{i}, keysame-key) offsets.append(offset) # 验证 offset 递增 for i in range(1, len(offsets)): self.assertGreater(offsets[i], offsets[i-1]) producer.close() class TestTransactionManager(unittest.TestCase): 事务管理器测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.broker Broker(host0.0.0.0, port29993, data_dirself.tmpdir) self.broker.create_topic(test, partitions1) self.bt threading.Thread(targetself.broker.start, daemonTrue) self.bt.start() time.sleep(0.3) self.tx_manager TransactionManager( self.broker.store, transaction_timeout_ms60000 ) def tearDown(self): self.tx_manager.stop() self.broker.stop() def test_prepare_commit(self): 测试准备和提交 tx_id test-tx-1 # 准备 result self.tx_manager.prepare( tx_id, test, 0, [{key: k1, value: v1}] ) self.assertTrue(result) # 检查状态 tx self.tx_manager.transactions.get(tx_id) self.assertEqual(tx.status, TransactionStatus.PREPARING) # 提交 result self.tx_manager.commit(tx_id) self.assertTrue(result) # 检查状态变更 tx self.tx_manager.transactions.get(tx_id) self.assertEqual(tx.status, TransactionStatus.COMMITTED) def test_prepare_rollback(self): 测试准备和回滚 tx_id test-tx-2 self.tx_manager.prepare(tx_id, test, 0, [{key: k2, value: v2}]) result self.tx_manager.rollback(tx_id) self.assertTrue(result) tx self.tx_manager.transactions.get(tx_id) self.assertEqual(tx.status, TransactionStatus.ROLLBACK) def test_double_commit(self): 测试重复提交 tx_id test-tx-3 self.tx_manager.prepare(tx_id, test, 0, [{key: k3, value: v3}]) self.tx_manager.commit(tx_id) # 再次提交应该失败 result self.tx_manager.commit(tx_id) self.assertFalse(result) if __name__ __main__: unittest.main()六、总结这一讲我们实现了两个重要的高级特性特性组件功能顺序消息​OrderedProducer保证相同 Key 的消息按发送顺序到达顺序消息​OrderedConsumer保证消息按顺序投递给业务处理器事务消息​TransactionManager管理事务生命周期和回查事务消息​TransactionProducer两阶段提交保证原子性事务消息​TransactionConsumer过滤 Half Message只消费已提交消息关键成果✅ 分区内严格顺序消费✅ 事务消息两阶段提交✅ 事务超时自动回查✅ Half Message 对消费者透明✅ 本地事务与消息发送原子性下一讲我们将实现延时消息与死信队列处理定时任务和异常消息的场景。开发之余的小工具推荐​处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。