基于Python實(shí)現(xiàn)配置熱加載的方法詳解

更新時(shí)間：2022年07月09日 16:47:06 作者：煙熏柿子學(xué)編程

所謂的配置熱加載，也就是說(shuō)當(dāng)服務(wù)收到配置更新消息之后，我們不用重啟服務(wù)就可以使用最新的配置去執(zhí)行任務(wù)。本文將介紹如何用Python實(shí)現(xiàn)配置熱加載，需要的可以參考一下

背景

由于最近工作需求，需要在已有項(xiàng)目添加一個(gè)新功能，實(shí)現(xiàn)配置熱加載的功能。所謂的配置熱加載，也就是說(shuō)當(dāng)服務(wù)收到配置更新消息之后，我們不用重啟服務(wù)就可以使用最新的配置去執(zhí)行任務(wù)。

如何實(shí)現(xiàn)

下面我分別采用多進(jìn)程、多線程、協(xié)程的方式去實(shí)現(xiàn)配置熱加載。

使用多進(jìn)程實(shí)現(xiàn)配置熱加載

如果我們代碼實(shí)現(xiàn)上使用多進(jìn)程，主進(jìn)程1來(lái)更新配置并發(fā)送指令，任務(wù)的調(diào)用是進(jìn)程2，如何實(shí)現(xiàn)配置熱加載呢？

使用signal信號(hào)量來(lái)實(shí)現(xiàn)熱加載

當(dāng)主進(jìn)程收到配置更新的消息之后(配置讀取是如何收到配置更新的消息的? 這里我們暫不討論)，主進(jìn)程就向進(jìn)子程1發(fā)送kill信號(hào)，子進(jìn)程1收到kill的信號(hào)就退出，之后由信號(hào)處理函數(shù)來(lái)啟動(dòng)一個(gè)新的進(jìn)程，使用最新的配置文件來(lái)繼續(xù)執(zhí)行任務(wù)。

main 函數(shù)

def main():
    # 啟動(dòng)一個(gè)進(jìn)程執(zhí)行任務(wù)
    p1 = Process(target=run, args=("p1",))
    p1.start()

    monitor(p1, run) # 注冊(cè)信號(hào)
    processes["case100"] = p1 #將進(jìn)程pid保存
    num = 0 
    while True: # 模擬獲取配置更新
        print(
            f"{multiprocessing.active_children()=}, count={len(multiprocessing.active_children())}\n")
        print(f"{processes=}\n")
        sleep(2)
        if num == 4:
            kill_process(processes["case100"]) # kill 當(dāng)前進(jìn)程
        if num == 8:
            kill_process(processes["case100"]) # kill 當(dāng)前進(jìn)程
        if num == 12:
            kill_process(processes["case100"]) # kill 當(dāng)前進(jìn)程
        num += 1

signal_handler 函數(shù)

def signal_handler(process: Process, func, signum, frame):
    # print(f"{signum=}")
    global counts

    if signum == 17:  # 17 is SIGCHILD 
        # 這個(gè)循環(huán)是為了忽略SIGTERM發(fā)出的信號(hào)，避免搶占了主進(jìn)程發(fā)出的SIGCHILD
        for signame in [SIGTERM, SIGCHLD, SIGQUIT]:
            signal.signal(signame, SIG_DFL)

        print("Launch a new process")
        p = multiprocessing.Process(target=func, args=(f"p{counts}",))
        p.start()
        monitor(p, run)
        processes["case100"] = p
        counts += 1

    if signum == 2:
        if process.is_alive():
            print(f"Kill {process} process")
            process.terminate()
        signal.signal(SIGCHLD, SIG_IGN)
        sys.exit("kill parent process")

完整代碼如下

#! /usr/local/bin/python3.8
from multiprocessing import Process
from typing import Dict
import signal
from signal import SIGCHLD, SIGTERM, SIGINT, SIGQUIT, SIG_DFL, SIG_IGN
import multiprocessing
from multiprocessing import Process
from typing import Callable
from data import processes
import sys
from functools import partial
import time

