使用Redis實(shí)現(xiàn)定時(shí)任務(wù)多節(jié)點(diǎn)部署及自動(dòng)任務(wù)分發(fā)的方法

from redis import Redisfrom apscheduler.schedulers.blocking import BlockingSchedulerimport timeimport randomr = Redis(host='localhost', port=6379, db=0, decode_responses=True)TASK_LIST = ["任務(wù)參數(shù)1", "任務(wù)參數(shù)2", "任務(wù)參數(shù)3", "任務(wù)參數(shù)4"]# 存放當(dāng)前已完成的集合鍵FINISHED_TASK_KEY = "task:finished"# 給每個(gè)任務(wù)順序分配不同的redis鍵TASK_KEYS = ["task: {}".format(task) for task in range(len(TASK_LIST))]# redis鍵對(duì)應(yīng)的任務(wù)參數(shù)TASK_MAP = {key: task for key, task in zip(TASK_KEYS, TASK_LIST)}EXPIRE = 60 * 60 * 23# 模擬定時(shí)執(zhí)行的任務(wù)，可接受不同的參數(shù)def task(param):
    print("----------------------------------")
    print("開始執(zhí)行任務(wù)：{}....".format(param))
    time.sleep(random.randint(15, 30))
    print("完成{}".format(param))
    print("----------------------------------")def start():
    # 每日任務(wù)開始前先刪除redis中已完成任務(wù)集合
    r.delete(FINISHED_TASK_KEY)
    # 添加redis中唯一任務(wù)鎖，若添加成功且任務(wù)完成表中無當(dāng)前任務(wù)則執(zhí)行
    for key in TASK_KEYS:
        status = r.set(key, 1, ex=EXPIRE, nx=True)
        # 檢查當(dāng)前任務(wù)是否已執(zhí)行
        is_finished = r.sismember(FINISHED_TASK_KEY, key)
        if status is True and is_finished is False:
            print("當(dāng)前節(jié)點(diǎn)獲取任務(wù)：{} 成功, 準(zhǔn)備執(zhí)行。。".format(key))
            # 獲取當(dāng)前redis鍵對(duì)應(yīng)的任務(wù)參數(shù)并執(zhí)行任務(wù)
            task_param = TASK_MAP.get(key)
            task(task_param)
            # 任務(wù)執(zhí)行完成將當(dāng)前任務(wù)添加到已完成集合，不允許其他節(jié)點(diǎn)再獲取該任務(wù)
            r.sadd(FINISHED_TASK_KEY, key)
            r.delete(key)
        elif status is True and is_finished is True:
            print("未執(zhí)行: {}，其他節(jié)點(diǎn)已完成該任務(wù)".format(key))
        else:
            print("未執(zhí)行：{}，其他節(jié)點(diǎn)正在執(zhí)行當(dāng)前任務(wù)".format(key))if __name__ == '__main__':
    # 這里可使用apscheduler定時(shí)任務(wù)框架調(diào)度start函數(shù)，該處模擬未使用定時(shí)任務(wù)
    start()