Python 3 - AsyncIO/APScheduler的全局变量

Question

一直在努力。

在此基础上螺纹：Using global variables in a function other than the one that created them

我应该能够在特定的时间调度任务来更新由thread_2使用的变量。

代码：在

import asyncio 
from concurrent.futures import ProcessPoolExecutor 
from apscheduler.schedulers.asyncio import AsyncIOScheduler 
from datetime import datetime 
import time 


def day_limits(): 

     global variable 
     variable = 90 
     print ('Day Variable: ',variable) 

def night_limits(): 

     global variable 
     variable = 65 
     print ('Night Variable: ',variable) 


def thread_2(): 


    while True: 

     c_hour = int(datetime.now().strftime("%H")) 
     c_min = int(datetime.now().strftime("%M")) 
     c_sec = int(datetime.now().strftime("%S")) 

     print ('%02d:%02d:%02d - Variable: %d ' % (c_hour,c_min,c_sec,variable)) 

     time.sleep(2) 


if __name__ == "__main__": 

    variable = 60 

    scheduler = AsyncIOScheduler() 
    scheduler.add_job(day_limits, 'cron', hour=7,misfire_grace_time=3600,timezone='GB') 
    scheduler.add_job(night_limits, 'cron', hour=19, minute=32,misfire_grace_time=3600,timezone='GB') 
    scheduler.start() 

    scheduler.print_jobs() 

    executor = ProcessPoolExecutor(1) 
    loop = asyncio.get_event_loop() 
    baa = asyncio.async(loop.run_in_executor(executor, thread_2)) 


    try: 
     loop.run_forever() 

    except (KeyboardInterrupt, Exception): 
     loop.stop() 
     scheduler.shutdown() 

结果：

19:31:54 - Variable: 60 
19:31:56 - Variable: 60 
19:31:58 - Variable: 60  
Night Variable: 65 
19:32:00 - Variable: 60 
19:32:02 - Variable: 60 

我失去了一些东西，但我什么也看不见！

想法？

谢谢！

Answer 1

因为您使用的是ProcessPoolExecutor，所以需要使用进程安全对象来代替普通整数。如果您只需要支持Linux（因此可以依靠fork()），则可以使用普通的全局multiprocessing.Value来执行此操作。

import asyncio 
import multiprocessing 
from concurrent.futures import ProcessPoolExecutor 
from apscheduler.schedulers.asyncio import AsyncIOScheduler 
from datetime import datetime 
import time 


def day_limits(): 
     variable.value = 90 
     print ('Day Variable: ',variable.value) 

def night_limits(): 
     variable.value = 65 
     print ('Night Variable: ',variable.value) 


def thread_2(): 
    while True: 
     c_hour = int(datetime.now().strftime("%H")) 
     c_min = int(datetime.now().strftime("%M")) 
     c_sec = int(datetime.now().strftime("%S")) 

     print ('%02d:%02d:%02d - Variable: %d ' % (c_hour,c_min,c_sec,variable.value)) 

     time.sleep(2) 


if __name__ == "__main__": 
    variable = multiprocessing.Value('i', 60) 

    scheduler = AsyncIOScheduler() 
    scheduler.add_job(day_limits, 'cron', hour=7,misfire_grace_time=3600,timezone='GB') 
    scheduler.add_job(night_limits, 'cron', hour=19, minute=32,misfire_grace_time=3600,timezone='GB') 
    scheduler.start() 

    scheduler.print_jobs() 

    executor = ProcessPoolExecutor(1) 
    loop = asyncio.get_event_loop() 
    baa = asyncio.async(loop.run_in_executor(executor, thread_2)) 


    try: 
     loop.run_forever() 

    except (KeyboardInterrupt, Exception): 
     loop.stop() 
     scheduler.shutdown() 

如果你需要同时支持Windows和Linux，你需要使用一个multiprocessing.Manager创建Value对象，并明确地传递对象的功能，你在Executor运行：

import asyncio 
import multiprocessing 
from concurrent.futures import ProcessPoolExecutor 
from apscheduler.schedulers.asyncio import AsyncIOScheduler 
from datetime import datetime 
import time 


def day_limits(): 
     variable.value = 90 
     print ('Day Variable: ',variable.value) 

def night_limits(): 
     variable.value = 65 
     print ('Night Variable: ',variable.value) 


def thread_2(variable): 
    while True: 
     c_hour = int(datetime.now().strftime("%H")) 
     c_min = int(datetime.now().strftime("%M")) 
     c_sec = int(datetime.now().strftime("%S")) 

     print ('%02d:%02d:%02d - Variable: %d ' % (c_hour,c_min,c_sec,variable.value)) 

     time.sleep(2) 


if __name__ == "__main__": 

    m = multiprocessing.Manager() 
    variable = m.Value('i', 60) 

    scheduler = AsyncIOScheduler() 
    scheduler.add_job(day_limits, 'cron', hour=7,misfire_grace_time=3600,timezone='GB') 
    scheduler.add_job(night_limits, 'cron', hour=19, minute=32,misfire_grace_time=3600,timezone='GB') 
    scheduler.start() 

    scheduler.print_jobs() 

    executor = ProcessPoolExecutor(1) 
    loop = asyncio.get_event_loop() 
    baa = asyncio.async(loop.run_in_executor(executor, thread_2, variable)) # Need to pass variable explicitly 

    try: 
     loop.run_forever() 

    except (KeyboardInterrupt, Exception): 
     loop.stop() 
     scheduler.shutdown() 

因为Windows缺少fork支持，所以您需要明确地将Value传递给您在Executor中运行的函数。如果你不这样做，那么子进程会说变量不存在。但是，由于您明确将Value传递给run_in_executor方法，因此不能使用普通的multiprocessing.Value - 您将得到一个RuntimeError，表示“同步对象应该仅通过继承在进程之间共享”。

使用multiprocessing.Manager解决这个问题; multiprocessing.Manager启动一个可以创建和管理进程共享对象的进程。调用m.Value()会将Proxy返回给共享的Value，并且Proxy可以传递到run_in_executor而不引发异常。

Answer 2

问题很简单：您正在使用ProcessPoolExecutor。所以这些工作是在另一个拥有自己内存空间的进程中运行的。 variable在那里设置正确，但它在父进程中具有不同的值（60），该值永远不会更改。