前三讲我们实现了消息的生产和存储。现在Producer 可以高效地发送消息Broker 可以可靠地存储消息。但消息队列的核心价值在于消费——多个消费者如何协同工作如何保证消息不重复、不丢失如何动态扩缩容这一讲我们来实现消费者组机制这是 Kafka 等成熟消息队列最核心的设计之一。一、消费者组架构1.1 什么是消费者组Topic: orders (3个分区) ┌──────────┐ ┌──────────┐ ┌──────────┐ │Partition0│ │Partition1│ │Partition2│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ └──────┬─────┴──────┬─────┘ │ │ ┌──────▼─────┐ ┌───▼──────┐ │ Consumer A │ │ Consumer B│ ← 同一个消费者组 └────────────┘ └───────────┘ 每个分区只能被组内一个消费者消费 消费者数 ≤ 分区数1.2 核心概念概念说明Consumer Group消费者组组内消费者共同消费一个 TopicGroup Coordinator组协调器负责管理组成员和分配分区Rebalance重平衡组成员变化时重新分配分区Offset消费偏移量记录每个分区已消费的位置Commit提交 Offset标记消息已被消费1.3 消费流程Consumer 启动 │ ├──▶ 向 Coordinator 注册 │ ├──▶ JoinGroup加入消费者组 │ ├──▶ SyncGroup获取分区分配 │ ├──▶ 开始消费从上次提交的 Offset 开始 │ │ │ ├──▶ 拉取消息 │ ├──▶ 处理消息 │ └──▶ 提交 Offset │ └──▶ 发生 Rebalance 时重新分配二、Offset 存储2.1 Offset 管理器# mq/consumer/offset_store.py import json import os import threading import time import logging from typing import Dict, Optional from dataclasses import dataclass, field logger logging.getLogger(__name__) dataclass class OffsetMetadata: Offset 元数据 offset: int metadata: str timestamp: float 0.0 def __post_init__(self): if not self.timestamp: self.timestamp time.time() class OffsetStore: Offset 存储 支持 - 自动提交 - 手动提交 - 持久化到磁盘 def __init__(self, data_dir: str ./offset_data, auto_commit: bool True, auto_commit_interval_ms: int 5000): self.data_dir data_dir self.auto_commit auto_commit self.auto_commit_interval_ms auto_commit_interval_ms # offsets[group_id][topic][partition] OffsetMetadata self.offsets: Dict[str, Dict[str, Dict[int, OffsetMetadata]]] {} self.lock threading.Lock() os.makedirs(data_dir, exist_okTrue) # 恢复已保存的 Offset self._recover() # 启动自动提交 if auto_commit: self._start_auto_commit() def commit(self, group_id: str, topic: str, partition: int, offset: int, metadata: str ): 提交 Offset with self.lock: if group_id not in self.offsets: self.offsets[group_id] {} if topic not in self.offsets[group_id]: self.offsets[group_id][topic] {} self.offsets[group_id][topic][partition] OffsetMetadata( offsetoffset, metadatametadata ) def get_offset(self, group_id: str, topic: str, partition: int) - int: 获取已提交的 Offset with self.lock: meta self.offsets.get(group_id, {}).get(topic, {}).get(partition) return meta.offset if meta else 0 def _save(self): 持久化 Offset 到磁盘 filepath os.path.join(self.data_dir, offsets.json) with self.lock: data {} for group_id, topics in self.offsets.items(): data[group_id] {} for topic, partitions in topics.items(): data[group_id][topic] {} for partition, meta in partitions.items(): data[group_id][topic][str(partition)] { offset: meta.offset, metadata: meta.metadata, timestamp: meta.timestamp } with open(filepath, w) as f: json.dump(data, f, indent2) def _recover(self): 从磁盘恢复 Offset filepath os.path.join(self.data_dir, offsets.json) if not os.path.exists(filepath): return try: with open(filepath, r) as f: data json.load(f) for group_id, topics in data.items(): self.offsets[group_id] {} for topic, partitions in topics.items(): self.offsets[group_id][topic] {} for partition, meta in partitions.items(): self.offsets[group_id][topic][int(partition)] \ OffsetMetadata( offsetmeta[offset], metadatameta.get(metadata, ), timestampmeta.get(timestamp, 0.0) ) except Exception as e: logger.error(fRecover offsets error: {e}) def _start_auto_commit(self): 启动自动提交 def auto_commit_loop(): while True: time.sleep(self.auto_commit_interval_ms / 1000.0) self._save() thread threading.Thread(targetauto_commit_loop, daemonTrue) thread.start() def close(self): 关闭存储 self._save()三、消费者组协调器3.1 组协调器# mq/consumer/coordinator.py import threading import time import logging from typing import Dict, List, Set, Optional, Callable from dataclasses import dataclass, field from enum import Enum, auto logger logging.getLogger(__name__) class MemberStatus(Enum): 成员状态 JOINING auto() # 正在加入 SYNCING auto() # 正在同步 STABLE auto() # 稳定运行 LEAVING auto() # 正在离开 dataclass class GroupMember: 组成员 member_id: str host: str port: int status: MemberStatus MemberStatus.JOINING last_heartbeat: float 0.0 assigned_partitions: List[tuple] field(default_factorylist) # [(topic, partition)] class GroupCoordinator: 消费者组协调器 职责 1. 管理组成员 2. 触发 Rebalance 3. 分配分区 4. 检测心跳超时 def __init__(self, group_id: str, session_timeout_ms: int 45000, heartbeat_interval_ms: int 3000, rebalance_timeout_ms: int 60000): self.group_id group_id self.session_timeout_ms session_timeout_ms self.heartbeat_interval_ms heartbeat_interval_ms self.rebalance_timeout_ms rebalance_timeout_ms self.members: Dict[str, GroupMember] {} self.generation 0 # Rebalance 代数 self.leader_id: Optional[str] None self.lock threading.Lock() # 分区分配策略 self.assignment_strategy RangeAssignor() # 回调 self.on_rebalance: Optional[Callable] None # 心跳检测线程 self.running True self.heartbeat_thread threading.Thread(targetself._heartbeat_check, daemonTrue) self.heartbeat_thread.start() def register(self, member_id: str, host: str, port: int) - bool: 注册消费者成员 Returns: 是否需要 Rebalance with self.lock: if member_id in self.members: member self.members[member_id] member.status MemberStatus.JOINING member.last_heartbeat time.time() return False member GroupMember( member_idmember_id, hosthost, portport ) self.members[member_id] member # 第一个成员成为 Leader if self.leader_id is None: self.leader_id member_id logger.info(fMember registered: {member_id} f(total: {len(self.members)})) return True # 需要 Rebalance def unregister(self, member_id: str): 注销成员 with self.lock: if member_id in self.members: del self.members[member_id] logger.info(fMember unregistered: {member_id} f(total: {len(self.members)})) # 重新选举 Leader if self.leader_id member_id and self.members: self.leader_id next(iter(self.members.keys())) def heartbeat(self, member_id: str) - bool: 心跳 Returns: 是否需要进行 Rebalance with self.lock: if member_id in self.members: self.members[member_id].last_heartbeat time.time() self.members[member_id].status MemberStatus.STABLE return False return True # 未知成员触发 Rebalance def trigger_rebalance(self, subscribed_topics: Dict[str, List[int]]): 触发重平衡 Args: subscribed_topics: 订阅的 topic 及其分区列表 with self.lock: self.generation 1 logger.info(fRebalance triggered (generation {self.generation})) # 收集活跃成员 active_members self._get_active_members() # 分配分区 assignments self.assignment_strategy.assign( active_members, subscribed_topics ) # 更新成员的分配 for member_id, partitions in assignments.items(): if member_id in self.members: self.members[member_id].assigned_partitions partitions self.members[member_id].status MemberStatus.STABLE # 通知回调 if self.on_rebalance: self.on_rebalance(assignments) return assignments def _get_active_members(self) - List[str]: 获取活跃成员列表 now time.time() active [] for member_id, member in self.members.items(): elapsed (now - member.last_heartbeat) * 1000 if elapsed self.session_timeout_ms: active.append(member_id) else: logger.warning(fMember heartbeat timeout: {member_id}) return active def _heartbeat_check(self): 心跳检测 while self.running: time.sleep(self.heartbeat_interval_ms / 1000.0) with self.lock: now time.time() for member_id, member in list(self.members.items()): elapsed (now - member.last_heartbeat) * 1000 if elapsed self.session_timeout_ms: logger.warning(fRemoving dead member: {member_id}) del self.members[member_id] def stop(self): 停止协调器 self.running False四、分区分配策略4.1 多种分配算法# mq/consumer/assignor.py from typing import Dict, List, Tuple from collections import defaultdict class PartitionAssignor: 分区分配器基类 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: raise NotImplementedError class RangeAssignor(PartitionAssignor): 范围分配 按 Topic 依次分配每个消费者获得连续的一段分区 优点实现简单 缺点可能导致分配不均 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) for topic, partitions in topics.items(): if not partitions: continue num_members len(members) num_partitions len(partitions) # 计算每个消费者应得的分区数 partitions_per_member num_partitions // num_members remainder num_partitions % num_members start 0 for i, member in enumerate(members): # 前 remainder 个消费者多分一个分区 count partitions_per_member (1 if i remainder else 0) for j in range(count): if start j num_partitions: assignments[member].append((topic, partitions[start j])) start count return dict(assignments) class RoundRobinAssignor(PartitionAssignor): 轮询分配 将所有分区排序后轮流分配给消费者 优点分配均匀 缺点分区数远大于消费者数时效果更好 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) # 收集所有分区 all_partitions [] for topic, partitions in topics.items(): for p in partitions: all_partitions.append((topic, p)) # 轮询分配 for i, (topic, partition) in enumerate(all_partitions): member members[i % len(members)] assignments[member].append((topic, partition)) return dict(assignments) class StickyAssignor(PartitionAssignor): 粘性分配 尽量保持已有的分配不变只在必要时调整 优点减少 Rebalance 时的分区移动 缺点实现复杂 def assign(self, members: List[str], topics: Dict[str, List[int]], previous_assignments: Dict[str, List[Tuple[str, int]]] None) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) if not previous_assignments: # 首次分配使用轮询 return RoundRobinAssignor().assign(members, topics) # 尝试保持已有分配 all_partitions set() for topic, partitions in topics.items(): for p in partitions: all_partitions.add((topic, p)) # 优先保留已有分配 assigned set() for member in members: if member in previous_assignments: for tp in previous_assignments[member]: if tp in all_partitions and tp not in assigned: assignments[member].append(tp) assigned.add(tp) # 分配剩余分区 remaining all_partitions - assigned for i, tp in enumerate(sorted(remaining)): member members[i % len(members)] assignments[member].append(tp) return dict(assignments)五、消费者实现5.1 完整消费者# mq/consumer/consumer.py import threading import time import json import logging import uuid from typing import List, Dict, Optional, Callable from ..protocol.message import * from ..transport.server import TCPClient from .offset_store import OffsetStore from .coordinator import GroupCoordinator logger logging.getLogger(__name__) class Consumer: 消费者 支持 - 消费者组 - 自动/手动提交 Offset - Rebalance 监听 - 批量拉取 def __init__(self, brokers: List[tuple] [(localhost, 9092)], group_id: str default-group, client_id: str None, auto_commit: bool True, auto_commit_interval_ms: int 5000, session_timeout_ms: int 45000, heartbeat_interval_ms: int 3000): self.brokers brokers self.group_id group_id self.client_id client_id or fconsumer-{uuid.uuid4().hex[:8]} self.auto_commit auto_commit self.auto_commit_interval_ms auto_commit_interval_ms # 网络连接 self.client TCPClient(*brokers[0]) # Offset 管理 self.offset_store OffsetStore( data_dirf./offset_data/{group_id}, auto_commitauto_commit, auto_commit_interval_msauto_commit_interval_ms ) # 组协调 self.coordinator GroupCoordinator( group_idgroup_id, session_timeout_mssession_timeout_ms, heartbeat_interval_msheartbeat_interval_ms ) # 订阅信息 self.subscriptions: Dict[str, List[int]] {} # topic - [partitions] self.assigned_partitions: List[tuple] [] # [(topic, partition)] # 消费状态 self.running False self.paused False # 消息处理回调 self.message_handler: Optional[Callable] None # Rebalance 监听器 self.rebalance_listeners: List[Callable] [] # 统计 self.metrics { fetched: 0, committed: 0, rebalances: 0 } def subscribe(self, topic: str, partitions: List[int] None): 订阅主题 Args: topic: 主题名 partitions: 分区列表None 表示所有分区 if partitions is None: # 简化假设每个 topic 有 3 个分区 partitions [0, 1, 2] self.subscriptions[topic] partitions logger.info(fSubscribed to {topic}: partitions{partitions}) # 注册到协调器 self.coordinator.register( self.client_id, self.client.host, self.client.port ) def start(self): 启动消费 self.running True # 连接到 Broker self.client.connect() # 注册 Rebalance 回调 self.coordinator.on_rebalance self._on_rebalance # 触发初始 Rebalance self._do_rebalance() # 启动消费循环 self.consumer_thread threading.Thread(targetself._consume_loop, daemonTrue) self.consumer_thread.start() logger.info(fConsumer started: {self.client_id}) def stop(self): 停止消费 self.running False # 提交最后的 Offset self.offset_store.close() # 注销 self.coordinator.unregister(self.client_id) self.client.disconnect() logger.info(Consumer stopped) def _do_rebalance(self): 执行重平衡 # 触发协调器分配 assignments self.coordinator.trigger_rebalance(self.subscriptions) # 更新自己的分配 self.assigned_partitions assignments.get(self.client_id, []) self.metrics[rebalances] 1 # 通知监听器 for listener in self.rebalance_listeners: try: listener(self.assigned_partitions) except Exception as e: logger.error(fRebalance listener error: {e}) logger.info(fAssigned partitions: {self.assigned_partitions}) def _on_rebalance(self, assignments: dict): Rebalance 回调 # 更新自己的分配 self.assigned_partitions assignments.get(self.client_id, []) logger.info(fRebalance completed, assigned: {self.assigned_partitions}) def _consume_loop(self): 消费循环 while self.running: if self.paused: time.sleep(0.1) continue if not self.assigned_partitions: time.sleep(0.5) continue # 遍历所有分配的分区 for topic, partition in self.assigned_partitions: self._fetch_and_process(topic, partition) time.sleep(0.05) # 防止空转 def _fetch_and_process(self, topic: str, partition: int): 拉取并处理消息 # 获取已提交的 Offset committed_offset self.offset_store.get_offset( self.group_id, topic, partition ) # 发送拉取请求 msg Message( msg_typeMessageType.FETCH_REQUEST, topictopic, valuejson.dumps({ topic: topic, partition: partition, offset: committed_offset, max_bytes: 1024 * 1024 }).encode() ) response self.client.send(msg) if not response: return result json.loads(response.value.decode()) messages result.get(messages, []) if not messages: return # 处理消息 for msg_data in messages: try: if self.message_handler: self.message_handler(msg_data) else: logger.debug(fReceived: {msg_data}) self.metrics[fetched] 1 # 自动提交 Offset if self.auto_commit: self.offset_store.commit( self.group_id, topic, partition, msg_data[offset] 1 ) self.metrics[committed] 1 except Exception as e: logger.error(fProcess message error: {e}) def commit_sync(self): 手动提交 Offset self.offset_store._save() logger.info(Offsets committed manually) def pause(self): 暂停消费 self.paused True def resume(self): 恢复消费 self.paused False def seek(self, topic: str, partition: int, offset: int): 重置消费位置 用于回溯消费 self.offset_store.commit(self.group_id, topic, partition, offset) logger.info(fSeek to {topic}/{partition}: offset{offset}) def add_rebalance_listener(self, listener: Callable): 添加 Rebalance 监听器 self.rebalance_listeners.append(listener) def get_metrics(self) - dict: 获取统计 return dict(self.metrics)六、演示# examples/consumer_group_demo.py import time import logging import sys import os import threading 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.consumer.assignor import RangeAssignor, RoundRobinAssignor def demo_basic_consumption(): 演示基本消费 print( * 80) print( 基本消费演示) print( * 80) broker Broker(host0.0.0.0, port19492, data_dir/tmp/mq_demo_c1) broker.create_topic(test, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 先生产一些消息 producer Producer(brokers[(localhost, 19492)]) producer.start() for i in range(10): producer.send(test, fMessage #{i}, keyfk{i}) producer.flush() producer.stop() # 消费消息 print(\n消费消息:) consumer Consumer( brokers[(localhost, 19492)], group_iddemo-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(test) consumer.start() time.sleep(1) consumer.stop() print(f 收到 {len(received)} 条消息) for msg in received[:5]: print(f offset{msg[offset]}: {msg[value]}) broker.stop() def demo_consumer_group(): 演示消费者组 print(\n * 80) print( 消费者组演示) print( * 80) broker Broker(host0.0.0.0, port19493, data_dir/tmp/mq_demo_c2) broker.create_topic(group-test, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 生产消息 producer Producer(brokers[(localhost, 19493)]) producer.start() for i in range(30): producer.send(group-test, fMsg-{i}, keyfk{i}) producer.flush() producer.stop() # 启动两个消费者同一组 print(\n启动消费者组2个消费者:) results {1: [], 2: []} def make_handler(consumer_id): def handler(msg): results[consumer_id].append(msg) print(f Consumer{consumer_id}: {msg[value]}) return handler consumer1 Consumer( brokers[(localhost, 19493)], group_idshared-group, client_idconsumer-1 ) consumer1.message_handler make_handler(1) consumer1.subscribe(group-test) consumer1.start() consumer2 Consumer( brokers[(localhost, 19493)], group_idshared-group, client_idconsumer-2 ) consumer2.message_handler make_handler(2) consumer2.subscribe(group-test) consumer2.start() time.sleep(2) consumer1.stop() consumer2.stop() print(f\n 消费统计:) print(f Consumer1: {len(results[1])} 条) print(f Consumer2: {len(results[2])} 条) print(f 总计: {len(results[1]) len(results[2])} 条) broker.stop() def demo_offset_management(): 演示 Offset 管理 print(\n * 80) print( Offset 管理演示) print( * 80) broker Broker(host0.0.0.0, port19494, data_dir/tmp/mq_demo_c3) broker.create_topic(offset-demo, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 生产消息 producer Producer(brokers[(localhost, 19494)]) producer.start() for i in range(10): producer.send(offset-demo, fData-{i}) producer.flush() producer.stop() # 第一次消费消费前5条 print(\n第一次消费前5条:) consumer Consumer( brokers[(localhost, 19494)], group_idoffset-group, auto_commitTrue ) first_batch [] consumer.message_handler lambda msg: first_batch.append(msg) consumer.subscribe(offset-demo) consumer.start() time.sleep(1) consumer.stop() print(f 消费了 {len(first_batch)} 条) # 第二次消费应该从 offset5 开始 print(\n第二次消费应从 offset5 开始:) consumer2 Consumer( brokers[(localhost, 19494)], group_idoffset-group, auto_commitTrue ) second_batch [] consumer2.message_handler lambda msg: second_batch.append(msg) consumer2.subscribe(offset-demo) consumer2.start() time.sleep(1) consumer2.stop() print(f 消费了 {len(second_batch)} 条) if second_batch: print(f 第一条: {second_batch[0][value]}) # 回溯消费 print(\n回溯消费从 offset0 开始:) consumer3 Consumer( brokers[(localhost, 19494)], group_idrewind-group, auto_commitTrue ) third_batch [] consumer3.message_handler lambda msg: third_batch.append(msg) consumer3.subscribe(offset-demo) consumer3.seek(offset-demo, 0, 0) # 从头开始 consumer3.start() time.sleep(1) consumer3.stop() print(f 消费了 {len(third_batch)} 条全部) broker.stop() if __name__ __main__: demo_basic_consumption() demo_consumer_group() demo_offset_management()七、测试# tests/test_consumer.py import unittest import time import threading import tempfile from mq.consumer.consumer import Consumer from mq.consumer.offset_store import OffsetStore from mq.consumer.coordinator import GroupCoordinator from mq.consumer.assignor import RangeAssignor, RoundRobinAssignor from mq.broker.broker import Broker from mq.producer.producer import Producer class TestOffsetStore(unittest.TestCase): Offset 存储测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.store OffsetStore(self.tmpdir, auto_commitFalse) def tearDown(self): self.store.close() def test_commit_and_get(self): 测试提交和获取 self.store.commit(group1, topic1, 0, 42) offset self.store.get_offset(group1, topic1, 0) self.assertEqual(offset, 42) def test_default_offset(self): 测试默认 Offset offset self.store.get_offset(unknown, topic, 0) self.assertEqual(offset, 0) class TestGroupCoordinator(unittest.TestCase): 组协调器测试 def setUp(self): self.coordinator GroupCoordinator(test-group) def test_register(self): 测试注册 need_rebalance self.coordinator.register(m1, host1, 9092) self.assertTrue(need_rebalance) # 重复注册不应该触发 Rebalance need_rebalance self.coordinator.register(m1, host1, 9092) self.assertFalse(need_rebalance) def test_unregister(self): 测试注销 self.coordinator.register(m1, host1, 9092) self.coordinator.register(m2, host2, 9092) self.coordinator.unregister(m1) self.assertEqual(len(self.coordinator.members), 1) class TestAssignor(unittest.TestCase): 分区分配测试 def test_range_assignor(self): 测试范围分配 assignor RangeAssignor() members [c1, c2] topics {test: [0, 1, 2]} assignments assignor.assign(members, topics) self.assertEqual(len(assignments), 2) # c1 应该有 2 个分区c2 有 1 个 self.assertEqual(len(assignments[c1]), 2) self.assertEqual(len(assignments[c2]), 1) def test_round_robin(self): 测试轮询分配 assignor RoundRobinAssignor() members [c1, c2, c3] topics {t1: [0, 1], t2: [0, 1]} assignments assignor.assign(members, topics) # 总共 4 个分区3 个消费者 total sum(len(v) for v in assignments.values()) self.assertEqual(total, 4) class TestConsumerIntegration(unittest.TestCase): 消费者集成测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.broker Broker(host0.0.0.0, port19592, 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_consume_messages(self): 测试消费消息 # 先生产 producer Producer(brokers[(localhost, 19592)]) producer.start() for i in range(5): producer.send(test, fmsg-{i}) producer.flush() producer.stop() # 再消费 consumer Consumer( brokers[(localhost, 19592)], group_idtest-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(test) consumer.start() time.sleep(1) consumer.stop() self.assertEqual(len(received), 5) if __name__ __main__: unittest.main()八、总结这一讲我们实现了消费者组的核心机制组件功能OffsetStoreOffset 持久化存储支持自动/手动提交GroupCoordinator组协调器管理成员和触发 RebalancePartitionAssignor分区分配策略Range/RoundRobin/StickyConsumer完整消费者支持组消费和 Offset 管理关键成果✅ 消费者组机制支持水平扩展✅ 自动 Rebalance动态感知成员变化✅ Offset 自动提交断点续传✅ 支持回溯消费seek✅ 三种分区分配策略下一讲我们将实现主从复制与高可用机制让消息队列在节点故障时依然可用。开发之余的小工具推荐处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。
第4讲：消费者组与Offset管理
前三讲我们实现了消息的生产和存储。现在Producer 可以高效地发送消息Broker 可以可靠地存储消息。但消息队列的核心价值在于消费——多个消费者如何协同工作如何保证消息不重复、不丢失如何动态扩缩容这一讲我们来实现消费者组机制这是 Kafka 等成熟消息队列最核心的设计之一。一、消费者组架构1.1 什么是消费者组Topic: orders (3个分区) ┌──────────┐ ┌──────────┐ ┌──────────┐ │Partition0│ │Partition1│ │Partition2│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ └──────┬─────┴──────┬─────┘ │ │ ┌──────▼─────┐ ┌───▼──────┐ │ Consumer A │ │ Consumer B│ ← 同一个消费者组 └────────────┘ └───────────┘ 每个分区只能被组内一个消费者消费 消费者数 ≤ 分区数1.2 核心概念概念说明Consumer Group消费者组组内消费者共同消费一个 TopicGroup Coordinator组协调器负责管理组成员和分配分区Rebalance重平衡组成员变化时重新分配分区Offset消费偏移量记录每个分区已消费的位置Commit提交 Offset标记消息已被消费1.3 消费流程Consumer 启动 │ ├──▶ 向 Coordinator 注册 │ ├──▶ JoinGroup加入消费者组 │ ├──▶ SyncGroup获取分区分配 │ ├──▶ 开始消费从上次提交的 Offset 开始 │ │ │ ├──▶ 拉取消息 │ ├──▶ 处理消息 │ └──▶ 提交 Offset │ └──▶ 发生 Rebalance 时重新分配二、Offset 存储2.1 Offset 管理器# mq/consumer/offset_store.py import json import os import threading import time import logging from typing import Dict, Optional from dataclasses import dataclass, field logger logging.getLogger(__name__) dataclass class OffsetMetadata: Offset 元数据 offset: int metadata: str timestamp: float 0.0 def __post_init__(self): if not self.timestamp: self.timestamp time.time() class OffsetStore: Offset 存储 支持 - 自动提交 - 手动提交 - 持久化到磁盘 def __init__(self, data_dir: str ./offset_data, auto_commit: bool True, auto_commit_interval_ms: int 5000): self.data_dir data_dir self.auto_commit auto_commit self.auto_commit_interval_ms auto_commit_interval_ms # offsets[group_id][topic][partition] OffsetMetadata self.offsets: Dict[str, Dict[str, Dict[int, OffsetMetadata]]] {} self.lock threading.Lock() os.makedirs(data_dir, exist_okTrue) # 恢复已保存的 Offset self._recover() # 启动自动提交 if auto_commit: self._start_auto_commit() def commit(self, group_id: str, topic: str, partition: int, offset: int, metadata: str ): 提交 Offset with self.lock: if group_id not in self.offsets: self.offsets[group_id] {} if topic not in self.offsets[group_id]: self.offsets[group_id][topic] {} self.offsets[group_id][topic][partition] OffsetMetadata( offsetoffset, metadatametadata ) def get_offset(self, group_id: str, topic: str, partition: int) - int: 获取已提交的 Offset with self.lock: meta self.offsets.get(group_id, {}).get(topic, {}).get(partition) return meta.offset if meta else 0 def _save(self): 持久化 Offset 到磁盘 filepath os.path.join(self.data_dir, offsets.json) with self.lock: data {} for group_id, topics in self.offsets.items(): data[group_id] {} for topic, partitions in topics.items(): data[group_id][topic] {} for partition, meta in partitions.items(): data[group_id][topic][str(partition)] { offset: meta.offset, metadata: meta.metadata, timestamp: meta.timestamp } with open(filepath, w) as f: json.dump(data, f, indent2) def _recover(self): 从磁盘恢复 Offset filepath os.path.join(self.data_dir, offsets.json) if not os.path.exists(filepath): return try: with open(filepath, r) as f: data json.load(f) for group_id, topics in data.items(): self.offsets[group_id] {} for topic, partitions in topics.items(): self.offsets[group_id][topic] {} for partition, meta in partitions.items(): self.offsets[group_id][topic][int(partition)] \ OffsetMetadata( offsetmeta[offset], metadatameta.get(metadata, ), timestampmeta.get(timestamp, 0.0) ) except Exception as e: logger.error(fRecover offsets error: {e}) def _start_auto_commit(self): 启动自动提交 def auto_commit_loop(): while True: time.sleep(self.auto_commit_interval_ms / 1000.0) self._save() thread threading.Thread(targetauto_commit_loop, daemonTrue) thread.start() def close(self): 关闭存储 self._save()三、消费者组协调器3.1 组协调器# mq/consumer/coordinator.py import threading import time import logging from typing import Dict, List, Set, Optional, Callable from dataclasses import dataclass, field from enum import Enum, auto logger logging.getLogger(__name__) class MemberStatus(Enum): 成员状态 JOINING auto() # 正在加入 SYNCING auto() # 正在同步 STABLE auto() # 稳定运行 LEAVING auto() # 正在离开 dataclass class GroupMember: 组成员 member_id: str host: str port: int status: MemberStatus MemberStatus.JOINING last_heartbeat: float 0.0 assigned_partitions: List[tuple] field(default_factorylist) # [(topic, partition)] class GroupCoordinator: 消费者组协调器 职责 1. 管理组成员 2. 触发 Rebalance 3. 分配分区 4. 检测心跳超时 def __init__(self, group_id: str, session_timeout_ms: int 45000, heartbeat_interval_ms: int 3000, rebalance_timeout_ms: int 60000): self.group_id group_id self.session_timeout_ms session_timeout_ms self.heartbeat_interval_ms heartbeat_interval_ms self.rebalance_timeout_ms rebalance_timeout_ms self.members: Dict[str, GroupMember] {} self.generation 0 # Rebalance 代数 self.leader_id: Optional[str] None self.lock threading.Lock() # 分区分配策略 self.assignment_strategy RangeAssignor() # 回调 self.on_rebalance: Optional[Callable] None # 心跳检测线程 self.running True self.heartbeat_thread threading.Thread(targetself._heartbeat_check, daemonTrue) self.heartbeat_thread.start() def register(self, member_id: str, host: str, port: int) - bool: 注册消费者成员 Returns: 是否需要 Rebalance with self.lock: if member_id in self.members: member self.members[member_id] member.status MemberStatus.JOINING member.last_heartbeat time.time() return False member GroupMember( member_idmember_id, hosthost, portport ) self.members[member_id] member # 第一个成员成为 Leader if self.leader_id is None: self.leader_id member_id logger.info(fMember registered: {member_id} f(total: {len(self.members)})) return True # 需要 Rebalance def unregister(self, member_id: str): 注销成员 with self.lock: if member_id in self.members: del self.members[member_id] logger.info(fMember unregistered: {member_id} f(total: {len(self.members)})) # 重新选举 Leader if self.leader_id member_id and self.members: self.leader_id next(iter(self.members.keys())) def heartbeat(self, member_id: str) - bool: 心跳 Returns: 是否需要进行 Rebalance with self.lock: if member_id in self.members: self.members[member_id].last_heartbeat time.time() self.members[member_id].status MemberStatus.STABLE return False return True # 未知成员触发 Rebalance def trigger_rebalance(self, subscribed_topics: Dict[str, List[int]]): 触发重平衡 Args: subscribed_topics: 订阅的 topic 及其分区列表 with self.lock: self.generation 1 logger.info(fRebalance triggered (generation {self.generation})) # 收集活跃成员 active_members self._get_active_members() # 分配分区 assignments self.assignment_strategy.assign( active_members, subscribed_topics ) # 更新成员的分配 for member_id, partitions in assignments.items(): if member_id in self.members: self.members[member_id].assigned_partitions partitions self.members[member_id].status MemberStatus.STABLE # 通知回调 if self.on_rebalance: self.on_rebalance(assignments) return assignments def _get_active_members(self) - List[str]: 获取活跃成员列表 now time.time() active [] for member_id, member in self.members.items(): elapsed (now - member.last_heartbeat) * 1000 if elapsed self.session_timeout_ms: active.append(member_id) else: logger.warning(fMember heartbeat timeout: {member_id}) return active def _heartbeat_check(self): 心跳检测 while self.running: time.sleep(self.heartbeat_interval_ms / 1000.0) with self.lock: now time.time() for member_id, member in list(self.members.items()): elapsed (now - member.last_heartbeat) * 1000 if elapsed self.session_timeout_ms: logger.warning(fRemoving dead member: {member_id}) del self.members[member_id] def stop(self): 停止协调器 self.running False四、分区分配策略4.1 多种分配算法# mq/consumer/assignor.py from typing import Dict, List, Tuple from collections import defaultdict class PartitionAssignor: 分区分配器基类 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: raise NotImplementedError class RangeAssignor(PartitionAssignor): 范围分配 按 Topic 依次分配每个消费者获得连续的一段分区 优点实现简单 缺点可能导致分配不均 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) for topic, partitions in topics.items(): if not partitions: continue num_members len(members) num_partitions len(partitions) # 计算每个消费者应得的分区数 partitions_per_member num_partitions // num_members remainder num_partitions % num_members start 0 for i, member in enumerate(members): # 前 remainder 个消费者多分一个分区 count partitions_per_member (1 if i remainder else 0) for j in range(count): if start j num_partitions: assignments[member].append((topic, partitions[start j])) start count return dict(assignments) class RoundRobinAssignor(PartitionAssignor): 轮询分配 将所有分区排序后轮流分配给消费者 优点分配均匀 缺点分区数远大于消费者数时效果更好 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) # 收集所有分区 all_partitions [] for topic, partitions in topics.items(): for p in partitions: all_partitions.append((topic, p)) # 轮询分配 for i, (topic, partition) in enumerate(all_partitions): member members[i % len(members)] assignments[member].append((topic, partition)) return dict(assignments) class StickyAssignor(PartitionAssignor): 粘性分配 尽量保持已有的分配不变只在必要时调整 优点减少 Rebalance 时的分区移动 缺点实现复杂 def assign(self, members: List[str], topics: Dict[str, List[int]], previous_assignments: Dict[str, List[Tuple[str, int]]] None) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) if not previous_assignments: # 首次分配使用轮询 return RoundRobinAssignor().assign(members, topics) # 尝试保持已有分配 all_partitions set() for topic, partitions in topics.items(): for p in partitions: all_partitions.add((topic, p)) # 优先保留已有分配 assigned set() for member in members: if member in previous_assignments: for tp in previous_assignments[member]: if tp in all_partitions and tp not in assigned: assignments[member].append(tp) assigned.add(tp) # 分配剩余分区 remaining all_partitions - assigned for i, tp in enumerate(sorted(remaining)): member members[i % len(members)] assignments[member].append(tp) return dict(assignments)五、消费者实现5.1 完整消费者# mq/consumer/consumer.py import threading import time import json import logging import uuid from typing import List, Dict, Optional, Callable from ..protocol.message import * from ..transport.server import TCPClient from .offset_store import OffsetStore from .coordinator import GroupCoordinator logger logging.getLogger(__name__) class Consumer: 消费者 支持 - 消费者组 - 自动/手动提交 Offset - Rebalance 监听 - 批量拉取 def __init__(self, brokers: List[tuple] [(localhost, 9092)], group_id: str default-group, client_id: str None, auto_commit: bool True, auto_commit_interval_ms: int 5000, session_timeout_ms: int 45000, heartbeat_interval_ms: int 3000): self.brokers brokers self.group_id group_id self.client_id client_id or fconsumer-{uuid.uuid4().hex[:8]} self.auto_commit auto_commit self.auto_commit_interval_ms auto_commit_interval_ms # 网络连接 self.client TCPClient(*brokers[0]) # Offset 管理 self.offset_store OffsetStore( data_dirf./offset_data/{group_id}, auto_commitauto_commit, auto_commit_interval_msauto_commit_interval_ms ) # 组协调 self.coordinator GroupCoordinator( group_idgroup_id, session_timeout_mssession_timeout_ms, heartbeat_interval_msheartbeat_interval_ms ) # 订阅信息 self.subscriptions: Dict[str, List[int]] {} # topic - [partitions] self.assigned_partitions: List[tuple] [] # [(topic, partition)] # 消费状态 self.running False self.paused False # 消息处理回调 self.message_handler: Optional[Callable] None # Rebalance 监听器 self.rebalance_listeners: List[Callable] [] # 统计 self.metrics { fetched: 0, committed: 0, rebalances: 0 } def subscribe(self, topic: str, partitions: List[int] None): 订阅主题 Args: topic: 主题名 partitions: 分区列表None 表示所有分区 if partitions is None: # 简化假设每个 topic 有 3 个分区 partitions [0, 1, 2] self.subscriptions[topic] partitions logger.info(fSubscribed to {topic}: partitions{partitions}) # 注册到协调器 self.coordinator.register( self.client_id, self.client.host, self.client.port ) def start(self): 启动消费 self.running True # 连接到 Broker self.client.connect() # 注册 Rebalance 回调 self.coordinator.on_rebalance self._on_rebalance # 触发初始 Rebalance self._do_rebalance() # 启动消费循环 self.consumer_thread threading.Thread(targetself._consume_loop, daemonTrue) self.consumer_thread.start() logger.info(fConsumer started: {self.client_id}) def stop(self): 停止消费 self.running False # 提交最后的 Offset self.offset_store.close() # 注销 self.coordinator.unregister(self.client_id) self.client.disconnect() logger.info(Consumer stopped) def _do_rebalance(self): 执行重平衡 # 触发协调器分配 assignments self.coordinator.trigger_rebalance(self.subscriptions) # 更新自己的分配 self.assigned_partitions assignments.get(self.client_id, []) self.metrics[rebalances] 1 # 通知监听器 for listener in self.rebalance_listeners: try: listener(self.assigned_partitions) except Exception as e: logger.error(fRebalance listener error: {e}) logger.info(fAssigned partitions: {self.assigned_partitions}) def _on_rebalance(self, assignments: dict): Rebalance 回调 # 更新自己的分配 self.assigned_partitions assignments.get(self.client_id, []) logger.info(fRebalance completed, assigned: {self.assigned_partitions}) def _consume_loop(self): 消费循环 while self.running: if self.paused: time.sleep(0.1) continue if not self.assigned_partitions: time.sleep(0.5) continue # 遍历所有分配的分区 for topic, partition in self.assigned_partitions: self._fetch_and_process(topic, partition) time.sleep(0.05) # 防止空转 def _fetch_and_process(self, topic: str, partition: int): 拉取并处理消息 # 获取已提交的 Offset committed_offset self.offset_store.get_offset( self.group_id, topic, partition ) # 发送拉取请求 msg Message( msg_typeMessageType.FETCH_REQUEST, topictopic, valuejson.dumps({ topic: topic, partition: partition, offset: committed_offset, max_bytes: 1024 * 1024 }).encode() ) response self.client.send(msg) if not response: return result json.loads(response.value.decode()) messages result.get(messages, []) if not messages: return # 处理消息 for msg_data in messages: try: if self.message_handler: self.message_handler(msg_data) else: logger.debug(fReceived: {msg_data}) self.metrics[fetched] 1 # 自动提交 Offset if self.auto_commit: self.offset_store.commit( self.group_id, topic, partition, msg_data[offset] 1 ) self.metrics[committed] 1 except Exception as e: logger.error(fProcess message error: {e}) def commit_sync(self): 手动提交 Offset self.offset_store._save() logger.info(Offsets committed manually) def pause(self): 暂停消费 self.paused True def resume(self): 恢复消费 self.paused False def seek(self, topic: str, partition: int, offset: int): 重置消费位置 用于回溯消费 self.offset_store.commit(self.group_id, topic, partition, offset) logger.info(fSeek to {topic}/{partition}: offset{offset}) def add_rebalance_listener(self, listener: Callable): 添加 Rebalance 监听器 self.rebalance_listeners.append(listener) def get_metrics(self) - dict: 获取统计 return dict(self.metrics)六、演示# examples/consumer_group_demo.py import time import logging import sys import os import threading 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.consumer.assignor import RangeAssignor, RoundRobinAssignor def demo_basic_consumption(): 演示基本消费 print( * 80) print( 基本消费演示) print( * 80) broker Broker(host0.0.0.0, port19492, data_dir/tmp/mq_demo_c1) broker.create_topic(test, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 先生产一些消息 producer Producer(brokers[(localhost, 19492)]) producer.start() for i in range(10): producer.send(test, fMessage #{i}, keyfk{i}) producer.flush() producer.stop() # 消费消息 print(\n消费消息:) consumer Consumer( brokers[(localhost, 19492)], group_iddemo-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(test) consumer.start() time.sleep(1) consumer.stop() print(f 收到 {len(received)} 条消息) for msg in received[:5]: print(f offset{msg[offset]}: {msg[value]}) broker.stop() def demo_consumer_group(): 演示消费者组 print(\n * 80) print( 消费者组演示) print( * 80) broker Broker(host0.0.0.0, port19493, data_dir/tmp/mq_demo_c2) broker.create_topic(group-test, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 生产消息 producer Producer(brokers[(localhost, 19493)]) producer.start() for i in range(30): producer.send(group-test, fMsg-{i}, keyfk{i}) producer.flush() producer.stop() # 启动两个消费者同一组 print(\n启动消费者组2个消费者:) results {1: [], 2: []} def make_handler(consumer_id): def handler(msg): results[consumer_id].append(msg) print(f Consumer{consumer_id}: {msg[value]}) return handler consumer1 Consumer( brokers[(localhost, 19493)], group_idshared-group, client_idconsumer-1 ) consumer1.message_handler make_handler(1) consumer1.subscribe(group-test) consumer1.start() consumer2 Consumer( brokers[(localhost, 19493)], group_idshared-group, client_idconsumer-2 ) consumer2.message_handler make_handler(2) consumer2.subscribe(group-test) consumer2.start() time.sleep(2) consumer1.stop() consumer2.stop() print(f\n 消费统计:) print(f Consumer1: {len(results[1])} 条) print(f Consumer2: {len(results[2])} 条) print(f 总计: {len(results[1]) len(results[2])} 条) broker.stop() def demo_offset_management(): 演示 Offset 管理 print(\n * 80) print( Offset 管理演示) print( * 80) broker Broker(host0.0.0.0, port19494, data_dir/tmp/mq_demo_c3) broker.create_topic(offset-demo, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() time.sleep(0.5) # 生产消息 producer Producer(brokers[(localhost, 19494)]) producer.start() for i in range(10): producer.send(offset-demo, fData-{i}) producer.flush() producer.stop() # 第一次消费消费前5条 print(\n第一次消费前5条:) consumer Consumer( brokers[(localhost, 19494)], group_idoffset-group, auto_commitTrue ) first_batch [] consumer.message_handler lambda msg: first_batch.append(msg) consumer.subscribe(offset-demo) consumer.start() time.sleep(1) consumer.stop() print(f 消费了 {len(first_batch)} 条) # 第二次消费应该从 offset5 开始 print(\n第二次消费应从 offset5 开始:) consumer2 Consumer( brokers[(localhost, 19494)], group_idoffset-group, auto_commitTrue ) second_batch [] consumer2.message_handler lambda msg: second_batch.append(msg) consumer2.subscribe(offset-demo) consumer2.start() time.sleep(1) consumer2.stop() print(f 消费了 {len(second_batch)} 条) if second_batch: print(f 第一条: {second_batch[0][value]}) # 回溯消费 print(\n回溯消费从 offset0 开始:) consumer3 Consumer( brokers[(localhost, 19494)], group_idrewind-group, auto_commitTrue ) third_batch [] consumer3.message_handler lambda msg: third_batch.append(msg) consumer3.subscribe(offset-demo) consumer3.seek(offset-demo, 0, 0) # 从头开始 consumer3.start() time.sleep(1) consumer3.stop() print(f 消费了 {len(third_batch)} 条全部) broker.stop() if __name__ __main__: demo_basic_consumption() demo_consumer_group() demo_offset_management()七、测试# tests/test_consumer.py import unittest import time import threading import tempfile from mq.consumer.consumer import Consumer from mq.consumer.offset_store import OffsetStore from mq.consumer.coordinator import GroupCoordinator from mq.consumer.assignor import RangeAssignor, RoundRobinAssignor from mq.broker.broker import Broker from mq.producer.producer import Producer class TestOffsetStore(unittest.TestCase): Offset 存储测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.store OffsetStore(self.tmpdir, auto_commitFalse) def tearDown(self): self.store.close() def test_commit_and_get(self): 测试提交和获取 self.store.commit(group1, topic1, 0, 42) offset self.store.get_offset(group1, topic1, 0) self.assertEqual(offset, 42) def test_default_offset(self): 测试默认 Offset offset self.store.get_offset(unknown, topic, 0) self.assertEqual(offset, 0) class TestGroupCoordinator(unittest.TestCase): 组协调器测试 def setUp(self): self.coordinator GroupCoordinator(test-group) def test_register(self): 测试注册 need_rebalance self.coordinator.register(m1, host1, 9092) self.assertTrue(need_rebalance) # 重复注册不应该触发 Rebalance need_rebalance self.coordinator.register(m1, host1, 9092) self.assertFalse(need_rebalance) def test_unregister(self): 测试注销 self.coordinator.register(m1, host1, 9092) self.coordinator.register(m2, host2, 9092) self.coordinator.unregister(m1) self.assertEqual(len(self.coordinator.members), 1) class TestAssignor(unittest.TestCase): 分区分配测试 def test_range_assignor(self): 测试范围分配 assignor RangeAssignor() members [c1, c2] topics {test: [0, 1, 2]} assignments assignor.assign(members, topics) self.assertEqual(len(assignments), 2) # c1 应该有 2 个分区c2 有 1 个 self.assertEqual(len(assignments[c1]), 2) self.assertEqual(len(assignments[c2]), 1) def test_round_robin(self): 测试轮询分配 assignor RoundRobinAssignor() members [c1, c2, c3] topics {t1: [0, 1], t2: [0, 1]} assignments assignor.assign(members, topics) # 总共 4 个分区3 个消费者 total sum(len(v) for v in assignments.values()) self.assertEqual(total, 4) class TestConsumerIntegration(unittest.TestCase): 消费者集成测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.broker Broker(host0.0.0.0, port19592, 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_consume_messages(self): 测试消费消息 # 先生产 producer Producer(brokers[(localhost, 19592)]) producer.start() for i in range(5): producer.send(test, fmsg-{i}) producer.flush() producer.stop() # 再消费 consumer Consumer( brokers[(localhost, 19592)], group_idtest-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(test) consumer.start() time.sleep(1) consumer.stop() self.assertEqual(len(received), 5) if __name__ __main__: unittest.main()八、总结这一讲我们实现了消费者组的核心机制组件功能OffsetStoreOffset 持久化存储支持自动/手动提交GroupCoordinator组协调器管理成员和触发 RebalancePartitionAssignor分区分配策略Range/RoundRobin/StickyConsumer完整消费者支持组消费和 Offset 管理关键成果✅ 消费者组机制支持水平扩展✅ 自动 Rebalance动态感知成员变化✅ Offset 自动提交断点续传✅ 支持回溯消费seek✅ 三种分区分配策略下一讲我们将实现主从复制与高可用机制让消息队列在节点故障时依然可用。开发之余的小工具推荐处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。