processes: Dict[str, Process] = {}
counts = 2


def run(process: Process):
    while True:
        print(f"{process} running...")
        time.sleep(1)


def kill_process(process: Process):
    print(f"kill {process}")
    process.terminate()


def monitor(process: Process, func: Callable):
    for signame in [SIGTERM, SIGCHLD, SIGINT, SIGQUIT]:
        # SIGTERM is kill signal.
        # No SIGCHILD is not trigger singnal_handler,
        # No SIGINT is not handler ctrl+c,
        # No SIGQUIT is RuntimeError: reentrant call inside <_io.BufferedWriter name='<stdout>'>
        signal.signal(signame, partial(signal_handler, process, func))


def signal_handler(process: Process, func, signum, frame):
    print(f"{signum=}")
    global counts

    if signum == 17:  # 17 is SIGTERM
        for signame in [SIGTERM, SIGCHLD, SIGQUIT]:
            signal.signal(signame, SIG_DFL)
        print("Launch a new process")
        p = multiprocessing.Process(target=func, args=(f"p{counts}",))
        p.start()
        monitor(p, run)
        processes["case100"] = p
        counts += 1

    if signum == 2:
        if process.is_alive():
            print(f"Kill {process} process")
            process.terminate()
        signal.signal(SIGCHLD, SIG_IGN)
        sys.exit("kill parent process")


def main():
    p1 = Process(target=run, args=("p1",))
    p1.start()
    monitor(p1, run)
    processes["case100"] = p1
    num = 0
    while True:
        print(
            f"{multiprocessing.active_children()=}, count={len(multiprocessing.active_children())}\n")
        print(f"{processes=}\n")
        time.sleep(2)
        if num == 4:
            kill_process(processes["case100"])
        if num == 8:
            kill_process(processes["case100"])
        if num == 12:
            kill_process(processes["case100"])
        num += 1


if __name__ == '__main__':
    main()

執(zhí)行結(jié)果如下

multiprocessing.active_children()=[<Process name='Process-1' pid=2533 parent=2532 started>], count=1

processes={'case100': <Process name='Process-1' pid=2533 parent=2532 started>}

p1 running...
p1 running...
kill <Process name='Process-1' pid=2533 parent=2532 started>
multiprocessing.active_children()=[<Process name='Process-1' pid=2533 parent=2532 started>], count=1

processes={'case100': <Process name='Process-1' pid=2533 parent=2532 started>}

signum=17
Launch a new process
p2 running...
p2 running...
multiprocessing.active_children()=[<Process name='Process-2' pid=2577 parent=2532 started>], count=1

processes={'case100': <Process name='Process-2' pid=2577 parent=2532 started>}

p2 running...
p2 running...
multiprocessing.active_children()=[<Process name='Process-2' pid=2577 parent=2532 started>], count=1

processes={'case100': <Process name='Process-2' pid=2577 parent=2532 started>}

p2 running...
p2 running...
multiprocessing.active_children()=[<Process name='Process-2' pid=2577 parent=2532 started>], count=1

processes={'case100': <Process name='Process-2' pid=2577 parent=2532 started>}

p2 running...
p2 running...
kill <Process name='Process-2' pid=2577 parent=2532 started>
signum=17
Launch a new process
multiprocessing.active_children()=[<Process name='Process-2' pid=2577 parent=2532 stopped exitcode=-SIGTERM>], count=1

processes={'case100': <Process name='Process-3' pid=2675 parent=2532 started>}

p3 running...
p3 running...
multiprocessing.active_children()=[<Process name='Process-3' pid=2675 parent=2532 started>], count=1

總結(jié)

好處：使用信號(hào)量可以處理多進(jìn)程之間通信的問(wèn)題。

壞處：代碼不好寫，寫出來(lái)代碼不好理解。信號(hào)量使用必須要很熟悉，不然很容易自己給自己寫了一個(gè)bug.(所有初學(xué)者慎用，老司機(jī)除外。)

