Python 3 - 带有 AsyncIO/APScheduler 的全局变量

Question

已经为此苦苦挣扎了一段时间。

基于此线程：在函数中使用全局变量而不是创建它们的函数

我应该能够通过在特定时间安排任务来更新 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（因此可以依赖 have 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，则需要使用 amultiprocessing.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()将 a 返回Proxy到 shared Value，并且Proxy可以在run_in_executor不引发异常的情况下传递给它。