分布式、多節(jié)點(diǎn)系統(tǒng)下定時(shí)任務(wù)重復(fù)執(zhí)行問(wèn)題解決方案

在分布式多節(jié)點(diǎn)系統(tǒng)下，或者是使用gunicorn等工具啟用多個(gè)worker的情況下，如何保證后端的定時(shí)任務(wù)、初始化任務(wù)只執(zhí)行一次呢？比如使用apscheduler或者flask-apscheduler實(shí)現(xiàn)的定時(shí)任務(wù)。

在這種情況下必須借助外部數(shù)據(jù)庫(kù)才能實(shí)現(xiàn)，當(dāng)然，不僅僅只能是Redis，你也可以利用當(dāng)前系統(tǒng)下有的MySQL、或者M(jìn)ongoDB數(shù)據(jù)庫(kù)，只需要自定義一張表，創(chuàng)建一個(gè)unique字段作為鎖即可。

我這里將使用python語(yǔ)言，以MySQL為例，使用sqlalchemy+pymysql作為數(shù)據(jù)庫(kù)操作方式，使用裝飾器的方式對(duì)原有任務(wù)函數(shù)進(jìn)行改造，以達(dá)到對(duì)分布式的支持。你可以將該方法擴(kuò)散到其他語(yǔ)言、其他后端框架或者僅僅是定時(shí)任務(wù)后臺(tái)的情況。

一、創(chuàng)建帶唯一值字段的數(shù)據(jù)表

無(wú)論是redis、mysql、mongodb，要實(shí)現(xiàn)一個(gè)鎖的功能，作為鎖的字段必須為唯一值字段。

我這里使用了sqlalchemy創(chuàng)建表、指定lock_key字段unique=True，并定義了add_lock與delete_lock方法，當(dāng)然你也可以手動(dòng)在數(shù)據(jù)庫(kù)創(chuàng)建表，使用pymysql原生SQL語(yǔ)句實(shí)現(xiàn)鎖方法。
![示例結(jié)構(gòu)]

image.png

  • 若需要使用原生SQL語(yǔ)句定義表,可參照如下方式：
