尧图网站建设 
前四讲我们构建了一个功能完整的消息队列——Producer 可以发送消息Broker 可以持久化存储Consumer 可以组消费。但这里有一个致命问题Broker 是单点。如果 Broker 宕机整个消息队列就不可用了。这一讲我们来实现主从复制Leader-Follower Replication让消息队列具备高可用能力。一、复制架构设计1.1 主从复制模型┌─────────────────────────────────────────┐ │ ZooKeeper / 协调服务 │ │ (选主、元数据管理) │ └─────────────────────────────────────────┘ │ ┌───────────┼───────────┐ │ │ │ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │ Leader │◄┤ Follower│◄┤Follower │ │ Broker │ │ Broker │ │ Broker │ │(读写) │ │(只读) │ │(只读) │ └────┬────┘ └─────────┘ └─────────┘ │ ┌────▼────┐ │ 复制日志 │ │ ┌─────┐ │ │ │ ISR │ │ ← In-Sync Replicas │ └─────┘ │ └──────────┘1.2 复制策略┌─────────────────────────────────────────────┐ │ 复制策略对比 │ ├──────────────┬───────────────┬───────────────┤ │ 策略 │ 可靠性 │ 性能 │ ├──────────────┼───────────────┼───────────────┤ │ acks0 │ 最低 │ 最高 │ │ (发完即忘) │ 可能丢消息 │ 无等待 │ ├──────────────┼───────────────┼───────────────┤ │ acks1 │ 中等 │ 较高 │ │ (Leader确认) │ Leader宕机丢 │ 一次网络往返 │ ├──────────────┼───────────────┼───────────────┤ │ acks-1/all │ 最高 │ 较低 │ │ (全部ISR确认) │ 不丢消息 │ 多次网络往返 │ └──────────────┴───────────────┴───────────────┘1.3 数据流Producer │ │ send(msg, acks-1) ▼ Leader Broker │ ├──▶ 写入本地日志 │ ├──▶ 并行复制到所有 Follower │ │ │ ├──▶ Follower 1: 写入完成 → ACK │ ├──▶ Follower 2: 写入完成 → ACK │ └──▶ Follower 3: 写入完成 → ACK │ └──▶ 收到所有 ISR 的 ACK → 返回给 Producer二、复制协议2.1 复制消息定义# mq/replication/protocol.py from dataclasses import dataclass, field from typing import List, Optional from enum import Enum, auto class ReplicaRole(Enum): 副本角色 LEADER auto() FOLLOWER auto() OBSERVER auto() dataclass class ReplicaState: 副本状态 broker_id: str role: ReplicaRole host: str port: int leader_epoch: int 0 high_watermark: int 0 # 已提交的最大 offset log_end_offset: int 0 # 本地日志末尾 offset is_alive: bool True last_heartbeat: float 0.0 dataclass class FetchReplicaRequest: 副本拉取请求 broker_id: str topic: str partition: int fetch_offset: int max_bytes: int 1048576 dataclass class FetchReplicaResponse: 副本拉取响应 messages: List[dict] high_watermark: int log_end_offset: int dataclass class LeaderAndIsr: Leader 和 ISR 信息 leader_id: str leader_epoch: int isr: List[str] # In-Sync Replicas replicas: List[str] # 所有副本三、副本管理器3.1 副本同步管理# mq/replication/replica_manager.py import threading import time import json import logging from typing import Dict, List, Optional, Set from ..protocol.message import Message, MessageType, WireProtocol from ..transport.server import TCPClient from .protocol import * logger logging.getLogger(__name__) class ReplicaManager: 副本管理器 管理 Topic-Partition 的主从复制 def __init__(self, broker_id: str, host: str, port: int, storage_engine): self.broker_id broker_id self.host host self.port port self.storage storage_engine # partition - LeaderAndIsr self.partition_state: Dict[tuple, LeaderAndIsr] {} # 到其他 Broker 的连接 self.connections: Dict[str, TCPClient] {} # 复制线程 self.replication_threads: Dict[tuple, threading.Thread] {} self.lock threading.Lock() self.running False def become_leader(self, topic: str, partition: int, leader_epoch: int, replicas: List[str]): 成为 Leader key (topic, partition) with self.lock: state LeaderAndIsr( leader_idself.broker_id, leader_epochleader_epoch, isr[self.broker_id], replicasreplicas ) self.partition_state[key] state logger.info(fBecame leader for {topic}/{partition} f(epoch{leader_epoch})) def become_follower(self, topic: str, partition: int, leader_id: str, leader_epoch: int): 成为 Follower key (topic, partition) with self.lock: state LeaderAndIsr( leader_idleader_id, leader_epochleader_epoch, isr[], replicas[] ) self.partition_state[key] state # 启动复制线程 if key not in self.replication_threads: thread threading.Thread( targetself._follower_fetch_loop, args(topic, partition, leader_id), daemonTrue ) self.replication_threads[key] thread thread.start() logger.info(fBecame follower for {topic}/{partition}, fleader{leader_id}) def replicate_to_followers(self, topic: str, partition: int, records: List[dict]) - bool: 复制消息到所有 Follower Returns: 是否成功复制到所有 ISR key (topic, partition) with self.lock: state self.partition_state.get(key) if not state or state.leader_id ! self.broker_id: return False followers [r for r in state.isr if r ! self.broker_id] if not followers: return True # 没有 Follower视为成功 # 并行复制到所有 Follower success True for follower_id in followers: try: self._send_to_follower(follower_id, topic, partition, records) except Exception as e: logger.error(fReplicate to {follower_id} failed: {e}) # 从 ISR 中移除 if follower_id in state.isr: state.isr.remove(follower_id) success False return success def _send_to_follower(self, follower_id: str, topic: str, partition: int, records: List[dict]): 发送数据到 Follower client self._get_connection(follower_id) payload json.dumps({ topic: topic, partition: partition, messages: records, leader_epoch: self.partition_state[(topic, partition)].leader_epoch }) msg Message( msg_typeMessageType.PRODUCE_REQUEST, topictopic, valuepayload.encode() ) response client.send(msg) if not response: raise ConnectionError(fNo response from {follower_id}) def _follower_fetch_loop(self, topic: str, partition: int, leader_id: str): Follower 拉取循环 while self.running: try: client self._get_connection(leader_id) # 获取当前 offset current_offset self.storage.get_latest_offset(topic, partition) # 发送拉取请求 payload json.dumps({ topic: topic, partition: partition, offset: current_offset 1, max_bytes: 1048576, is_follower: True }) msg Message( msg_typeMessageType.FETCH_REQUEST, topictopic, valuepayload.encode() ) response client.send(msg) if response: result json.loads(response.value.decode()) messages result.get(messages, []) for msg_data in messages: self.storage.produce( topic, partition, msg_data[value].encode(), msg_data.get(key, ) ) time.sleep(0.1) # 防止空转 except Exception as e: logger.error(fFollower fetch error: {e}) time.sleep(1) # 错误后等待 def _get_connection(self, broker_id: str) - TCPClient: 获取到指定 Broker 的连接 if broker_id not in self.connections: # 从配置获取地址简化实现 host, port self._resolve_broker_address(broker_id) client TCPClient(host, port) client.connect() self.connections[broker_id] client return self.connections[broker_id] def _resolve_broker_address(self, broker_id: str) - tuple: 解析 Broker 地址简化实现 # 实际项目中从配置中心获取 return (localhost, 9092 int(broker_id.split(-)[-1])) def update_isr(self, topic: str, partition: int): 更新 ISR 列表 key (topic, partition) with self.lock: state self.partition_state.get(key) if not state: return # 检查所有副本的状态 alive_followers [] for replica in state.replicas: if replica self.broker_id: continue try: client self._get_connection(replica) # 发送心跳检查 msg Message(msg_typeMessageType.HEARTBEAT) response client.send(msg) if response: alive_followers.append(replica) except: pass # 更新 ISR state.isr [self.broker_id] alive_followers logger.debug(fUpdated ISR for {topic}/{partition}: {state.isr}) def start(self): 启动副本管理器 self.running True logger.info(Replica manager started) def stop(self): 停止副本管理器 self.running False for client in self.connections.values(): client.disconnect() logger.info(Replica manager stopped)四、高可用 Broker4.1 支持复制的 Broker# mq/broker/ha_broker.py import threading import time import json import logging from typing import Dict, List, Optional from ..protocol.message import Message, MessageType from ..transport.server import TCPServer from ..storage.segment_manager import SegmentManager from ..replication.replica_manager import ReplicaManager from ..replication.protocol import ReplicaRole logger logging.getLogger(__name__) class HABroker: 高可用 Broker 支持 - 主从复制 - 自动故障转移 - 多副本存储 def __init__(self, broker_id: str, host: str 0.0.0.0, port: int 9092, data_dir: str ./mq_data, replication_factor: int 2): self.broker_id broker_id self.host host self.port port self.data_dir data_dir self.replication_factor replication_factor # 存储引擎 self.storage SegmentManager(data_dir) # 副本管理器 self.replica_manager ReplicaManager( broker_id, host, port, self.storage ) # 网络服务器 self.server TCPServer(host, port, self._handle_message) # 分区角色映射 self.partition_roles: Dict[tuple, ReplicaRole] {} # 元数据简化硬编码集群信息 self.cluster_nodes { broker-0: (localhost, 29092), broker-1: (localhost, 29093), broker-2: (localhost, 29094), } self.running False def start(self): 启动 Broker self.running True # 启动副本管理器 self.replica_manager.start() # 启动网络服务 self.server.start() logger.info(fHABroker {self.broker_id} started on {self.host}:{self.port}) def stop(self): 停止 Broker self.running False self.replica_manager.stop() self.server.stop() self.storage.close() logger.info(fHABroker {self.broker_id} stopped) def create_topic(self, topic: str, partitions: int 3): 创建 Topic for p in range(partitions): # 分配副本到不同的 Broker replicas self._assign_replicas(topic, p) # 如果是 Leader初始化 if replicas[0] self.broker_id: self.replica_manager.become_leader( topic, p, 0, replicas ) self.partition_roles[(topic, p)] ReplicaRole.LEADER else: # 作为 Follower self.replica_manager.become_follower( topic, p, replicas[0], 0 ) self.partition_roles[(topic, p)] ReplicaRole.FOLLOWER logger.info(fCreated topic {topic}/{p}: frole{self.partition_roles[(topic, p)].name}, freplicas{replicas}) def _assign_replicas(self, topic: str, partition: int) - List[str]: 分配副本简单轮询 nodes list(self.cluster_nodes.keys()) start_idx hash(f{topic}:{partition}) % len(nodes) replicas [] for i in range(min(self.replication_factor, len(nodes))): idx (start_idx i) % len(nodes) replicas.append(nodes[idx]) return replicas def _handle_message(self, msg: Message, sock) - Optional[Message]: 处理消息 if msg.msg_type MessageType.PRODUCE_REQUEST: return self._handle_produce(msg) elif msg.msg_type MessageType.FETCH_REQUEST: return self._handle_fetch(msg) elif msg.msg_type MessageType.HEARTBEAT: return Message(msg_typeMessageType.HEARTBEAT) return None def _handle_produce(self, msg: Message) - Message: 处理生产请求 try: data json.loads(msg.value.decode()) topic data.get(topic, msg.topic) partition data.get(partition, 0) messages data.get(messages, []) # 检查是否是 Leader key (topic, partition) if self.partition_roles.get(key) ! ReplicaRole.LEADER: # 如果不是 Leader返回重定向 leader_id self._find_leader(topic, partition) return Message( msg_typeMessageType.PRODUCE_RESPONSE, valuejson.dumps({ success: False, error: fNOT_LEADER, leader: leader_id }).encode() ) # 写入本地存储 offsets [] for m in messages: value m.get(value, ).encode() key_str m.get(key, ) offset self.storage.append(value, key_str) offsets.append(offset) # 复制到 Follower records [{ offset: offsets[i], key: messages[i].get(key, ), value: messages[i].get(value, ) } for i in range(len(messages))] self.replica_manager.replicate_to_followers( topic, partition, records ) return Message( msg_typeMessageType.PRODUCE_RESPONSE, valuejson.dumps({ success: True, offsets: offsets }).encode() ) except Exception as e: logger.error(fHA produce error: {e}) return Message( msg_typeMessageType.PRODUCE_RESPONSE, valuejson.dumps({ success: False, error: str(e) }).encode() ) def _handle_fetch(self, msg: Message) - Message: 处理拉取请求 try: data json.loads(msg.value.decode()) topic data.get(topic, msg.topic) partition data.get(partition, 0) offset data.get(offset, 0) is_follower data.get(is_follower, False) # Follower 可以从任何副本读取 # 普通消费者只能从 Leader 读取 if not is_follower: key (topic, partition) if self.partition_roles.get(key) ! ReplicaRole.LEADER: leader_id self._find_leader(topic, partition) return Message( msg_typeMessageType.FETCH_RESPONSE, valuejson.dumps({ messages: [], error: NOT_LEADER, leader: leader_id }).encode() ) # 读取消息 messages self.storage.read(offset, 10) message_list [] for msg_offset, key_str, value in messages: message_list.append({ offset: msg_offset, key: key_str, value: value.decode(utf-8, errorsreplace) }) return Message( msg_typeMessageType.FETCH_RESPONSE, valuejson.dumps({ messages: message_list, next_offset: offset len(message_list) }).encode() ) except Exception as e: logger.error(fHA fetch error: {e}) return Message( msg_typeMessageType.FETCH_RESPONSE, valuejson.dumps({ messages: [], error: str(e) }).encode() ) def _find_leader(self, topic: str, partition: int) - Optional[str]: 查找分区的 Leader # 简化实现从副本管理器获取 for (t, p), state in self.replica_manager.partition_state.items(): if t topic and p partition: return state.leader_id return None五、故障转移5.1 自动故障检测与转移# mq/replication/failover.py import threading import time import logging from typing import Dict, Optional, Callable from ..transport.server import TCPClient from ..protocol.message import Message, MessageType logger logging.getLogger(__name__) class FailureDetector: 故障检测器 通过心跳检测节点存活状态 def __init__(self, check_interval_ms: int 3000, failure_threshold: int 3): self.check_interval_ms check_interval_ms self.failure_threshold failure_threshold self.nodes: Dict[str, dict] {} self.failure_counts: Dict[str, int] {} self.on_node_down: Optional[Callable] None self.on_node_up: Optional[Callable] None self.running False self.check_thread: Optional[threading.Thread] None def add_node(self, node_id: str, host: str, port: int): 添加监控节点 self.nodes[node_id] {host: host, port: port, alive: True} self.failure_counts[node_id] 0 logger.info(fAdded node for monitoring: {node_id}) def start(self): 启动检测 self.running True self.check_thread threading.Thread(targetself._check_loop, daemonTrue) self.check_thread.start() logger.info(Failure detector started) def stop(self): 停止检测 self.running False if self.check_thread: self.check_thread.join(timeout2) def _check_loop(self): 检测循环 while self.running: time.sleep(self.check_interval_ms / 1000.0) for node_id, info in self.nodes.items(): alive self._ping(node_id, info[host], info[port]) if alive and not info[alive]: # 节点恢复 logger.info(fNode recovered: {node_id}) info[alive] True self.failure_counts[node_id] 0 if self.on_node_up: self.on_node_up(node_id) elif not alive and info[alive]: # 可能故障 self.failure_counts[node_id] 1 if self.failure_counts[node_id] self.failure_threshold: logger.warning(fNode down detected: {node_id}) info[alive] False if self.on_node_down: self.on_node_down(node_id) def _ping(self, node_id: str, host: str, port: int) - bool: Ping 节点 try: client TCPClient(host, port) client.connect() msg Message(msg_typeMessageType.HEARTBEAT) response client.send(msg) client.disconnect() return response is not None except: return False class FailoverManager: 故障转移管理器 当 Leader 宕机时从 ISR 中选举新的 Leader def __init__(self, replica_managers: Dict[str, ReplicaManager]): self.replica_managers replica_managers self.election_in_progress: Dict[tuple, bool] {} self.lock threading.Lock() def handle_leader_failure(self, topic: str, partition: int, failed_leader: str): 处理 Leader 故障 key (topic, partition) with self.lock: if self.election_in_progress.get(key, False): logger.info(fElection already in progress for {topic}/{partition}) return self.election_in_progress[key] True try: # 从 ISR 中选择新 Leader new_leader self._elect_new_leader(topic, partition, failed_leader) if new_leader: logger.info(fNew leader elected for {topic}/{partition}: {new_leader}) # 通知所有副本 self._broadcast_new_leader(topic, partition, new_leader) else: logger.error(fNo suitable leader found for {topic}/{partition}) finally: with self.lock: self.election_in_progress[key] False def _elect_new_leader(self, topic: str, partition: int, failed_leader: str) - Optional[str]: 选举新 Leader # 收集所有副本的状态 candidates [] for broker_id, rm in self.replica_managers.items(): if broker_id failed_leader: continue key (topic, partition) state rm.partition_state.get(key) if state and broker_id in state.isr: candidates.append(broker_id) # 选择第一个可用的简化实际应该比较 LEO 等 return candidates[0] if candidates else None def _broadcast_new_leader(self, topic: str, partition: int, new_leader: str): 广播新 Leader for broker_id, rm in self.replica_managers.items(): if broker_id new_leader: rm.become_leader(topic, partition, rm.partition_state[(topic, partition)].leader_epoch 1, []) else: rm.become_follower(topic, partition, new_leader, rm.partition_state[(topic, partition)].leader_epoch 1)六、演示# examples/ha_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.ha_broker import HABroker from mq.producer.producer import Producer from mq.consumer.consumer import Consumer from mq.replication.failover import FailureDetector, FailoverManager def demo_replication(): 演示主从复制 print( * 80) print( 主从复制演示) print( * 80) with tempfile.TemporaryDirectory() as tmpdir: # 启动 3 个 Broker print(\n启动 3 节点集群...) brokers {} for i in range(3): broker HABroker( broker_idfbroker-{i}, host0.0.0.0, port29092 i, data_dirf{tmpdir}/broker_{i}, replication_factor2 ) broker.create_topic(ha-test, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() brokers[fbroker-{i}] broker time.sleep(1) # 查看分区角色分配 print(\n分区角色分配:) for bid, broker in brokers.items(): for (topic, partition), role in broker.partition_roles.items(): print(f {bid}: {topic}/{partition} {role.name}) # 生产消息 print(\n生产消息:) producer Producer(brokers[(localhost, 29092)]) producer.start() for i in range(10): offset producer.send(ha-test, fHA Message #{i}) print(f 发送: offset{offset}) producer.flush() producer.stop() # 消费消息 print(\n消费消息:) consumer Consumer( brokers[(localhost, 29092)], group_idha-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(ha-test) consumer.start() time.sleep(1) consumer.stop() print(f 收到 {len(received)} 条消息) # 停止所有 Broker for broker in brokers.values(): broker.stop() def demo_failover(): 演示故障转移 print(\n * 80) print(⚡ 故障转移演示) print( * 80) with tempfile.TemporaryDirectory() as tmpdir: # 启动 3 节点集群 print(\n启动 3 节点集群...) brokers {} for i in range(3): broker HABroker( broker_idfbroker-{i}, host0.0.0.0, port29192 i, data_dirf{tmpdir}/failover_{i}, replication_factor3 ) broker.create_topic(failover-test, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() brokers[fbroker-{i}] broker time.sleep(1) # 查看初始 Leader leader None for bid, broker in brokers.items(): role broker.partition_roles.get((failover-test, 0)) if role and role.name LEADER: leader bid print(f\n初始 Leader: {leader}) # 生产一些消息 print(\n生产消息...) producer Producer(brokers[(localhost, 29192)]) producer.start() for i in range(5): producer.send(failover-test, fPre-failover msg #{i}) producer.flush() producer.stop() # 模拟 Leader 宕机 print(f\n 模拟 {leader} 宕机...) brokers[leader].stop() del brokers[leader] time.sleep(2) # 检查新的 Leader print(\n检查新 Leader:) for bid, broker in brokers.items(): role broker.partition_roles.get((failover-test, 0)) if role: print(f {bid}: {role.name}) # 继续生产消息 print(\n继续生产消息...) # 找到新的 Leader 端口 new_leader_port 29192 for bid, broker in brokers.items(): role broker.partition_roles.get((failover-test, 0)) if role and role.name LEADER: new_leader_port broker.port break producer2 Producer(brokers[(localhost, new_leader_port)]) producer2.start() for i in range(5): producer2.send(failover-test, fPost-failover msg #{i}) producer2.flush() producer2.stop() # 消费所有消息 print(\n消费所有消息验证数据完整性:) consumer Consumer( brokers[(localhost, new_leader_port)], group_idfailover-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(failover-test) consumer.start() time.sleep(1) consumer.stop() print(f 共收到 {len(received)} 条消息) for msg in received: print(f offset{msg[offset]}: {msg[value]}) # 停止剩余 Broker for broker in brokers.values(): broker.stop() if __name__ __main__: demo_replication() demo_failover()七、测试# tests/test_ha.py import unittest import time import threading import tempfile from mq.broker.ha_broker import HABroker from mq.replication.replica_manager import ReplicaManager from mq.replication.failover import FailureDetector, FailoverManager from mq.replication.protocol import ReplicaRole from mq.producer.producer import Producer from mq.consumer.consumer import Consumer class TestReplicaManager(unittest.TestCase): 副本管理器测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.rm ReplicaManager(broker-0, localhost, 29092, None) def test_become_leader(self): 测试成为 Leader self.rm.become_leader(test, 0, 0, [broker-0, broker-1]) state self.rm.partition_state.get((test, 0)) self.assertIsNotNone(state) self.assertEqual(state.leader_id, broker-0) self.assertIn(broker-0, state.isr) class TestFailureDetector(unittest.TestCase): 故障检测器测试 def test_detection(self): 测试故障检测 detector FailureDetector(check_interval_ms100, failure_threshold2) down_nodes [] detector.on_node_down lambda n: down_nodes.append(n) # 添加一个不存在的节点 detector.add_node(fake-node, 192.0.2.1, 9999) detector.start() time.sleep(0.3) detector.stop() self.assertIn(fake-node, down_nodes) class TestHABrokerIntegration(unittest.TestCase): HA Broker 集成测试 def setUp(self): self.tmpdir tempfile.mkdtemp() # 启动 2 节点集群 self.brokers {} for i in range(2): broker HABroker( broker_idfbroker-{i}, host0.0.0.0, port29292 i, data_dirf{self.tmpdir}/broker_{i}, replication_factor2 ) broker.create_topic(test, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() self.brokers[fbroker-{i}] broker time.sleep(0.5) def tearDown(self): for broker in self.brokers.values(): broker.stop() def test_produce_and_consume(self): 测试生产和消费 # 找到 Leader leader_port None for broker in self.brokers.values(): role broker.partition_roles.get((test, 0)) if role and role.name LEADER: leader_port broker.port break self.assertIsNotNone(leader_port) # 生产消息 producer Producer(brokers[(localhost, leader_port)]) producer.start() producer.send(test, HA integration test) producer.flush() producer.stop() # 消费消息 consumer Consumer( brokers[(localhost, leader_port)], group_idtest-group ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(test) consumer.start() time.sleep(0.5) consumer.stop() self.assertEqual(len(received), 1) if __name__ __main__: unittest.main()八、总结这一讲我们实现了主从复制和高可用机制组件功能ReplicaManager副本管理Leader/Follower 角色切换HABroker高可用 Broker支持读写分离FailureDetector故障检测心跳超时判断FailoverManager故障转移自动选举新 Leader关键成果✅ 主从复制数据多副本存储✅ 自动故障检测和转移✅ 读写分离Leader 读写Follower 只读✅ ISR 机制保证数据一致性✅ 三种 ACK 级别0/1/-1下一讲我们将实现顺序消息与事务消息解决分布式场景下的消息有序性和原子性问题。开发之余的小工具推荐处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。 第5讲：主从复制与高可用机制
前四讲我们构建了一个功能完整的消息队列——Producer 可以发送消息Broker 可以持久化存储Consumer 可以组消费。但这里有一个致命问题Broker 是单点。如果 Broker 宕机整个消息队列就不可用了。这一讲我们来实现主从复制Leader-Follower Replication让消息队列具备高可用能力。一、复制架构设计1.1 主从复制模型┌─────────────────────────────────────────┐ │ ZooKeeper / 协调服务 │ │ (选主、元数据管理) │ └─────────────────────────────────────────┘ │ ┌───────────┼───────────┐ │ │ │ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │ Leader │◄┤ Follower│◄┤Follower │ │ Broker │ │ Broker │ │ Broker │ │(读写) │ │(只读) │ │(只读) │ └────┬────┘ └─────────┘ └─────────┘ │ ┌────▼────┐ │ 复制日志 │ │ ┌─────┐ │ │ │ ISR │ │ ← In-Sync Replicas │ └─────┘ │ └──────────┘1.2 复制策略┌─────────────────────────────────────────────┐ │ 复制策略对比 │ ├──────────────┬───────────────┬───────────────┤ │ 策略 │ 可靠性 │ 性能 │ ├──────────────┼───────────────┼───────────────┤ │ acks0 │ 最低 │ 最高 │ │ (发完即忘) │ 可能丢消息 │ 无等待 │ ├──────────────┼───────────────┼───────────────┤ │ acks1 │ 中等 │ 较高 │ │ (Leader确认) │ Leader宕机丢 │ 一次网络往返 │ ├──────────────┼───────────────┼───────────────┤ │ acks-1/all │ 最高 │ 较低 │ │ (全部ISR确认) │ 不丢消息 │ 多次网络往返 │ └──────────────┴───────────────┴───────────────┘1.3 数据流Producer │ │ send(msg, acks-1) ▼ Leader Broker │ ├──▶ 写入本地日志 │ ├──▶ 并行复制到所有 Follower │ │ │ ├──▶ Follower 1: 写入完成 → ACK │ ├──▶ Follower 2: 写入完成 → ACK │ └──▶ Follower 3: 写入完成 → ACK │ └──▶ 收到所有 ISR 的 ACK → 返回给 Producer二、复制协议2.1 复制消息定义# mq/replication/protocol.py from dataclasses import dataclass, field from typing import List, Optional from enum import Enum, auto class ReplicaRole(Enum): 副本角色 LEADER auto() FOLLOWER auto() OBSERVER auto() dataclass class ReplicaState: 副本状态 broker_id: str role: ReplicaRole host: str port: int leader_epoch: int 0 high_watermark: int 0 # 已提交的最大 offset log_end_offset: int 0 # 本地日志末尾 offset is_alive: bool True last_heartbeat: float 0.0 dataclass class FetchReplicaRequest: 副本拉取请求 broker_id: str topic: str partition: int fetch_offset: int max_bytes: int 1048576 dataclass class FetchReplicaResponse: 副本拉取响应 messages: List[dict] high_watermark: int log_end_offset: int dataclass class LeaderAndIsr: Leader 和 ISR 信息 leader_id: str leader_epoch: int isr: List[str] # In-Sync Replicas replicas: List[str] # 所有副本三、副本管理器3.1 副本同步管理# mq/replication/replica_manager.py import threading import time import json import logging from typing import Dict, List, Optional, Set from ..protocol.message import Message, MessageType, WireProtocol from ..transport.server import TCPClient from .protocol import * logger logging.getLogger(__name__) class ReplicaManager: 副本管理器 管理 Topic-Partition 的主从复制 def __init__(self, broker_id: str, host: str, port: int, storage_engine): self.broker_id broker_id self.host host self.port port self.storage storage_engine # partition - LeaderAndIsr self.partition_state: Dict[tuple, LeaderAndIsr] {} # 到其他 Broker 的连接 self.connections: Dict[str, TCPClient] {} # 复制线程 self.replication_threads: Dict[tuple, threading.Thread] {} self.lock threading.Lock() self.running False def become_leader(self, topic: str, partition: int, leader_epoch: int, replicas: List[str]): 成为 Leader key (topic, partition) with self.lock: state LeaderAndIsr( leader_idself.broker_id, leader_epochleader_epoch, isr[self.broker_id], replicasreplicas ) self.partition_state[key] state logger.info(fBecame leader for {topic}/{partition} f(epoch{leader_epoch})) def become_follower(self, topic: str, partition: int, leader_id: str, leader_epoch: int): 成为 Follower key (topic, partition) with self.lock: state LeaderAndIsr( leader_idleader_id, leader_epochleader_epoch, isr[], replicas[] ) self.partition_state[key] state # 启动复制线程 if key not in self.replication_threads: thread threading.Thread( targetself._follower_fetch_loop, args(topic, partition, leader_id), daemonTrue ) self.replication_threads[key] thread thread.start() logger.info(fBecame follower for {topic}/{partition}, fleader{leader_id}) def replicate_to_followers(self, topic: str, partition: int, records: List[dict]) - bool: 复制消息到所有 Follower Returns: 是否成功复制到所有 ISR key (topic, partition) with self.lock: state self.partition_state.get(key) if not state or state.leader_id ! self.broker_id: return False followers [r for r in state.isr if r ! self.broker_id] if not followers: return True # 没有 Follower视为成功 # 并行复制到所有 Follower success True for follower_id in followers: try: self._send_to_follower(follower_id, topic, partition, records) except Exception as e: logger.error(fReplicate to {follower_id} failed: {e}) # 从 ISR 中移除 if follower_id in state.isr: state.isr.remove(follower_id) success False return success def _send_to_follower(self, follower_id: str, topic: str, partition: int, records: List[dict]): 发送数据到 Follower client self._get_connection(follower_id) payload json.dumps({ topic: topic, partition: partition, messages: records, leader_epoch: self.partition_state[(topic, partition)].leader_epoch }) msg Message( msg_typeMessageType.PRODUCE_REQUEST, topictopic, valuepayload.encode() ) response client.send(msg) if not response: raise ConnectionError(fNo response from {follower_id}) def _follower_fetch_loop(self, topic: str, partition: int, leader_id: str): Follower 拉取循环 while self.running: try: client self._get_connection(leader_id) # 获取当前 offset current_offset self.storage.get_latest_offset(topic, partition) # 发送拉取请求 payload json.dumps({ topic: topic, partition: partition, offset: current_offset 1, max_bytes: 1048576, is_follower: True }) msg Message( msg_typeMessageType.FETCH_REQUEST, topictopic, valuepayload.encode() ) response client.send(msg) if response: result json.loads(response.value.decode()) messages result.get(messages, []) for msg_data in messages: self.storage.produce( topic, partition, msg_data[value].encode(), msg_data.get(key, ) ) time.sleep(0.1) # 防止空转 except Exception as e: logger.error(fFollower fetch error: {e}) time.sleep(1) # 错误后等待 def _get_connection(self, broker_id: str) - TCPClient: 获取到指定 Broker 的连接 if broker_id not in self.connections: # 从配置获取地址简化实现 host, port self._resolve_broker_address(broker_id) client TCPClient(host, port) client.connect() self.connections[broker_id] client return self.connections[broker_id] def _resolve_broker_address(self, broker_id: str) - tuple: 解析 Broker 地址简化实现 # 实际项目中从配置中心获取 return (localhost, 9092 int(broker_id.split(-)[-1])) def update_isr(self, topic: str, partition: int): 更新 ISR 列表 key (topic, partition) with self.lock: state self.partition_state.get(key) if not state: return # 检查所有副本的状态 alive_followers [] for replica in state.replicas: if replica self.broker_id: continue try: client self._get_connection(replica) # 发送心跳检查 msg Message(msg_typeMessageType.HEARTBEAT) response client.send(msg) if response: alive_followers.append(replica) except: pass # 更新 ISR state.isr [self.broker_id] alive_followers logger.debug(fUpdated ISR for {topic}/{partition}: {state.isr}) def start(self): 启动副本管理器 self.running True logger.info(Replica manager started) def stop(self): 停止副本管理器 self.running False for client in self.connections.values(): client.disconnect() logger.info(Replica manager stopped)四、高可用 Broker4.1 支持复制的 Broker# mq/broker/ha_broker.py import threading import time import json import logging from typing import Dict, List, Optional from ..protocol.message import Message, MessageType from ..transport.server import TCPServer from ..storage.segment_manager import SegmentManager from ..replication.replica_manager import ReplicaManager from ..replication.protocol import ReplicaRole logger logging.getLogger(__name__) class HABroker: 高可用 Broker 支持 - 主从复制 - 自动故障转移 - 多副本存储 def __init__(self, broker_id: str, host: str 0.0.0.0, port: int 9092, data_dir: str ./mq_data, replication_factor: int 2): self.broker_id broker_id self.host host self.port port self.data_dir data_dir self.replication_factor replication_factor # 存储引擎 self.storage SegmentManager(data_dir) # 副本管理器 self.replica_manager ReplicaManager( broker_id, host, port, self.storage ) # 网络服务器 self.server TCPServer(host, port, self._handle_message) # 分区角色映射 self.partition_roles: Dict[tuple, ReplicaRole] {} # 元数据简化硬编码集群信息 self.cluster_nodes { broker-0: (localhost, 29092), broker-1: (localhost, 29093), broker-2: (localhost, 29094), } self.running False def start(self): 启动 Broker self.running True # 启动副本管理器 self.replica_manager.start() # 启动网络服务 self.server.start() logger.info(fHABroker {self.broker_id} started on {self.host}:{self.port}) def stop(self): 停止 Broker self.running False self.replica_manager.stop() self.server.stop() self.storage.close() logger.info(fHABroker {self.broker_id} stopped) def create_topic(self, topic: str, partitions: int 3): 创建 Topic for p in range(partitions): # 分配副本到不同的 Broker replicas self._assign_replicas(topic, p) # 如果是 Leader初始化 if replicas[0] self.broker_id: self.replica_manager.become_leader( topic, p, 0, replicas ) self.partition_roles[(topic, p)] ReplicaRole.LEADER else: # 作为 Follower self.replica_manager.become_follower( topic, p, replicas[0], 0 ) self.partition_roles[(topic, p)] ReplicaRole.FOLLOWER logger.info(fCreated topic {topic}/{p}: frole{self.partition_roles[(topic, p)].name}, freplicas{replicas}) def _assign_replicas(self, topic: str, partition: int) - List[str]: 分配副本简单轮询 nodes list(self.cluster_nodes.keys()) start_idx hash(f{topic}:{partition}) % len(nodes) replicas [] for i in range(min(self.replication_factor, len(nodes))): idx (start_idx i) % len(nodes) replicas.append(nodes[idx]) return replicas def _handle_message(self, msg: Message, sock) - Optional[Message]: 处理消息 if msg.msg_type MessageType.PRODUCE_REQUEST: return self._handle_produce(msg) elif msg.msg_type MessageType.FETCH_REQUEST: return self._handle_fetch(msg) elif msg.msg_type MessageType.HEARTBEAT: return Message(msg_typeMessageType.HEARTBEAT) return None def _handle_produce(self, msg: Message) - Message: 处理生产请求 try: data json.loads(msg.value.decode()) topic data.get(topic, msg.topic) partition data.get(partition, 0) messages data.get(messages, []) # 检查是否是 Leader key (topic, partition) if self.partition_roles.get(key) ! ReplicaRole.LEADER: # 如果不是 Leader返回重定向 leader_id self._find_leader(topic, partition) return Message( msg_typeMessageType.PRODUCE_RESPONSE, valuejson.dumps({ success: False, error: fNOT_LEADER, leader: leader_id }).encode() ) # 写入本地存储 offsets [] for m in messages: value m.get(value, ).encode() key_str m.get(key, ) offset self.storage.append(value, key_str) offsets.append(offset) # 复制到 Follower records [{ offset: offsets[i], key: messages[i].get(key, ), value: messages[i].get(value, ) } for i in range(len(messages))] self.replica_manager.replicate_to_followers( topic, partition, records ) return Message( msg_typeMessageType.PRODUCE_RESPONSE, valuejson.dumps({ success: True, offsets: offsets }).encode() ) except Exception as e: logger.error(fHA produce error: {e}) return Message( msg_typeMessageType.PRODUCE_RESPONSE, valuejson.dumps({ success: False, error: str(e) }).encode() ) def _handle_fetch(self, msg: Message) - Message: 处理拉取请求 try: data json.loads(msg.value.decode()) topic data.get(topic, msg.topic) partition data.get(partition, 0) offset data.get(offset, 0) is_follower data.get(is_follower, False) # Follower 可以从任何副本读取 # 普通消费者只能从 Leader 读取 if not is_follower: key (topic, partition) if self.partition_roles.get(key) ! ReplicaRole.LEADER: leader_id self._find_leader(topic, partition) return Message( msg_typeMessageType.FETCH_RESPONSE, valuejson.dumps({ messages: [], error: NOT_LEADER, leader: leader_id }).encode() ) # 读取消息 messages self.storage.read(offset, 10) message_list [] for msg_offset, key_str, value in messages: message_list.append({ offset: msg_offset, key: key_str, value: value.decode(utf-8, errorsreplace) }) return Message( msg_typeMessageType.FETCH_RESPONSE, valuejson.dumps({ messages: message_list, next_offset: offset len(message_list) }).encode() ) except Exception as e: logger.error(fHA fetch error: {e}) return Message( msg_typeMessageType.FETCH_RESPONSE, valuejson.dumps({ messages: [], error: str(e) }).encode() ) def _find_leader(self, topic: str, partition: int) - Optional[str]: 查找分区的 Leader # 简化实现从副本管理器获取 for (t, p), state in self.replica_manager.partition_state.items(): if t topic and p partition: return state.leader_id return None五、故障转移5.1 自动故障检测与转移# mq/replication/failover.py import threading import time import logging from typing import Dict, Optional, Callable from ..transport.server import TCPClient from ..protocol.message import Message, MessageType logger logging.getLogger(__name__) class FailureDetector: 故障检测器 通过心跳检测节点存活状态 def __init__(self, check_interval_ms: int 3000, failure_threshold: int 3): self.check_interval_ms check_interval_ms self.failure_threshold failure_threshold self.nodes: Dict[str, dict] {} self.failure_counts: Dict[str, int] {} self.on_node_down: Optional[Callable] None self.on_node_up: Optional[Callable] None self.running False self.check_thread: Optional[threading.Thread] None def add_node(self, node_id: str, host: str, port: int): 添加监控节点 self.nodes[node_id] {host: host, port: port, alive: True} self.failure_counts[node_id] 0 logger.info(fAdded node for monitoring: {node_id}) def start(self): 启动检测 self.running True self.check_thread threading.Thread(targetself._check_loop, daemonTrue) self.check_thread.start() logger.info(Failure detector started) def stop(self): 停止检测 self.running False if self.check_thread: self.check_thread.join(timeout2) def _check_loop(self): 检测循环 while self.running: time.sleep(self.check_interval_ms / 1000.0) for node_id, info in self.nodes.items(): alive self._ping(node_id, info[host], info[port]) if alive and not info[alive]: # 节点恢复 logger.info(fNode recovered: {node_id}) info[alive] True self.failure_counts[node_id] 0 if self.on_node_up: self.on_node_up(node_id) elif not alive and info[alive]: # 可能故障 self.failure_counts[node_id] 1 if self.failure_counts[node_id] self.failure_threshold: logger.warning(fNode down detected: {node_id}) info[alive] False if self.on_node_down: self.on_node_down(node_id) def _ping(self, node_id: str, host: str, port: int) - bool: Ping 节点 try: client TCPClient(host, port) client.connect() msg Message(msg_typeMessageType.HEARTBEAT) response client.send(msg) client.disconnect() return response is not None except: return False class FailoverManager: 故障转移管理器 当 Leader 宕机时从 ISR 中选举新的 Leader def __init__(self, replica_managers: Dict[str, ReplicaManager]): self.replica_managers replica_managers self.election_in_progress: Dict[tuple, bool] {} self.lock threading.Lock() def handle_leader_failure(self, topic: str, partition: int, failed_leader: str): 处理 Leader 故障 key (topic, partition) with self.lock: if self.election_in_progress.get(key, False): logger.info(fElection already in progress for {topic}/{partition}) return self.election_in_progress[key] True try: # 从 ISR 中选择新 Leader new_leader self._elect_new_leader(topic, partition, failed_leader) if new_leader: logger.info(fNew leader elected for {topic}/{partition}: {new_leader}) # 通知所有副本 self._broadcast_new_leader(topic, partition, new_leader) else: logger.error(fNo suitable leader found for {topic}/{partition}) finally: with self.lock: self.election_in_progress[key] False def _elect_new_leader(self, topic: str, partition: int, failed_leader: str) - Optional[str]: 选举新 Leader # 收集所有副本的状态 candidates [] for broker_id, rm in self.replica_managers.items(): if broker_id failed_leader: continue key (topic, partition) state rm.partition_state.get(key) if state and broker_id in state.isr: candidates.append(broker_id) # 选择第一个可用的简化实际应该比较 LEO 等 return candidates[0] if candidates else None def _broadcast_new_leader(self, topic: str, partition: int, new_leader: str): 广播新 Leader for broker_id, rm in self.replica_managers.items(): if broker_id new_leader: rm.become_leader(topic, partition, rm.partition_state[(topic, partition)].leader_epoch 1, []) else: rm.become_follower(topic, partition, new_leader, rm.partition_state[(topic, partition)].leader_epoch 1)六、演示# examples/ha_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.ha_broker import HABroker from mq.producer.producer import Producer from mq.consumer.consumer import Consumer from mq.replication.failover import FailureDetector, FailoverManager def demo_replication(): 演示主从复制 print( * 80) print( 主从复制演示) print( * 80) with tempfile.TemporaryDirectory() as tmpdir: # 启动 3 个 Broker print(\n启动 3 节点集群...) brokers {} for i in range(3): broker HABroker( broker_idfbroker-{i}, host0.0.0.0, port29092 i, data_dirf{tmpdir}/broker_{i}, replication_factor2 ) broker.create_topic(ha-test, partitions3) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() brokers[fbroker-{i}] broker time.sleep(1) # 查看分区角色分配 print(\n分区角色分配:) for bid, broker in brokers.items(): for (topic, partition), role in broker.partition_roles.items(): print(f {bid}: {topic}/{partition} {role.name}) # 生产消息 print(\n生产消息:) producer Producer(brokers[(localhost, 29092)]) producer.start() for i in range(10): offset producer.send(ha-test, fHA Message #{i}) print(f 发送: offset{offset}) producer.flush() producer.stop() # 消费消息 print(\n消费消息:) consumer Consumer( brokers[(localhost, 29092)], group_idha-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(ha-test) consumer.start() time.sleep(1) consumer.stop() print(f 收到 {len(received)} 条消息) # 停止所有 Broker for broker in brokers.values(): broker.stop() def demo_failover(): 演示故障转移 print(\n * 80) print(⚡ 故障转移演示) print( * 80) with tempfile.TemporaryDirectory() as tmpdir: # 启动 3 节点集群 print(\n启动 3 节点集群...) brokers {} for i in range(3): broker HABroker( broker_idfbroker-{i}, host0.0.0.0, port29192 i, data_dirf{tmpdir}/failover_{i}, replication_factor3 ) broker.create_topic(failover-test, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() brokers[fbroker-{i}] broker time.sleep(1) # 查看初始 Leader leader None for bid, broker in brokers.items(): role broker.partition_roles.get((failover-test, 0)) if role and role.name LEADER: leader bid print(f\n初始 Leader: {leader}) # 生产一些消息 print(\n生产消息...) producer Producer(brokers[(localhost, 29192)]) producer.start() for i in range(5): producer.send(failover-test, fPre-failover msg #{i}) producer.flush() producer.stop() # 模拟 Leader 宕机 print(f\n 模拟 {leader} 宕机...) brokers[leader].stop() del brokers[leader] time.sleep(2) # 检查新的 Leader print(\n检查新 Leader:) for bid, broker in brokers.items(): role broker.partition_roles.get((failover-test, 0)) if role: print(f {bid}: {role.name}) # 继续生产消息 print(\n继续生产消息...) # 找到新的 Leader 端口 new_leader_port 29192 for bid, broker in brokers.items(): role broker.partition_roles.get((failover-test, 0)) if role and role.name LEADER: new_leader_port broker.port break producer2 Producer(brokers[(localhost, new_leader_port)]) producer2.start() for i in range(5): producer2.send(failover-test, fPost-failover msg #{i}) producer2.flush() producer2.stop() # 消费所有消息 print(\n消费所有消息验证数据完整性:) consumer Consumer( brokers[(localhost, new_leader_port)], group_idfailover-group, auto_commitTrue ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(failover-test) consumer.start() time.sleep(1) consumer.stop() print(f 共收到 {len(received)} 条消息) for msg in received: print(f offset{msg[offset]}: {msg[value]}) # 停止剩余 Broker for broker in brokers.values(): broker.stop() if __name__ __main__: demo_replication() demo_failover()七、测试# tests/test_ha.py import unittest import time import threading import tempfile from mq.broker.ha_broker import HABroker from mq.replication.replica_manager import ReplicaManager from mq.replication.failover import FailureDetector, FailoverManager from mq.replication.protocol import ReplicaRole from mq.producer.producer import Producer from mq.consumer.consumer import Consumer class TestReplicaManager(unittest.TestCase): 副本管理器测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.rm ReplicaManager(broker-0, localhost, 29092, None) def test_become_leader(self): 测试成为 Leader self.rm.become_leader(test, 0, 0, [broker-0, broker-1]) state self.rm.partition_state.get((test, 0)) self.assertIsNotNone(state) self.assertEqual(state.leader_id, broker-0) self.assertIn(broker-0, state.isr) class TestFailureDetector(unittest.TestCase): 故障检测器测试 def test_detection(self): 测试故障检测 detector FailureDetector(check_interval_ms100, failure_threshold2) down_nodes [] detector.on_node_down lambda n: down_nodes.append(n) # 添加一个不存在的节点 detector.add_node(fake-node, 192.0.2.1, 9999) detector.start() time.sleep(0.3) detector.stop() self.assertIn(fake-node, down_nodes) class TestHABrokerIntegration(unittest.TestCase): HA Broker 集成测试 def setUp(self): self.tmpdir tempfile.mkdtemp() # 启动 2 节点集群 self.brokers {} for i in range(2): broker HABroker( broker_idfbroker-{i}, host0.0.0.0, port29292 i, data_dirf{self.tmpdir}/broker_{i}, replication_factor2 ) broker.create_topic(test, partitions1) bt threading.Thread(targetbroker.start, daemonTrue) bt.start() self.brokers[fbroker-{i}] broker time.sleep(0.5) def tearDown(self): for broker in self.brokers.values(): broker.stop() def test_produce_and_consume(self): 测试生产和消费 # 找到 Leader leader_port None for broker in self.brokers.values(): role broker.partition_roles.get((test, 0)) if role and role.name LEADER: leader_port broker.port break self.assertIsNotNone(leader_port) # 生产消息 producer Producer(brokers[(localhost, leader_port)]) producer.start() producer.send(test, HA integration test) producer.flush() producer.stop() # 消费消息 consumer Consumer( brokers[(localhost, leader_port)], group_idtest-group ) received [] consumer.message_handler lambda msg: received.append(msg) consumer.subscribe(test) consumer.start() time.sleep(0.5) consumer.stop() self.assertEqual(len(received), 1) if __name__ __main__: unittest.main()八、总结这一讲我们实现了主从复制和高可用机制组件功能ReplicaManager副本管理Leader/Follower 角色切换HABroker高可用 Broker支持读写分离FailureDetector故障检测心跳超时判断FailoverManager故障转移自动选举新 Leader关键成果✅ 主从复制数据多副本存储✅ 自动故障检测和转移✅ 读写分离Leader 读写Follower 只读✅ ISR 机制保证数据一致性✅ 三种 ACK 级别0/1/-1下一讲我们将实现顺序消息与事务消息解决分布式场景下的消息有序性和原子性问题。开发之余的小工具推荐处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。