小編寫這篇文章的目的，主要是給大家講解一下，關(guān)于實(shí)現(xiàn)配置熱加載的方法，具體是怎么操作呢？下面就給大家詳細(xì)的解答下。

　　背景

　　由于最近有相關(guān)的工作需求，需要進(jìn)行增添相關(guān)的新功能，實(shí)現(xiàn)配置熱加載的功能。所謂的配置熱加載，也就是說當(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來更新配置并發(fā)送指令，任務(wù)的調(diào)用是進(jìn)程2，如何實(shí)現(xiàn)配置熱加載呢？

　　使用signal信號(hào)量來實(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ù)來啟動(dòng)一個(gè)新的進(jìn)程，使用最新的配置文件來繼續(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)#注冊信號(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)程之間通信的問題。

　　自媒體培訓(xùn)

　　壞處：代碼不好寫，寫出來代碼不好理解。信號(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掉的問題。歡迎有對(duì)信號(hào)量比較熟悉的大佬，前來指點(diǎn)迷津，不甚感謝。

　　采用multiprocessing.Event來實(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)用了不同庫中，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é)的，關(guān)于python方面的一些知識(shí)了，希望可以給大家?guī)韼椭?/p>