So implementieren Sie das Hotloaden von Konfigurationsdateien in Python

Hintergrund

Aufgrund aktueller Arbeitsanforderungen ist es erforderlich, dem bestehenden Projekt eine neue Funktion hinzuzufügen, um die Konfigurations-Hot-Reloading-Funktion zu implementieren. Das sogenannte Hot-Reloading der Konfiguration bedeutet, dass wir, nachdem der Dienst die Konfigurationsaktualisierungsnachricht erhalten hat, die neueste Konfiguration verwenden können, um Aufgaben auszuführen, ohne den Dienst neu starten zu müssen.

So implementieren Sie

Im Folgenden verwende ich Multiprozess-, Multithread- und Coroutine-Methoden, um das Hotloaden der Konfiguration zu implementieren.

Verwenden Sie mehrere Prozesse, um das Hotloaden der Konfiguration zu implementieren

Wenn wir bei der Codeimplementierung mehrere Prozesse verwenden, aktualisiert der Hauptprozess 1 die Konfiguration und sendet Anweisungen, und der Aufgabenaufruf ist Prozess 2. Wie implementiert man das Hotloaden der Konfiguration?

Verwenden Sie Signalsemaphore, um Hot Loading zu implementieren

Wenn der Hauptprozess die Konfigurationsaktualisierungsnachricht empfängt (wie empfängt der Konfigurationsleser die Konfigurationsaktualisierungsnachricht? Wir werden hier nicht darauf eingehen), fährt der Hauptprozess mit Unterprozess 1 fort sendet ein Kill-Signal. Nach dem Empfang des Kill-Signals startet die Signalverarbeitungsfunktion einen neuen Prozess und verwendet die neueste Konfigurationsdatei, um mit der Ausführung der Aufgabe fortzufahren.

Hauptfunktion

def main():
    # 启动一个进程执行任务
    p1 = Process(target=run, args=("p1",))
    p1.start()

    monitor(p1, run) # 注册信号
    processes["case100"] = p1 #将进程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 当前进程
        if num == 8:
            kill_process(processes["case100"]) # kill 当前进程
        if num == 12:
            kill_process(processes["case100"]) # kill 当前进程
        num += 1

signal_handler-Funktion

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

    if signum == 17:  # 17 is SIGCHILD 
        # 这个循环是为了忽略SIGTERM发出的信号，避免抢占了主进程发出的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")

Der vollständige Code lautet wie folgt:

#! /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=&#39;<stdout>&#39;>
        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__ == &#39;__main__&#39;:
    main()

Die Ausführungsergebnisse lauten wie folgt: Vorteile: Durch die Verwendung von Semaphoren kann die Kommunikation zwischen mehreren abgewickelt werden Prozessproblem.

Nachteile: Der Code ist schwer zu schreiben und der geschriebene Code ist schwer zu verstehen. Sie müssen mit der Verwendung von Semaphoren vertraut sein, sonst ist es einfach, einen Fehler selbst zu schreiben (Alle Anfänger sollten ihn mit Vorsicht verwenden, außer erfahrene Fahrer.) Eine andere Sache, die nicht besonders verstanden wird, ist der -Prozess. Beenden() Das gesendete Signal ist <code>SIGTERM und die Zahl ist 15, aber das erste Mal, dass signal_handler das Signal empfängt, ist Zahl=17, wenn ich es verarbeiten möchte das Signal von 15, dann Dies führt zu dem Problem, dass der vorherige Prozess nicht abgebrochen werden kann. Wer sich mit Semaphoren auskennt, kann uns gerne einen Rat geben. Vielen Dank.

Verwendung von multiprocessing.Event zum Implementieren des Konfigurations-Hotloadings

Die Implementierungslogik besteht darin, dass der Hauptprozess 1 die Konfiguration aktualisiert und Anweisungen sendet. Prozess 2 startet die Planungsaufgabe.

Zu diesem Zeitpunkt sendet der Hauptprozess 1, nachdem er die Konfiguration aktualisiert hat, eine Anweisung an Prozess 2. Die Anweisung zu diesem Zeitpunkt besteht darin, Event zu verwenden, um ein asynchrones Ereignis zu benachrichtigen. process.terminate() 发送出信号是SIGTERM number是15，但是第一次signal_handler收到信号却是number=17，如果我要去处理15的信号，就会导致前一个进程不能kill掉的问题。欢迎有对信号量比较熟悉的大佬，前来指点迷津，不甚感谢。

采用multiprocessing.Event 来实现配置热加载

实现逻辑是主进程1 更新配置并发送指令。进程2启动调度任务。

这时候当主进程1更新好配置之后，发送指令给进程2，这时候的指令就是用Event一个异步事件通知。

直接上代码

scheduler 函数

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

processes={&#39;case100&#39;: <Process name=&#39;Process-1&#39; pid=2533 parent=2532 started>}

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

processes={&#39;case100&#39;: <Process name=&#39;Process-1&#39; pid=2533 parent=2532 started>}

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

processes={&#39;case100&#39;: <Process name=&#39;Process-2&#39; pid=2577 parent=2532 started>}

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

processes={&#39;case100&#39;: <Process name=&#39;Process-2&#39; pid=2577 parent=2532 started>}

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

processes={&#39;case100&#39;: <Process name=&#39;Process-2&#39; pid=2577 parent=2532 started>}

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

processes={&#39;case100&#39;: <Process name=&#39;Process-3&#39; pid=2675 parent=2532 started>}

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

event_scheduler 函数

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

        task_schedule_event.set() # 设置set之后, is_set 为True

        print(f"Schedule will start ...")
        while task_schedule_event.is_set(): # is_set 为True的话，那么任务就会一直执行
            run(case_configurations)

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

完整代码如下

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()

总结

使用Event事件通知，代码不易出错，代码编写少，易读。相比之前信号量的方法，推荐大家多使用这种方式。

使用多线程或协程的方式，其实和上述实现方式一致。唯一区别就是调用了不同库中，queue和 event

Gehe direkt zum Code🎜🎜🎜Scheduler-Funktion🎜🎜

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...

🎜🎜event_scheduler-Funktion🎜🎜

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

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

🎜🎜Der vollständige Code lautet wie folgt🎜🎜rrreee🎜Die Ausführungsergebnisse lauten wie folgt🎜rrreee🎜 🎜Zusammenfassung🎜🎜 🎜Verwenden Sie das Ereignis Ereignisbenachrichtigung, der Code ist weniger fehleranfällig, der Code lässt sich weniger schreiben und ist leichter zu lesen. Im Vergleich zur vorherigen Semaphor-Methode wird empfohlen, diese Methode häufiger zu verwenden. 🎜🎜Die Verwendung von Multithreading oder Coroutine entspricht tatsächlich der oben genannten Implementierungsmethode. Der einzige Unterschied besteht darin, dass unterschiedliche Bibliotheken aufgerufen werden, queue und event.🎜rrreee

Das obige ist der detaillierte Inhalt vonSo implementieren Sie das Hotloaden von Konfigurationsdateien in Python. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!