使用Redis实现定时任务多节点部署及自动任务分发的方法

from redis import Redisfrom apscheduler.schedulers.blocking import BlockingSchedulerimport timeimport randomr = Redis(host='localhost', port=6379, db=0, decode_responses=True)TASK_LIST = ["任务参数1", "任务参数2", "任务参数3", "任务参数4"]# 存放当前已完成的集合键FINISHED_TASK_KEY = "task:finished"# 给每个任务顺序分配不同的redis键TASK_KEYS = ["task: {}".format(task) for task in range(len(TASK_LIST))]# redis键对应的任务参数TASK_MAP = {key: task for key, task in zip(TASK_KEYS, TASK_LIST)}EXPIRE = 60 * 60 * 23# 模拟定时执行的任务，可接受不同的参数def task(param):
    print("----------------------------------")
    print("开始执行任务：{}....".format(param))
    time.sleep(random.randint(15, 30))
    print("完成{}".format(param))
    print("----------------------------------")def start():
    # 每日任务开始前先删除redis中已完成任务集合
    r.delete(FINISHED_TASK_KEY)
    # 添加redis中唯一任务锁，若添加成功且任务完成表中无当前任务则执行
    for key in TASK_KEYS:
        status = r.set(key, 1, ex=EXPIRE, nx=True)
        # 检查当前任务是否已执行
        is_finished = r.sismember(FINISHED_TASK_KEY, key)
        if status is True and is_finished is False:
            print("当前节点获取任务：{} 成功, 准备执行。。".format(key))
            # 获取当前redis键对应的任务参数并执行任务
            task_param = TASK_MAP.get(key)
            task(task_param)
            # 任务执行完成将当前任务添加到已完成集合，不允许其他节点再获取该任务
            r.sadd(FINISHED_TASK_KEY, key)
            r.delete(key)
        elif status is True and is_finished is True:
            print("未执行: {}，其他节点已完成该任务".format(key))
        else:
            print("未执行：{}，其他节点正在执行当前任务".format(key))if __name__ == '__main__':
    # 这里可使用apscheduler定时任务框架调度start函数，该处模拟未使用定时任务
    start()