我知道一般来说，对象不应该在多进程之间共享，并且可能由此产生问题。但我的要求是有必要这样做。
我有一个复杂的对象，其中包含所有漂亮的协程异步等待。 一个函数，它在其自己的单独进程中在此对象上运行长时间运行的进程。现在，我想在主进程中运行一个 IPython shell，并在该长时间运行的进程在另一个进程中运行时对这个复杂对象进行操作。
为了跨进程共享这个复杂的对象，我尝试了在SO上遇到的多处理BaseManager方法：
import multiprocessing
import multiprocessing.managers as m
class MyManager(m.BaseManager):
pass
MyManager.register('complex_asynio_based_class', complex_asynio_based_class)
manager = MyManager()
manager.start()
c = manager.complex_asynio_based_class()
process = multiprocessing.Process(
target=long_running_process,
args=(c,),
)
但这会产生错误：
Unserializable message: Traceback (most recent call last):
File "/usr/3.6/lib/python3.6/multiprocessing/managers.py", line 283, in serve_client
send(msg)
File "/usr/3.6/lib/python3.6/multiprocessing/connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "/usr/3.6/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
TypeError: can't pickle coroutine objects
它不起作用，因为对象中有协程。我想不出更好的解决方案来让它工作，我被困住了。
如果不是 Python，我会为长时间运行的进程生成一个线程，并且仍然能够对其进行操作。
如果我没记错的话，这应该是多进程应用程序运行后台进程和主进程的常见模式，主进程只对其执行一些只读操作，就像我的情况一样，而不是修改它。我想知道它通常是如何完成的？
无法拾取的复杂对象如何在多进程中共享？
正在运行的协程无法在进程之间自动共享，因为协程在拥有异步类的进程的特定事件循环内运行。协程具有无法被腌制的状态，即使可以，它在其事件循环的上下文之外也没有意义。
您可以做的是为异步类创建一个基于回调的适配器，每个协程方法都由一个基于回调的方法表示，该方法的语义是"开始执行 X 并在完成后调用此函数"。如果回调是多处理感知的，则可以从其他进程调用这些操作。然后，您可以在每个进程中启动一个事件循环，并在基于回调的代理调用上创建一个协程外观。
例如，考虑一个简单的异步类：
class Async:
async def repeat(self, n, s):
for i in range(n):
print(s, i, os.getpid())
await asyncio.sleep(.2)
return s
基于回调的适配器可以使用公共asyncioAPI 将repeat协程转换为 JavaScript "回调地狱"风格的经典异步函数：
class CallbackAdapter:
def repeat_start(self, n, s, on_success):
fut = asyncio.run_coroutine_threadsafe(
self._async.repeat(n, s), self._loop)
# Once the coroutine is done, notify the caller.
fut.add_done_callback(lambda _f: on_success(fut.result()))
(转换可以自动化，上面手动编写的代码只是说明了概念。
CallbackAdapter可以注册到多处理，因此不同的进程可以通过多处理提供的代理启动适配器的方法(以及原始异步协程)。这只要求传递的回调on_success多处理友好。
作为最后一步，可以绕一圈，为基于回调的 API 创建一个异步适配器 (！)，在另一个进程中也启动一个事件循环，还可以使用 asyncio 和async def。此适配器对适配器类将具有功能齐全的repeat协程，该协程有效地代理原始Async.repeat协程，而无需尝试将协程状态。
下面是上述方法的示例实现：
import asyncio, multiprocessing.managers, threading, os
class Async:
# The async class we are bridging. This class is unaware of multiprocessing
# or of any of the code that follows.
async def repeat(self, n, s):
for i in range(n):
print(s, i, 'pid', os.getpid())
await asyncio.sleep(.2)
return s
def start_asyncio_thread():
# Since the manager controls the main thread, we have to spin up the event
# loop in a dedicated thread and use asyncio.run_coroutine_threadsafe to
# submit stuff to the loop.
setup_done = threading.Event()
loop = None
def loop_thread():
nonlocal loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
setup_done.set()
loop.run_forever()
threading.Thread(target=loop_thread).start()
setup_done.wait()
return loop
class CallbackAdapter:
_loop = None
# the callback adapter to the async class, also running in the
# worker process
def __init__(self, obj):
self._async = obj
if CallbackAdapter._loop is None:
CallbackAdapter._loop = start_asyncio_thread()
def repeat_start(self, n, s, on_success):
# Submit a coroutine to the event loop and obtain a Task/Future. This
# is normally done with loop.create_task, but repeat_start will be
# called from the main thread, owned by the multiprocessng manager,
# while the event loop will run in a separate thread.
future = asyncio.run_coroutine_threadsafe(
self._async.repeat(n, s), self._loop)
# Once the coroutine is done, notify the caller.
# We could propagate exceptions by accepting an additional on_error
# callback, and nesting fut.result() in a try/except that decides
# whether to call on_success or on_error.
future.add_done_callback(lambda _f: on_success(future.result()))
def remote_event_future(manager):
# Return a function/future pair that can be used to locally monitor an
# event in another process.
#
# The returned function and future have the following property: when the
# function is invoked, possibly in another process, the future completes.
# The function can be passed as a callback argument to a multiprocessing
# proxy object and therefore invoked by a different process.
loop = asyncio.get_event_loop()
result_pipe = manager.Queue()
future = loop.create_future()
def _wait_for_remote():
result = result_pipe.get()
loop.call_soon_threadsafe(future.set_result, result)
t = threading.Thread(target=_wait_for_remote)
t.start()
return result_pipe.put, future
class AsyncAdapter:
# The async adapter for a callback-based API, e.g. the CallbackAdapter.
# Designed to run in a different process and communicate to the callback
# adapter via a multiprocessing proxy.
def __init__(self, cb_proxy, manager):
self._cb = cb_proxy
self._manager = manager
async def repeat(self, n, s):
set_result, future = remote_event_future(self._manager)
self._cb.repeat_start(n, s, set_result)
return await future
class CommManager(multiprocessing.managers.SyncManager):
pass
CommManager.register('Async', Async)
CommManager.register('CallbackAdapter', CallbackAdapter)
def get_manager():
manager = CommManager()
manager.start()
return manager
def other_process(manager, cb_proxy):
print('other_process (pid %d)' % os.getpid())
aadapt = AsyncAdapter(cb_proxy, manager)
loop = asyncio.get_event_loop()
# Create two coroutines printing different messages, and gather their
# results.
results = loop.run_until_complete(asyncio.gather(
aadapt.repeat(3, 'message A'),
aadapt.repeat(2, 'message B')))
print('coroutine results (pid %d): %s' % (os.getpid(), results))
print('other_process (pid %d) done' % os.getpid())
def start_other_process(loop, manager, async_proxy):
cb_proxy = manager.CallbackAdapter(async_proxy)
other = multiprocessing.Process(target=other_process,
args=(manager, cb_proxy,))
other.start()
return other
def main():
loop = asyncio.get_event_loop()
manager = get_manager()
async_proxy = manager.Async()
# Create two external processes that drive coroutines in our event loop.
# Note that all messages are printed with the same PID.
start_other_process(loop, manager, async_proxy)
start_other_process(loop, manager, async_proxy)
loop.run_forever()
if __name__ == '__main__':
main()
代码在 Python 3.5 上正常运行，但由于多处理中的错误，在 3.6 和 3.7 上失败。
我已经使用多处理模块和异步模块一段时间了。
您不会在进程之间共享对象。在一个进程中创建一个对象(引用)，返回一个代理对象并与其他进程共享。其他进程使用代理对象来调用引用的方法。
在代码中，引用对象是complex_asynio_based_class实例。
这是您可以引用的愚蠢代码。主线程是运行UDP服务器和其他几个异步操作的单个异步循环。长时间运行的进程只需检查循环状态。
import multiprocessing
import multiprocessing.managers as m
import asyncio
import logging
import time
logging.basicConfig(filename="main.log", level=logging.DEBUG)
class MyManager(m.BaseManager):
pass
class sinkServer(asyncio.Protocol):
def connection_made(self, transport):
self.transport = transport
def datagram_received(self, data, addr):
message = data.decode()
logging.info('Data received: {!r}'.format(message))
class complex_asynio_based_class:
def __init__(self, addr=('127.0.0.1', '8080')):
self.loop = asyncio.new_event_loop()
listen = self.loop.create_datagram_endpoint(sinkServer, local_addr=addr,
reuse_address=True, reuse_port=True)
self.loop.run_until_complete(listen)
for name, delay in zip("abcdef", (1,2,3,4,5,6)):
self.loop.run_until_complete(self.slow_op(name, delay))
def run(self):
self.loop.run_forever()
def stop(self):
self.loop.stop()
def is_running(self):
return self.loop.is_running()
async def slow_op(self, name, delay):
logging.info("my name: {}".format(name))
asyncio.sleep(delay)
def long_running_process(co):
logging.debug('address: {!r}'.format(co))
logging.debug("status: {}".format(co.is_running()))
time.sleep(6)
logging.debug("status: {}".format(co.is_running()))
MyManager.register('complex_asynio_based_class', complex_asynio_based_class)
manager = MyManager()
manager.start()
c = manager.complex_asynio_based_class()
process = multiprocessing.Process(
target=long_running_process,
args=(c,),
)
process.start()
c.run() #run the loop