前三講我們實(shí)現(xiàn)了消息的生產(chǎn)和存儲(chǔ)。現(xiàn)在Producer 可以高效地發(fā)送消息Broker 可以可靠地存儲(chǔ)消息。但消息隊(duì)列的核心價(jià)值在于消費(fèi)——多個(gè)消費(fèi)者如何協(xié)同工作如何保證消息不重復(fù)、不丟失如何動(dòng)態(tài)擴(kuò)縮容這一講我們來實(shí)現(xiàn)消費(fèi)者組機(jī)制這是 Kafka 等成熟消息隊(duì)列最核心的設(shè)計(jì)之一。一、消費(fèi)者組架構(gòu)1.1 什么是消費(fèi)者組Topic: orders (3個(gè)分區(qū)) ┌──────────┐ ┌──────────┐ ┌──────────┐ │Partition0│ │Partition1│ │Partition2│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ └──────┬─────┴──────┬─────┘ │ │ ┌──────▼─────┐ ┌───▼──────┐ │ Consumer A │ │ Consumer B│ ← 同一個(gè)消費(fèi)者組 └────────────┘ └───────────┘ 每個(gè)分區(qū)只能被組內(nèi)一個(gè)消費(fèi)者消費(fèi) 消費(fèi)者數(shù) ≤ 分區(qū)數(shù)1.2 核心概念概念說明Consumer Group?消費(fèi)者組組內(nèi)消費(fèi)者共同消費(fèi)一個(gè) TopicGroup Coordinator?組協(xié)調(diào)器負(fù)責(zé)管理組成員和分配分區(qū)Rebalance?重平衡組成員變化時(shí)重新分配分區(qū)Offset?消費(fèi)偏移量記錄每個(gè)分區(qū)已消費(fèi)的位置Commit?提交 Offset標(biāo)記消息已被消費(fèi)1.3 消費(fèi)流程Consumer 啟動(dòng) │ ├──? 向 Coordinator 注冊(cè) │ ├──? JoinGroup加入消費(fèi)者組 │ ├──? SyncGroup獲取分區(qū)分配 │ ├──? 開始消費(fèi)從上次提交的 Offset 開始 │ │ │ ├──? 拉取消息 │ ├──? 處理消息 │ └──? 提交 Offset │ └──? 發(fā)生 Rebalance 時(shí)重新分配二、Offset 存儲(chǔ)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 元數(shù)據(jù) offset: int metadata: str timestamp: float 0.0 def __post_init__(self): if not self.timestamp: self.timestamp time.time() class OffsetStore: Offset 存儲(chǔ) 支持 - 自動(dòng)提交 - 手動(dòng)提交 - 持久化到磁盤 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) # 恢復(fù)已保存的 Offset self._recover() # 啟動(dòng)自動(dòng)提交 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): 從磁盤恢復(fù) 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): 啟動(dòng)自動(dòng)提交 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): 關(guān)閉存儲(chǔ) self._save()三、消費(fèi)者組協(xié)調(diào)器3.1 組協(xié)調(diào)器# 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): 成員狀態(tài) JOINING auto() # 正在加入 SYNCING auto() # 正在同步 STABLE auto() # 穩(wěn)定運(yùn)行 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: 消費(fèi)者組協(xié)調(diào)器 職責(zé) 1. 管理組成員 2. 觸發(fā) Rebalance 3. 分配分區(qū) 4. 檢測(cè)心跳超時(shí) 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 代數(shù) self.leader_id: Optional[str] None self.lock threading.Lock() # 分區(qū)分配策略 self.assignment_strategy RangeAssignor() # 回調(diào) self.on_rebalance: Optional[Callable] None # 心跳檢測(cè)線程 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: 注冊(cè)消費(fèi)者成員 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 # 第一個(gè)成員成為 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: 是否需要進(jìn)行 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 # 未知成員觸發(fā) Rebalance def trigger_rebalance(self, subscribed_topics: Dict[str, List[int]]): 觸發(fā)重平衡 Args: subscribed_topics: 訂閱的 topic 及其分區(qū)列表 with self.lock: self.generation 1 logger.info(fRebalance triggered (generation {self.generation})) # 收集活躍成員 active_members self._get_active_members() # 分配分區(qū) 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 # 通知回調(diào) 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): 心跳檢測(cè) 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): 停止協(xié)調(diào)器 self.running False四、分區(qū)分配策略4.1 多種分配算法# mq/consumer/assignor.py from typing import Dict, List, Tuple from collections import defaultdict class PartitionAssignor: 分區(qū)分配器基類 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: raise NotImplementedError class RangeAssignor(PartitionAssignor): 范圍分配 按 Topic 依次分配每個(gè)消費(fèi)者獲得連續(xù)的一段分區(qū) 優(yōu)點(diǎn)實(shí)現(xiàn)簡(jiǎn)單 缺點(diǎn)可能導(dǎo)致分配不均 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) # 計(jì)算每個(gè)消費(fèi)者應(yīng)得的分區(qū)數(shù) partitions_per_member num_partitions // num_members remainder num_partitions % num_members start 0 for i, member in enumerate(members): # 前 remainder 個(gè)消費(fèi)者多分一個(gè)分區(qū) 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): 輪詢分配 將所有分區(qū)排序后輪流分配給消費(fèi)者 優(yōu)點(diǎn)分配均勻 缺點(diǎn)分區(qū)數(shù)遠(yuǎn)大于消費(fèi)者數(shù)時(shí)效果更好 def assign(self, members: List[str], topics: Dict[str, List[int]]) - Dict[str, List[Tuple[str, int]]]: assignments defaultdict(list) # 收集所有分區(qū) 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): 粘性分配 盡量保持已有的分配不變只在必要時(shí)調(diào)整 優(yōu)點(diǎn)減少 Rebalance 時(shí)的分區(qū)移動(dòng) 缺點(diǎn)實(shí)現(xiàn)復(fù)雜 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)) # 優(yōu)先保留已有分配 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) # 分配剩余分區(qū) remaining all_partitions - assigned for i, tp in enumerate(sorted(remaining)): member members[i % len(members)] assignments[member].append(tp) return dict(assignments)五、消費(fèi)者實(shí)現(xiàn)5.1 完整消費(fèi)者# 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: 消費(fèi)者 支持 - 消費(fèi)者組 - 自動(dòng)/手動(dòng)提交 Offset - Rebalance 監(jiān)聽 - 批量拉取 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 # 網(wǎng)絡(luò)連接 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 ) # 組協(xié)調(diào) 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)] # 消費(fèi)狀態(tài) self.running False self.paused False # 消息處理回調(diào) self.message_handler: Optional[Callable] None # Rebalance 監(jiān)聽器 self.rebalance_listeners: List[Callable] [] # 統(tǒng)計(jì) self.metrics { fetched: 0, committed: 0, rebalances: 0 } def subscribe(self, topic: str, partitions: List[int] None): 訂閱主題 Args: topic: 主題名 partitions: 分區(qū)列表None 表示所有分區(qū) if partitions is None: # 簡(jiǎn)化假設(shè)每個(gè) topic 有 3 個(gè)分區(qū) partitions [0, 1, 2] self.subscriptions[topic] partitions logger.info(fSubscribed to {topic}: partitions{partitions}) # 注冊(cè)到協(xié)調(diào)器 self.coordinator.register( self.client_id, self.client.host, self.client.port ) def start(self): 啟動(dòng)消費(fèi) self.running True # 連接到 Broker self.client.connect() # 注冊(cè) Rebalance 回調(diào) self.coordinator.on_rebalance self._on_rebalance # 觸發(fā)初始 Rebalance self._do_rebalance() # 啟動(dòng)消費(fèi)循環(huán) self.consumer_thread threading.Thread(targetself._consume_loop, daemonTrue) self.consumer_thread.start() logger.info(fConsumer started: {self.client_id}) def stop(self): 停止消費(fèi) 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): 執(zhí)行重平衡 # 觸發(fā)協(xié)調(diào)器分配 assignments self.coordinator.trigger_rebalance(self.subscriptions) # 更新自己的分配 self.assigned_partitions assignments.get(self.client_id, []) self.metrics[rebalances] 1 # 通知監(jiān)聽器 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 回調(diào) # 更新自己的分配 self.assigned_partitions assignments.get(self.client_id, []) logger.info(fRebalance completed, assigned: {self.assigned_partitions}) def _consume_loop(self): 消費(fèi)循環(huán) while self.running: if self.paused: time.sleep(0.1) continue if not self.assigned_partitions: time.sleep(0.5) continue # 遍歷所有分配的分區(qū) for topic, partition in self.assigned_partitions: self._fetch_and_process(topic, partition) time.sleep(0.05) # 防止空轉(zhuǎn) def _fetch_and_process(self, topic: str, partition: int): 拉取并處理消息 # 獲取已提交的 Offset committed_offset self.offset_store.get_offset( self.group_id, topic, partition ) # 發(fā)送拉取請(qǐng)求 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 # 自動(dòng)提交 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): 手動(dòng)提交 Offset self.offset_store._save() logger.info(Offsets committed manually) def pause(self): 暫停消費(fèi) self.paused True def resume(self): 恢復(fù)消費(fèi) self.paused False def seek(self, topic: str, partition: int, offset: int): 重置消費(fèi)位置 用于回溯消費(fèi) 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 監(jiān)聽器 self.rebalance_listeners.append(listener) def get_metrics(self) - dict: 獲取統(tǒng)計(jì) 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(): 演示基本消費(fèi) print( * 80) print( 基本消費(fèi)演示) 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) # 先生產(chǎn)一些消息 producer Producer(brokers[(localhost, 19492)]) producer.start() for i in range(10): producer.send(test, fMessage #{i}, keyfk{i}) producer.flush() producer.stop() # 消費(fèi)消息 print(\n消費(fèi)消息:) 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(): 演示消費(fèi)者組 print(\n * 80) print( 消費(fèi)者組演示) 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) # 生產(chǎn)消息 producer Producer(brokers[(localhost, 19493)]) producer.start() for i in range(30): producer.send(group-test, fMsg-{i}, keyfk{i}) producer.flush() producer.stop() # 啟動(dòng)兩個(gè)消費(fèi)者同一組 print(\n啟動(dòng)消費(fèi)者組2個(gè)消費(fèi)者:) 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 消費(fèi)統(tǒng)計(jì):) print(f Consumer1: {len(results[1])} 條) print(f Consumer2: {len(results[2])} 條) print(f 總計(jì): {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) # 生產(chǎn)消息 producer Producer(brokers[(localhost, 19494)]) producer.start() for i in range(10): producer.send(offset-demo, fData-{i}) producer.flush() producer.stop() # 第一次消費(fèi)消費(fèi)前5條 print(\n第一次消費(fèi)前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 消費(fèi)了 {len(first_batch)} 條) # 第二次消費(fèi)應(yīng)該從 offset5 開始 print(\n第二次消費(fèi)應(yīng)從 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 消費(fèi)了 {len(second_batch)} 條) if second_batch: print(f 第一條: {second_batch[0][value]}) # 回溯消費(fèi) print(\n回溯消費(fèi)從 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 消費(fèi)了 {len(third_batch)} 條全部) broker.stop() if __name__ __main__: demo_basic_consumption() demo_consumer_group() demo_offset_management()七、測(cè)試# 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 存儲(chǔ)測(cè)試 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): 測(cè)試提交和獲取 self.store.commit(group1, topic1, 0, 42) offset self.store.get_offset(group1, topic1, 0) self.assertEqual(offset, 42) def test_default_offset(self): 測(cè)試默認(rèn) Offset offset self.store.get_offset(unknown, topic, 0) self.assertEqual(offset, 0) class TestGroupCoordinator(unittest.TestCase): 組協(xié)調(diào)器測(cè)試 def setUp(self): self.coordinator GroupCoordinator(test-group) def test_register(self): 測(cè)試注冊(cè) need_rebalance self.coordinator.register(m1, host1, 9092) self.assertTrue(need_rebalance) # 重復(fù)注冊(cè)不應(yīng)該觸發(fā) Rebalance need_rebalance self.coordinator.register(m1, host1, 9092) self.assertFalse(need_rebalance) def test_unregister(self): 測(cè)試注銷 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): 分區(qū)分配測(cè)試 def test_range_assignor(self): 測(cè)試范圍分配 assignor RangeAssignor() members [c1, c2] topics {test: [0, 1, 2]} assignments assignor.assign(members, topics) self.assertEqual(len(assignments), 2) # c1 應(yīng)該有 2 個(gè)分區(qū)c2 有 1 個(gè) self.assertEqual(len(assignments[c1]), 2) self.assertEqual(len(assignments[c2]), 1) def test_round_robin(self): 測(cè)試輪詢分配 assignor RoundRobinAssignor() members [c1, c2, c3] topics {t1: [0, 1], t2: [0, 1]} assignments assignor.assign(members, topics) # 總共 4 個(gè)分區(qū)3 個(gè)消費(fèi)者 total sum(len(v) for v in assignments.values()) self.assertEqual(total, 4) class TestConsumerIntegration(unittest.TestCase): 消費(fèi)者集成測(cè)試 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): 測(cè)試消費(fèi)消息 # 先生產(chǎn) producer Producer(brokers[(localhost, 19592)]) producer.start() for i in range(5): producer.send(test, fmsg-{i}) producer.flush() producer.stop() # 再消費(fèi) 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()八、總結(jié)這一講我們實(shí)現(xiàn)了消費(fèi)者組的核心機(jī)制組件功能OffsetStore?Offset 持久化存儲(chǔ)支持自動(dòng)/手動(dòng)提交GroupCoordinator?組協(xié)調(diào)器管理成員和觸發(fā) RebalancePartitionAssignor?分區(qū)分配策略Range/RoundRobin/StickyConsumer?完整消費(fèi)者支持組消費(fèi)和 Offset 管理關(guān)鍵成果? 消費(fèi)者組機(jī)制支持水平擴(kuò)展? 自動(dòng) Rebalance動(dòng)態(tài)感知成員變化? Offset 自動(dòng)提交斷點(diǎn)續(xù)傳? 支持回溯消費(fèi)seek? 三種分區(qū)分配策略下一講我們將實(shí)現(xiàn)主從復(fù)制與高可用機(jī)制讓消息隊(duì)列在節(jié)點(diǎn)故障時(shí)依然可用。開發(fā)之余的小工具推薦?處理 Base64、JWT 解析、JSON 格式化、Crontab 計(jì)算、PDF 合并壓縮這些碎片需求我常用一個(gè)純前端本地工具箱zz365.top子頁(yè) PDF 大師PDF 大師 - zz365工具箱。所有計(jì)算在瀏覽器完成文件不上傳服務(wù)器關(guān)頁(yè)即清。免費(fèi)、無登錄、無廣告適合開發(fā)者當(dāng)常駐標(biāo)簽頁(yè)。