RocketMQ与Kafka事务消息机制深度对比

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""Simplified Sentinel Implementation in PythonIncludes: TokenBucket, SlidingWindowCounter, CircuitBreaker, SlotChain""" import timeimport threadingimport randomfrom collections import dequefrom enum import Enumfrom abc import ABC, abstractmethod  # ============================================================# 1. 令牌桶限流器 (Token Bucket Rate Limiter)# ============================================================class TokenBucket:    """令牌桶限流器：以固定速率生成令牌，支持突发流量。"""     def __init__(self, rate: int, capacity: int = None):        """        :param rate: 每秒生成的令牌数 (QPS)        :param capacity: 桶容量，默认与 rate 相同        """        self.rate = rate        self.capacity = capacity or rate        self.tokens = float(self.capacity)        self.last_refill = time.monotonic()        self.lock = threading.Lock()     def _refill(self):        now = time.monotonic()        elapsed = now - self.last_refill        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)        self.last_refill = now     def acquire(self, tokens: int = 1) -> bool:        """尝试获取令牌，成功返回 True，否则返回 False。"""        with self.lock:            self._refill()            if self.tokens >= tokens:                self.tokens -= tokens                return True            return False     def wait_and_acquire(self, tokens: int = 1, timeout: float = 0) -> bool:        """等待并获取令牌，支持超时。"""        deadline = time.monotonic() + timeout        while True:            with self.lock:                self._refill()                if self.tokens >= tokens:                    self.tokens -= tokens                    return True            remaining = deadline - time.monotonic()            if remaining <= 0:                return False            time.sleep(min(0.01, remaining))  # ============================================================# 2. 滑动窗口计数器 (Sliding Window Counter)# ============================================================class SlidingWindowCounter:    """滑动窗口计数器：将时间窗口划分为多个桶，实现精确 QPS 统计。"""     def __init__(self, window_size: int = 1, bucket_count: int = 10):        """        :param window_size: 窗口大小（秒）        :param bucket_count: 桶数量        """        self.window_size = window_size        self.bucket_count = bucket_count        self.bucket_span = window_size / bucket_count        self.buckets = deque([0] * bucket_count, maxlen=bucket_count)        self.last_bucket_time = time.monotonic()        self.lock = threading.Lock()     def _slide_window(self):        now = time.monotonic()        elapsed = now - self.last_bucket_time        steps = int(elapsed / self.bucket_span)        if steps > 0:            for _ in range(min(steps, self.bucket_count)):                self.buckets.append(0)            self.last_bucket_time = now     def increment(self, count: int = 1):        with self.lock:            self._slide_window()            self.buckets[-1] += count     def get_count(self) -> int:        with self.lock:            self._slide_window()            return sum(self.buckets)     def get_qps(self) -> float:        return self.get_count() / self.window_size  # ============================================================# 3. 熔断器 (Circuit Breaker)# ============================================================class CircuitState(Enum):    CLOSED = "CLOSED"        # 正常状态    OPEN = "OPEN"            # 熔断状态    HALF_OPEN = "HALF_OPEN"  # 半开状态  class CircuitBreaker:    """熔断器：支持慢调用比例、异常比例、异常数三种策略。"""     def __init__(        self,        name: str,        slow_ratio_threshold: float = 0.5,        max_slow_rt: float = 1.0,        error_ratio_threshold: float = 0.5,        error_count_threshold: int = 20,        min_request_count: int = 5,        recovery_timeout: float = 10.0,        stat_window: int = 60,    ):        self.name = name        self.slow_ratio_threshold = slow_ratio_threshold        self.max_slow_rt = max_slow_rt        self.error_ratio_threshold = error_ratio_threshold        self.error_count_threshold = error_count_threshold        self.min_request_count = min_request_count        self.recovery_timeout = recovery_timeout        self.stat_window = stat_window         self.state = CircuitState.CLOSED        self.last_state_change = time.monotonic()        self.counter = SlidingWindowCounter(window_size=stat_window, bucket_count=stat_window)        self.slow_counter = SlidingWindowCounter(window_size=stat_window, bucket_count=stat_window)        self.error_counter = SlidingWindowCounter(window_size=stat_window, bucket_count=stat_window)        self.lock = threading.Lock()     def _transition_to(self, new_state: CircuitState):        self.state = new_state        self.last_state_change = time.monotonic()     def _check_and_transition(self):        total = self.counter.get_count()        if total < self.min_request_count:            return         slow_count = self.slow_counter.get_count()        error_count = self.error_counter.get_count()         # 慢调用比例策略        if slow_count / total >= self.slow_ratio_threshold:            self._transition_to(CircuitState.OPEN)            return         # 异常比例策略        if error_count / total >= self.error_ratio_threshold:            self._transition_to(CircuitState.OPEN)            return         # 异常数策略        if error_count >= self.error_count_threshold:            self._transition_to(CircuitState.OPEN)            return     def allow_request(self) -> bool:        with self.lock:            if self.state == CircuitState.CLOSED:                return True            if self.state == CircuitState.OPEN:                elapsed = time.monotonic() - self.last_state_change                if elapsed >= self.recovery_timeout:                    self._transition_to(CircuitState.HALF_OPEN)                    return True                return False            # HALF_OPEN: 允许一个探测请求            return True     def record_success(self, rt: float = 0):        with self.lock:            self.counter.increment()            if rt > self.max_slow_rt:                self.slow_counter.increment()            if self.state == CircuitState.HALF_OPEN:                self._transition_to(CircuitState.CLOSED)     def record_error(self, rt: float = 0):        with self.lock:            self.counter.increment()            self.error_counter.increment()            if rt > self.max_slow_rt:                self.slow_counter.increment()            if self.state == CircuitState.HALF_OPEN:                self._transition_to(CircuitState.OPEN)            elif self.state == CircuitState.CLOSED:                self._check_and_transition()  # ============================================================# 4. 简单 Slot 链 (Slot Chain)# ============================================================class Slot(ABC):    """抽象 Slot 基类。"""     def __init__(self):        self.next_slot = None     def fire(self, context: dict):        self.entry(context)        if self.next_slot:            self.next_slot.fire(context)        self.exit(context)     @abstractmethod    def entry(self, context: dict):        pass     def exit(self, context: dict):        pass  class StatisticSlot(Slot):    """统计 Slot：记录请求数、RT 等指标。"""     def entry(self, context: dict):        context["start_time"] = time.monotonic()        resource = context.get("resource", "unknown")        print(f"[StatisticSlot] Recording entry for resource: {resource}")     def exit(self, context: dict):        elapsed = time.monotonic() - context.get("start_time", 0)        context["rt"] = elapsed        resource = context.get("resource", "unknown")        print(f"[StatisticSlot] Resource '{resource}' RT: {elapsed:.4f}s")  class FlowSlot(Slot):    """流量控制 Slot：基于令牌桶进行限流判断。"""     def __init__(self, limiters: dict = None):        super().__init__()        self.limiters = limiters or {}     def entry(self, context: dict):        resource = context.get("resource", "unknown")        limiter = self.limiters.get(resource)        if limiter and not limiter.acquire():            raise BlockException(f"Flow control: {resource} is blocked")  class DegradeSlot(Slot):    """熔断降级 Slot：基于熔断器进行降级判断。"""     def __init__(self, breakers: dict = None):        super().__init__()        self.breakers = breakers or {}     def entry(self, context: dict):        resource = context.get("resource", "unknown")        breaker = self.breakers.get(resource)        if breaker and not breaker.allow_request():            raise BlockException(f"Degrade: {resource} is degraded")  class BlockException(Exception):    """限流/降级异常。"""    pass  class SlotChain:    """Slot 链管理器：组装并执行 Slot 链。"""     def __init__(self):        self.head = None        self.tail = None     def add_last(self, slot: Slot):        if not self.head:            self.head = slot            self.tail = slot        else:            self.tail.next_slot = slot            self.tail = slot     def entry(self, resource: str):        context = {"resource": resource}        try:            self.head.fire(context)            return context        except BlockException as e:            print(f"Blocked: {e}")            raise  # ============================================================# 5. Demo: 测试简化 Sentinel# ============================================================def demo():    print("=" * 60)    print("Simplified Sentinel Demo")    print("=" * 60)     # 创建限流器    limiters = {        "api:search": TokenBucket(rate=5, capacity=10),        "api:recommend": TokenBucket(rate=3, capacity=5),    }     # 创建熔断器    breakers = {        "api:search": CircuitBreaker(            name="api:search",            slow_ratio_threshold=0.3,            error_ratio_threshold=0.3,            recovery_timeout=5.0,        ),    }     # 构建 Slot 链    chain = SlotChain()    chain.add_last(StatisticSlot())    chain.add_last(FlowSlot(limiters=limiters))    chain.add_last(DegradeSlot(breakers=breakers))     print("\n--- Test 1: Normal flow ---")    for i in range(10):        try:            ctx = chain.entry("api:search")            time.sleep(random.uniform(0.05, 0.15))            if random.random() < 0.1:                raise RuntimeError("Simulated error")            print(f"  Request {i+1}: OK (RT={ctx.get('rt', 0):.4f}s)")        except BlockException:            print(f"  Request {i+1}: BLOCKED")        except RuntimeError:            print(f"  Request {i+1}: ERROR")            breaker = breakers.get("api:search")            if breaker:                breaker.record_error(rt=random.uniform(0.5, 2.0))     print("\n--- Test 2: Token bucket limit test ---")    bucket = TokenBucket(rate=10, capacity=10)    passed = 0    for i in range(50):        if bucket.acquire():            passed += 1    print(f"  Passed: {passed}/50 (rate=10, should be ~10)")     print("\n--- Test 3: Sliding window counter ---")    swc = SlidingWindowCounter(window_size=1, bucket_count=10)    for _ in range(100):        swc.increment()    print(f"  Count: {swc.get_count()} (should be 100)")    print(f"  QPS: {swc.get_qps():.1f} (should be ~100)")     print("\n--- Test 4: Circuit breaker ---")    cb = CircuitBreaker(name="test", error_ratio_threshold=0.3, recovery_timeout=3.0)    for i in range(20):        if cb.allow_request():            if random.random() < 0.5:                cb.record_error()                print(f"  Request {i+1}: ERROR")            else:                cb.record_success()                print(f"  Request {i+1}: OK")        else:            print(f"  Request {i+1}: REJECTED (circuit open)")    print(f"  Final state: {cb.state.value}")  if __name__ == "__main__":    demo()