import time
import threading
import queue

class TrafficControlManager:
def __init__(self, max_requests_per_second, error_threshold=0.5):
self.max_requests_per_second = max_requests_per_second
self.error_threshold = error_threshold
self.request_count = 0
self.error_count = 0
self.last_second = time.time()
self.is_degraded = False
self.degrade_end_time = 0
self.lock = threading.Lock()

def can_process(self):
current_time = time.time()

with self.lock:
# 检查是否处于降级状态
if self.is_degraded:
if current_time < self.degrade_end_time: return False else: # 退出降级状态 self.is_degraded = False self.error_count = 0 # 重置计数器 if current_time - self.last_second >= 1:
self.request_count = 0
self.error_count = 0
self.last_second = current_time

# 检查流量控制
if self.request_count >= self.max_requests_per_second:
return False

self.request_count += 1
return True

def record_error(self):
with self.lock:
self.error_count += 1
# 检查是否需要降级
if self.request_count > 0 and self.error_count / self.request_count >= self.error_threshold:
self.is_degraded = True
self.degrade_end_time = time.time() + 10 # 降级10秒
print(f’Degraded due to high error rate: {self.error_count}/{self.request_count}’)

# 测试自定义流量控制与降级
def process_request(manager, request_id):
if manager.can_process():
try:
# 模拟处理，随机抛出异常
import random
if random.random() > 0.5:
raise Exception(‘Random error’)
print(f’Processing request {request_id}’)
# 模拟处理时间
time.sleep(0.1)
except Exception as e:
print(f’Error processing request {request_id}: {e}’)
manager.record_error()
else:
print(f’Blocked or degraded request {request_id}’)

# 创建流量控制管理器
manager = TrafficControlManager(max_requests_per_second=10)

# 模拟高并发请求
threads = []
for i in range(20):
t = threading.Thread(target=process_request, args=(manager, i))
threads.append(t)
t.start()
time.sleep(0.05)

# 等待所有线程完成
for t in threads:
t.join()

print(‘All requests processed’)