還有一點(diǎn)不是特別理解的就是process.terminate() 發(fā)送出信號(hào)是SIGTERM number是15，但是第一次signal_handler收到信號(hào)卻是number=17，如果我要去處理15的信號(hào)，就會(huì)導(dǎo)致前一個(gè)進(jìn)程不能kill掉的問(wèn)題。歡迎有對(duì)信號(hào)量比較熟悉的大佬，前來(lái)指點(diǎn)迷津，不甚感謝。

采用multiprocessing.Event 來(lái)實(shí)現(xiàn)配置熱加載

實(shí)現(xiàn)邏輯是主進(jìn)程1 更新配置并發(fā)送指令。進(jìn)程2啟動(dòng)調(diào)度任務(wù)。

這時(shí)候當(dāng)主進(jìn)程1更新好配置之后，發(fā)送指令給進(jìn)程2，這時(shí)候的指令就是用Event一個(gè)異步事件通知。

直接上代碼

scheduler 函數(shù)

def scheduler():
    while True:
        print('wait message...')
        case_configurations = scheduler_notify_queue.get()
        print(f"Got case configurations {case_configurations=}...")

        task_schedule_event.set() # 設(shè)置set之后, is_set 為True

        print(f"Schedule will start ...")
        while task_schedule_event.is_set(): # is_set 為True的話，那么任務(wù)就會(huì)一直執(zhí)行
            run(case_configurations)

        print("Clearing all scheduling job ...")

event_scheduler 函數(shù)

def event_scheduler(case_config):

    scheduler_notify_queue.put(case_config)
    print(f"Put cases config to the Queue ...")

    task_schedule_event.clear() # clear之后，is_set 為False
    print(f"Clear scheduler jobs ...")

    print(f"Schedule job ...")

完整代碼如下

import multiprocessing
import time


scheduler_notify_queue = multiprocessing.Queue()
task_schedule_event = multiprocessing.Event()


def run(case_configurations: str):
    print(f'{case_configurations} running...')
    time.sleep(3)


def scheduler():
    while True:
        print('wait message...')
        case_configurations = scheduler_notify_queue.get()

        print(f"Got case configurations {case_configurations=}...")
        task_schedule_event.set()

        print(f"Schedule will start ...")
        while task_schedule_event.is_set():
            run(case_configurations)

        print("Clearing all scheduling job ...")


def event_scheduler(case_config: str):

    scheduler_notify_queue.put(case_config)
    print(f"Put cases config to the Queue ...")

    task_schedule_event.clear()
    print(f"Clear scheduler jobs ...")

    print(f"Schedule job ...")


def main():
    scheduler_notify_queue.put('1')
    p = multiprocessing.Process(target=scheduler)
    p.start()

    count = 1
    print(f'{count=}')
    while True:
        if count == 5:
            event_scheduler('100')
        if count == 10:
            event_scheduler('200')
        count += 1
        time.sleep(1)


if __name__ == '__main__':
    main()

執(zhí)行結(jié)果如下

wait message...
Got case configurations case_configurations='1'...
Schedule will start ...
1 running...
1 running...
Put cases config to the Queue ...
Clear scheduler jobs ...
Schedule job ...
Clearing all scheduling job ...
wait message...
Got case configurations case_configurations='100'...
Schedule will start ...
100 running...
Put cases config to the Queue ...
Clear scheduler jobs ...
Schedule job ...
Clearing all scheduling job ...
wait message...
Got case configurations case_configurations='200'...
Schedule will start ...
200 running...
200 running...

總結(jié)

使用Event事件通知，代碼不易出錯(cuò)，代碼編寫少，易讀。相比之前信號(hào)量的方法，推薦大家多使用這種方式。

使用多線程或協(xié)程的方式，其實(shí)和上述實(shí)現(xiàn)方式一致。唯一區(qū)別就是調(diào)用了不同庫(kù)中，queue和 event.

# threading
scheduler_notify_queue = queue.Queue()
task_schedule_event = threading.Event()

# async
scheduler_notify_queue = asyncio.Queue()
task_schedule_event = asyncio.Event()