Skill编程实战：如何设计高可用的异步任务调度系统

5次阅读

没有评论

共计 3035 个字符，预计需要花费 8 分钟才能阅读完成。

在分布式系统中，异步任务调度是实现解耦和性能优化的重要手段。然而，任务丢失、重复执行和性能瓶颈等问题常常困扰着开发者。本文将基于 Skill 编程理念，分享一套结合 Redis 持久化与幂等设计的高可用异步任务调度解决方案。

在分布式环境下，异步任务调度面临的主要挑战包括：

任务丢失 ：由于网络分区或节点宕机，任务可能在传输或处理过程中丢失
重复执行 ：重试机制可能导致任务被多次消费，造成数据不一致
性能瓶颈 ：随着任务量增长，传统的数据库轮询方式难以满足高吞吐需求

常见的异步任务调度方案有以下几种：

数据库轮询 ：实现简单但吞吐量低，频繁查询会增加数据库压力
RabbitMQ 延迟队列 ：支持消息持久化但集群扩展复杂，延迟队列实现成本高
Redis Stream：高吞吐量，支持消息持久化和消费者组，是分布式任务调度的理想选择

使用 Redis 的 LPUSH/BRPOP 命令实现任务队列，BRPOP 在队列为空时会阻塞连接，避免频繁轮询。

# 生产者示例
import redis

r = redis.Redis(host='localhost', port=6379)

def produce_task(task_data, priority='normal'):
    queue_name = f'task_queue:{priority}'
    r.lpush(queue_name, json.dumps(task_data))

通过 Lua 脚本保证 ACK 和重试机制的原子性，避免网络问题导致的中间状态。

-- Lua 脚本示例
local task = redis.call('RPOP', KEYS[1])
if task then
    redis.call('HSET', KEYS[2], ARGV[1], task)
    redis.call('EXPIRE', KEYS[2], ARGV[2])
    return task
end
return nil

为每个任务生成唯一的 SHA256 指纹，在处理前检查指纹是否已存在。

import hashlib

def get_task_fingerprint(task_data):
    return hashlib.sha256(json.dumps(task_data).encode()).hexdigest()

def produce_task_with_priority(task_data, priority='normal'):
    """
    生产带有优先级的任务
    :param task_data: 任务数据
    :param priority: 优先级（high/normal/low）"""
    task_id = str(uuid.uuid4())
    task = {
        'id': task_id,
        'data': task_data,
        'timestamp': int(time.time()),
        'retry_count': 0,
        'fingerprint': get_task_fingerprint(task_data)
    }
    queue_name = f'task_queue:{priority}'
    r.lpush(queue_name, json.dumps(task))
    return task_id

def consumer_worker():
    while True:
        # 优先处理高优先级队列
        queue_order = ['task_queue:high', 'task_queue:normal', 'task_queue:low']
        task_json = r.brpop(queue_order, timeout=30)

        if task_json:
            task = json.loads(task_json[1])
            try:
                process_task(task)
                ack_task(task['id'])
            except Exception as e:
                handle_failed_task(task, str(e))

def dead_letter_handler():
    while True:
        failed_task = r.rpop('dead_letter_queue')
        if not failed_task:
            time.sleep(60)
            continue

        task = json.loads(failed_task)
        if task['retry_count'] < MAX_RETRY:
            retry_task(task)
        else:
            notify_admin(task)

基于队列长度和消费速度动态调整消费者线程数：

def adjust_workers():
    queue_length = r.llen('task_queue:high') + r.llen('task_queue:normal')
    current_workers = get_current_worker_count()

    if queue_length > THRESHOLD_HIGH and current_workers < MAX_WORKERS:
        scale_up_workers()
    elif queue_length < THRESHOLD_LOW and current_workers > MIN_WORKERS:
        scale_down_workers()

使用 Prometheus 客户端库暴露关键指标：

from prometheus_client import Counter, Gauge

TASKS_PROCESSED = Counter('tasks_processed_total', 'Total processed tasks')
TASKS_FAILED = Counter('tasks_failed_total', 'Total failed tasks')
QUEUE_LENGTH = Gauge('task_queue_length', 'Current task queue length')

def process_task(task):
    try:
        # 任务处理逻辑
        TASKS_PROCESSED.inc()
    except Exception:
        TASKS_FAILED.inc()
        raise

热数据：近期任务存储在 Redis 中
冷数据：历史任务归档到对象存储（如 S3）

根据业务需求选择合适的持久化策略：

AOF 每秒 fsync：平衡性能和数据安全性
RDB 快照：适合允许少量数据丢失的场景

当 Redis 集群进行 resharding 时，可以通过以下方式减少影响：

使用 hash tag 确保相关 key 在同一 slot
客户端实现自动重试逻辑
监控集群状态，避免在高峰期进行迁移

stateDiagram-v2
    [*] --> Pending: 任务创建
    Pending --> Processing: 消费者获取
    Processing --> Completed: 处理成功
    Processing --> Failed: 处理失败
    Failed --> Pending: 重试
    Failed --> DeadLetter: 超过重试次数

如何设计跨地域的任务路由策略？考虑因素包括：