一、业务场景
在煤炉自动下单和雅虎代拍场景中，系统需要同时监控成千上万个用户设置的关键词和商品。每个监控任务都需要定期执行——搜索新上架的商品、检查价格变化、触发自动下单。
我最早实现的是一个简单的while True循环 + time.sleep()方案：
python
import timedef monitor_task(keyword, max_price): while True: items = search_mercari(keyword, max_price) for item in items: if match_condition(item): place_order(item) time.sleep(5) # 每5秒检查一次
当监控任务从10个增长到1000个的时候，这个方案彻底崩溃了——1000个线程同时运行，每个线程5秒一次请求，每秒就是200次请求，服务器CPU直接飙到100%。
二、APScheduler方案
我首先用APScheduler重构了任务调度：
python
from apscheduler.schedulers.background import BackgroundSchedulerfrom apscheduler.triggers.interval import IntervalTriggerscheduler = BackgroundScheduler()def add_monitor_task(keyword, max_price, user_id): scheduler.add_job( func=monitor_once, trigger=IntervalTrigger(seconds=5), args=[keyword, max_price, userid], id=f"monitor{userid}{keyword}", replace_existing=True )
APScheduler比手动循环好一些，但本质上还是每个任务独立调度。当任务数量达到几千个时，调度器本身的开销就很大了。
三、基于Redis ZSET的集中式任务调度
最终方案是把所有监控任务放在Redis的有序集合（ZSET）中，由一组Worker竞争消费：
python
import redisimport timeimport jsonclass TaskScheduler: def init(self): self.redis = redis.Redis(decode_responses=True) self.task_queue = "scheduler:tasks" self.running_key = "scheduler:running" def add_task(self, task_id, task_data, interval=5): """添加一个周期性任务""" # 计算下一次执行时间 next_run = time.time() + interval # 存储任务详情 detail_key = f"task:detail:{task_id}" self.redis.hset(detail_key, mapping={ 'data': json.dumps(task_data), 'interval': interval, 'last_run': '0' }) # 加入ZSET，score为下一次执行时间 self.redis.zadd(self.task_queue, {task_id: next_run}) def acquire_task(self, worker_id): """Worker获取一个待执行的任务""" # 获取score最小的任务（最早需要执行的） tasks = self.redis.zrange(self.task_queue, 0, 0, withscores=True) if not tasks: return None task_id, score = tasks[0] now = time.time() if score > now: # 还没到执行时间 return None # 尝试锁定这个任务（使用Lua脚本保证原子性） lock_script = """ local task_id = ARGV[1] local worker_id = ARGV[2] local now = tonumber(ARGV[3]) -- 从ZSET中移除 local removed = redis.call('ZREM', KEYS[1], task_id) if removed == 0 then return nil end -- 记录正在运行 redis.call('HSET', KEYS[2], task_id, worker_id) return 'ok' """ result = self.redis.eval( lock_script, 2, self.task_queue, self.running_key, task_id, worker_id, now ) if result is None: return None # 获取任务详情 detail_key = f"task:detail:{task_id}" task_data = self.redis.hgetall(detail_key) return { 'task_id': task_id, 'data': json.loads(task_data['data']), 'interval': int(task_data['interval']) } def complete_task(self, task_id): """任务执行完成，重新调度""" detail_key = f"task:detail:{task_id}" interval = int(self.redis.hget(detail_key, 'interval')) next_run = time.time() + interval # 更新最后执行时间 self.redis.hset(detail_key, 'last_run', str(time.time())) # 重新加入ZSET self.redis.zadd(self.task_queue, {task_id: next_run}) # 从运行中移除 self.redis.hdel(self.running_key, task_id)
Worker的实现：
python
import threadingimport timeclass Worker: def init(self, worker_id, scheduler): self.worker_id = worker_id self.scheduler = scheduler self.running = True def run(self): while self.running: task = self.scheduler.acquire_task(self.worker_id) if task: self._execute(task) else: time.sleep(0.1) # 没有任务时短暂休眠 def _execute(self, task): try: # 执行监控逻辑 items = search_mercari( task['data']['keyword'], task['data']['max_price'] ) for item in items: if match_condition(item, task['data']): place_order(item, task['data']['user_id']) except Exception as e: # 记录错误，但不影响任务重新调度 log_error(task['task_id'], str(e)) finally: self.scheduler.complete_task(task['task_id'])
四、弹性扩缩容
这套架构支持动态增减Worker——流量大的时候启动更多Worker，流量小的时候减少Worker：
python
class WorkerManager: def init(self, scheduler): self.scheduler = scheduler self.workers = [] self.min_workers = 2 self.max_workers = 20 def adjust_workers(self): # 根据队列长度调整Worker数量 queue_length = self.scheduler.redis.zcard(self.scheduler.task_queue) target = max(self.min_workers, min(self.max_workers, queuelength // 50 + 1)) while len(self.workers) < target: worker = Worker(f"worker{len(self.workers)}", self.scheduler) threading.Thread(target=worker.run, daemon=True).start() self.workers.append(worker)
五、总结
这套基于Redis ZSET的分布式任务调度方案，支撑了数千个煤炉监控任务的同时运行，任务延迟控制在200毫秒以内。核心经验是：不要把调度逻辑分散在各个任务中，而是集中管理、统一调度。
