基于Python實現(xiàn)配置熱加載的方法詳解
更新時間：2022年07月09日 16:47:06 作者：煙熏柿子學(xué)編程
所謂的配置熱加載，也就是說當(dāng)服務(wù)收到配置更新消息之后，我們不用重啟服務(wù)就可以使用最新的配置去執(zhí)行任務(wù)。本文將介紹如何用Python實現(xiàn)配置熱加載，需要的可以參考一下
背景
由于最近工作需求，需要在已有項目添加一個新功能，實現(xiàn)配置熱加載的功能。所謂的配置熱加載，也就是說當(dāng)服務(wù)收到配置更新消息之后，我們不用重啟服務(wù)就可以使用最新的配置去執(zhí)行任務(wù)。
如何實現(xiàn)
下面我分別采用多進(jìn)程、多線程、協(xié)程的方式去實現(xiàn)配置熱加載。
使用多進(jìn)程實現(xiàn)配置熱加載
如果我們代碼實現(xiàn)上使用多進(jìn)程， 主進(jìn)程1來更新配置并發(fā)送指令，任務(wù)的調(diào)用是進(jìn)程2，如何實現(xiàn)配置熱加載呢？
使用signal信號量來實現(xiàn)熱加載
當(dāng)主進(jìn)程收到配置更新的消息之后(配置讀取是如何收到配置更新的消息的? 這里我們暫不討論)， 主進(jìn)程就向進(jìn)子程1發(fā)送kill信號，子進(jìn)程1收到kill的信號就退出，之后由信號處理函數(shù)來啟動一個新的進(jìn)程，使用最新的配置文件來繼續(xù)執(zhí)行任務(wù)。
main 函數(shù)
def main():
# 啟動一個進(jìn)程執(zhí)行任務(wù)
p1 = Process(target=run, args=("p1",))
p1.start()
monitor(p1, run) # 注冊信號
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
# 這個循環(huán)是為了忽略SIGTERM發(fā)出的信號，避免搶占了主進(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é)
好處：使用信號量可以處理多進(jìn)程之間通信的問題。
壞處：代碼不好寫，寫出來代碼不好理解。信號量使用必須要很熟悉，不然很容易自己給自己寫了一個bug.(所有初學(xué)者慎用，老司機(jī)除外。)
還有一點不是特別理解的就是process.terminate() 發(fā)送出信號是SIGTERM number是15，但是第一次signal_handler收到信號卻是number=17，如果我要去處理15的信號，就會導(dǎo)致前一個進(jìn)程不能kill掉的問題。歡迎有對信號量比較熟悉的大佬，前來指點迷津，不甚感謝。
采用multiprocessing.Event 來實現(xiàn)配置熱加載
實現(xiàn)邏輯是主進(jìn)程1 更新配置并發(fā)送指令。進(jìn)程2啟動調(diào)度任務(wù)。
這時候當(dāng)主進(jìn)程1更新好配置之后，發(fā)送指令給進(jìn)程2，這時候的指令就是用Event一個異步事件通知。
直接上代碼
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ù)就會一直執(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事件通知，代碼不易出錯，代碼編寫少，易讀。相比之前信號量的方法，推薦大家多使用這種方式。
使用多線程或協(xié)程的方式，其實和上述實現(xiàn)方式一致。唯一區(qū)別就是調(diào)用了不同庫中，queue和 event.
# threading scheduler_notify_queue = queue.Queue() task_schedule_event = threading.Event() # async scheduler_notify_queue = asyncio.Queue() task_schedule_event = asyncio.Event()
結(jié)語
具體的實現(xiàn)的方式有很多，也各自有各自的優(yōu)劣勢。我們需要去深刻理解到需求本身，才去做技術(shù)選型。
以上就是基于Python實現(xiàn)配置熱加載的方法詳解的詳細(xì)內(nèi)容，更多關(guān)于Python配置熱加載的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
這篇文章主要介紹了Cpy和Python的效率對比,本文用一個循環(huán) 100000000 遍的代碼對比了Cpy和Python運行效率測試,需要的朋友可以參考下2015-03-03
這篇文章主要給大家介紹了關(guān)于如何利用Pyecharts可視化微信好友的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家學(xué)習(xí)或者使用Pyecharts具有一定的參考學(xué)習(xí)價值，需要的朋友們下面來一起學(xué)習(xí)學(xué)習(xí)吧2019-07-07
這篇文章主要介紹了Python os模塊常用方法和屬性總結(jié),文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-02-02
這篇文章主要介紹了python 瀑布線指標(biāo)編寫實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-06-06
PyCharm無法調(diào)用numpy(報錯ModuleNotFoundError:No?module?named?&a
本文主要介紹了PyCharm無法調(diào)用numpy(報錯ModuleNotFoundError:No?module?named?'numpy')，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-02-02
python使用pynput庫操作、監(jiān)控你的鼠標(biāo)和鍵盤
這篇文章主要介紹了python使用pynput庫操作、監(jiān)控你的鼠標(biāo)和鍵盤，幫助大家更好的理解和學(xué)習(xí)使用python，感興趣的朋友可以了解下2021-03-03
本文詳細(xì)介紹了如何使用Python查找字符串中包含的多個元素,包括基本字符串操作和使用正則表達(dá)式進(jìn)行高級搜索,具有一定的參考價值,感興趣的可以了解一下2024-03-03
這篇文章主要為大家詳細(xì)介紹了python3將視頻流保存為本地視頻文件，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-06-06
Python對Tornado請求與響應(yīng)的數(shù)據(jù)處理
這篇文章主要介紹了Python對Tornado請求與響應(yīng)的數(shù)據(jù)處理,需要的朋友可以參考下2020-02-02
基于opencv和pillow實現(xiàn)人臉識別系統(tǒng)(附demo)
人臉識別就是一個程序能識別給定圖像或視頻中的人臉，本文主要介紹了opencv和pillow實現(xiàn)人臉識別系統(tǒng)，本文不涉及分類器、訓(xùn)練識別器等算法原理，感興趣的可以了解一下2021-11-11