CREATE TABLE `task_lock` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`lock_key` varchar(20) COLLATE utf8mb4_bin NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `lock_key` (`lock_key`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
  • SQLAlchemy代碼示例
    sql_model.py 
from sqlalchemy import MetaData, Table, Column, BigInteger, String, create_engine
from sqlalchemy import insert, delete
from sqlalchemy.exc import IntegrityError


metadata = MetaData()

engine = create_engine("mysql+pymysql://root:mypassword@127.0.0.1:3306/test?charset=utf8mb4")

# 定義唯一值字段的任務(wù)鎖表
locks = Table("taskLock", metadata,
              Column("id", BigInteger(), primary_key=True, autoincrement=True),
              Column("lock_key", String(50), unique=True, nullable=False, comment="任務(wù)鎖")
        )

# 創(chuàng)建表
metadata.create_all(engine)

# 加鎖方法，bool值表示是否成功
def add_lock(lock_value):
    """添加唯一鎖"""
    ins = insert(locks).values(lock_key=lock_value)
    # 若當(dāng)前鎖值不存在，則可以插入成功，返回True
    try:
        engine.connect().execute(ins)
        return True
    # 若當(dāng)前插入鎖值已存在，則會(huì)觸發(fā)并捕獲該異常，返回False
    except IntegrityError:
        return False

# 刪除鎖方法，只要成功添加了鎖，任務(wù)執(zhí)行后，無(wú)論成功還是失敗都必須調(diào)用刪除方法
def delete_lock(lock_value):
    """刪除鎖"""
    d = delete(locks).where(locks.c.lock_key == lock_value)
    engine.connect().execute(d)

二、定義單節(jié)點(diǎn)任務(wù)裝飾器

為什么使用裝飾器？使用裝飾器的方式對(duì)原有函數(shù)進(jìn)行改造，可保留原始函數(shù)代碼不變，且復(fù)用性、可讀性更高。建議大家多多使用裝飾器哦，這里的裝飾器需要傳參，所以需要額外增加一層用來(lái)接收參數(shù)，關(guān)于裝飾器的學(xué)習(xí)，可以參考其他文檔。

decorators.py


from sql_models import add_lock, delete_lock


# 單節(jié)點(diǎn)任務(wù)裝飾器，被裝飾的任務(wù)在分布式多節(jié)點(diǎn)下同一時(shí)間只能運(yùn)行一次
def single_task(task):
    def wrap(func):
        def inner(*args, **kwargs):
            add_result = add_lock(task)
            if add_result:
                print("當(dāng)前節(jié)點(diǎn)獲取任務(wù)：{}!".format(task))
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    raise e
                finally:
                    delete_lock(task)

            else:
                print("當(dāng)前節(jié)點(diǎn)未獲取任務(wù)：{}".format(task))
                return
        return inner
    return wrap

該裝飾器的功能很簡(jiǎn)單，可接收一個(gè)任務(wù)名，該任務(wù)名將用作數(shù)據(jù)庫(kù)中lock_key的唯一值寫(xiě)入。在執(zhí)行原函數(shù)前，會(huì)先嘗試加鎖，即寫(xiě)入lock_key值，若寫(xiě)入成功，則獲得鎖，可以繼續(xù)執(zhí)行該任務(wù)函數(shù)；若加鎖失敗，即寫(xiě)入lock_key時(shí)數(shù)據(jù)庫(kù)已存在當(dāng)前值，說(shuō)明其他節(jié)點(diǎn)正在執(zhí)行該任務(wù)，則無(wú)法獲得鎖，不能執(zhí)行該任務(wù)函數(shù)，只會(huì)打印提示信息。

三、裝飾需單節(jié)點(diǎn)運(yùn)行的任務(wù)

需要被裝飾的任務(wù)一般是定時(shí)任務(wù)，或者是初始化時(shí)可能重復(fù)運(yùn)行任務(wù)。

tasks.py
在這里定義了3個(gè)模擬任務(wù)task1、task2、task3，并只對(duì)task1與task3使用single_task裝飾器進(jìn)行單節(jié)點(diǎn)運(yùn)行裝飾：

import time
import random

from decorators import single_task
from apscheduler.schedulers.background import BlockingScheduler


@single_task("task1")
def task1(arg1, arg2):
    print("----------------------------------------------")
    print("開(kāi)始執(zhí)行task1")
    time.sleep(random.randint(1, 5))
    print("task1執(zhí)行完成")
    print("----------------------------------------------" + "\n")
    return arg1 + arg2


def task2(arg1, arg2):
    print("----------------------------------------------")
    print("開(kāi)始執(zhí)行task2")
    time.sleep(random.randint(1, 5))
    print("task2執(zhí)行完成")
    print("----------------------------------------------" + "\n")
    return arg1 * arg2


@single_task("task3")
def task3(arg1, arg2):
    print("----------------------------------------------")
    print("開(kāi)始執(zhí)行task2")
    time.sleep(random.randint(1, 5))
    print("task2執(zhí)行完成")
    print("----------------------------------------------" + "\n")
    return arg1 / arg2

四、使用apscheduler開(kāi)啟定時(shí)任務(wù)

tasks.py

配置定時(shí)任務(wù)：

import time
import random

from decorators import single_task
from apscheduler.schedulers.background import BlockingScheduler


@single_task("task1")
def task1(arg1, arg2):
    print("----------------------------------------------")
    print("開(kāi)始執(zhí)行task1")
    time.sleep(random.randint(1, 5))
    print("task1執(zhí)行完成")
    print("----------------------------------------------" + "\n")
    return arg1 + arg2


def task2(arg1, arg2):
    print("----------------------------------------------")
    print("開(kāi)始執(zhí)行task2")
    time.sleep(random.randint(1, 5))
    print("task2執(zhí)行完成")
    print("----------------------------------------------" + "\n")
    return arg1 * arg2


@single_task("task3")
def task3(arg1, arg2):
    print("----------------------------------------------")
    print("開(kāi)始執(zhí)行task2")
    time.sleep(random.randint(1, 5))
    print("task2執(zhí)行完成")
    print("----------------------------------------------" + "\n")
    return arg1 / arg2


if __name__ == '__main__':
    print("開(kāi)始執(zhí)行定時(shí)任務(wù)")
    scheduler = BlockingScheduler()
    scheduler.add_job(task1, args=(5, 5), trigger="interval", seconds=20)
    scheduler.add_job(task2, args=(5, 5), trigger="interval", seconds=20)
    scheduler.add_job(task3, args=(5, 5), trigger="interval", seconds=20)
    scheduler.start()

五、任務(wù)演示

此處簡(jiǎn)單模擬多個(gè)命令行窗口，幾乎同時(shí)運(yùn)行，模擬多節(jié)點(diǎn)的情況：


image.png

可以發(fā)現(xiàn)實(shí)現(xiàn)了我們的設(shè)計(jì)，在兩個(gè)節(jié)點(diǎn)中，task1和task3只會(huì)運(yùn)行一次，而未加裝飾器的task2不受限制，每個(gè)節(jié)點(diǎn)中都會(huì)重復(fù)運(yùn)行。

多節(jié)點(diǎn)，比如存在寫(xiě)入數(shù)據(jù)庫(kù)或者文件操作的定時(shí)任務(wù)，則設(shè)置單節(jié)點(diǎn)運(yùn)行是非常有必要的。

總結(jié)

同理，你可以使用相同的方式，利用MongDB或其他數(shù)據(jù)庫(kù)實(shí)現(xiàn)單節(jié)點(diǎn)鎖的裝飾器，若有Redis則實(shí)現(xiàn)起來(lái)就更加容易和普遍了。

以上，希望文章能對(duì)你有所幫助。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀(guān)點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

友情鏈接更多精彩內(nèi)容