如何将python asyncio与线程结合起来？

问题描述 投票：0回答：4

我已经使用 Python asyncio 和 aiohttp 成功构建了一个 RESTful 微服务，它监听 POST 事件以从各种 feeder 收集实时事件。

然后构建一个内存结构，将最后 24 小时的事件缓存在嵌套的 defaultdict/deque 结构中。

现在我想定期检查该结构到光盘，最好使用pickle。

由于内存结构可能>100MB，我希望避免在检查结构时耽误传入事件处理。

我宁愿创建结构的快照副本（例如深度复制），然后花时间将其写入磁盘并按照预设的时间间隔重复。

我一直在寻找有关如何组合线程（线程是否是最佳解决方案？）和 asyncio 的示例，但找不到对我有帮助的东西。

非常感谢任何入门指南！

python multithreading python-asyncio aiohttp
4个回答
99
投票

使用 

BaseEventLoop.run_in_executor
:

将方法委托给线程或子进程非常简单 
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

def cpu_bound_operation(x):
    time.sleep(x) # This is some operation that is CPU-bound

@asyncio.coroutine
def main():
    # Run cpu_bound_operation in the ProcessPoolExecutor
    # This will make your coroutine block, but won't block
    # the event loop; other coroutines can run in meantime.
    yield from loop.run_in_executor(p, cpu_bound_operation, 5)


loop = asyncio.get_event_loop()
p = ProcessPoolExecutor(2) # Create a ProcessPool with 2 processes
loop.run_until_complete(main())

至于用

ProcessPoolExecutor
还是
ThreadPoolExecutor
，这个就很难说了；腌制一个大对象肯定会消耗一些 CPU 周期，这最初会让你认为 
ProcessPoolExecutor
是正确的选择。但是，将 100MB 对象传递到池中的 
Process
需要在主进程中对实例进行酸洗，通过 IPC 将字节发送到子进程，在子进程中取消酸洗，然后再次酸洗它，这样您就可以将其写入磁盘。鉴于此，我的猜测是，pickling/unpickling 开销将足够大，因此您最好使用 ThreadPoolExecutor，即使您会因为 GIL 而受到性能影响。
也就是说，测试这两种方法并找出答案非常简单，所以你不妨这样做。

    





    

    
    
    

    

        

            

                

                    
18
投票

                    
                

            

        

        
run_in_executor
，但我发现这个函数在大多数情况下有点恶心，因为它需要 
partial()
 作为关键字参数，而且除了单个执行器和默认事件循环之外，我从不使用任何其他东西来调用它。因此，我用合理的默认值和自动关键字参数处理围绕它制作了一个方便的包装器。


from time import sleep
import asyncio as aio
loop = aio.get_event_loop()

class Executor:
    """In most cases, you can just use the 'execute' instance as a
    function, i.e. y = await execute(f, a, b, k=c) => run f(a, b, k=c) in
    the executor, assign result to y. The defaults can be changed, though,
    with your own instantiation of Executor, i.e. execute =
    Executor(nthreads=4)"""
    def __init__(self, loop=loop, nthreads=1):
        from concurrent.futures import ThreadPoolExecutor
        self._ex = ThreadPoolExecutor(nthreads)
        self._loop = loop
    def __call__(self, f, *args, **kw):
        from functools import partial
        return self._loop.run_in_executor(self._ex, partial(f, *args, **kw))
execute = Executor()

...

def cpu_bound_operation(t, alpha=30):
    sleep(t)
    return 20*alpha

async def main():
    y = await execute(cpu_bound_operation, 5, alpha=-2)

loop.run_until_complete(main())


    



    

    
    
    

    

        

            

                

                    
9
投票

                    
                

            

        

        
loop.call_soon_threadsafe
 和 
asyncio.Queue
 作为沟通的中间渠道。

Python 3 的当前文档还有一个关于 
使用 asyncio 开发 - 并发和多线程
: 的部分
import asyncio

# This method represents your blocking code
def blocking(loop, queue):
    import time
    while True:
        loop.call_soon_threadsafe(queue.put_nowait, 'Blocking A')
        time.sleep(2)
        loop.call_soon_threadsafe(queue.put_nowait, 'Blocking B')
        time.sleep(2)

# This method represents your async code
async def nonblocking(queue):
    await asyncio.sleep(1)
    while True:
        queue.put_nowait('Non-blocking A')
        await asyncio.sleep(2)
        queue.put_nowait('Non-blocking B')
        await asyncio.sleep(2)

# The main sets up the queue as the communication channel and synchronizes them
async def main():
    queue = asyncio.Queue()
    loop = asyncio.get_running_loop()

    blocking_fut = loop.run_in_executor(None, blocking, loop, queue)
    nonblocking_task = loop.create_task(nonblocking(queue))

    running = True  # use whatever exit condition
    while running:
        # Get messages from both blocking and non-blocking in parallel
        message = await queue.get()
        # You could send any messages, and do anything you want with them
        print(message)

asyncio.run(main())



如何
发送异步任务以在其他线程中循环运行
也可能对您有帮助。
如果您需要更“强大”的示例，请查看我的
从线程代码启动异步任务的包装器
。它会为你处理线程安全部分（大部分）并让你做这样的事情：
# See https://gist.github.com/Lonami/3f79ed774d2e0100ded5b171a47f2caf for the full example

async def async_main(queue):
    # your async code can go here
    while True:
        command = await queue.get()
        if command.id == 'print':
            print('Hello from async!')
        elif command.id == 'double':
            await queue.put(command.data * 2)

with LaunchAsync(async_main) as queue:
    # your threaded code can go here
    queue.put(Command('print'))
    queue.put(Command('double', 7))
    response = queue.get(timeout=1)
    print('The result of doubling 7 is', response)


    



    

    
    
    

    

        

            

                

                    
0
投票

                    
                

            

        

        
run_in_executor()
 对您来说就足够了，直到复杂的实时变化开始破坏您的结构状态。由于复制和酸洗不会立即完成，因此不完整的更改
可能会进入转储
。在这种情况下，您必须同步对结构的访问以避免这种情况。
在简单的情况下，你只需要这样的东西：


lock = asyncio.Lock()

async with lock:
    await loop.run_in_executor(...)

...

async with lock:
    # do something else in other tasks



但是，如果该函数花费的时间太长，则需要与您的结构配合使用的所有其他任务都将不得不等待。即使使用该结构已经是安全的。您可以考虑仅在正确的位置直接在函数中进行同步，但是您不能在异步函数之外使用
asyncio.Lock
，它也不是线程安全的，并且使用
threading
您会阻塞事件循环。 
但是，使用 
aiologic
 可以执行不阻塞事件循环的同步（我是 
aiologic
 的创建者）。使用 aiologic.Lock，您可以执行以下操作：
lock = aiologic.Lock()

with lock:
    structure_copy = copy.deepcopy(structure)

...

async with lock:
    # do something else in async tasks



您甚至不必在当前事件循环中创建 
aiologic.Lock
 的实例。因此，您可以安全地使实例成为结构的成员，并在异步和同步环境中使用它。这是创建通用线程安全对象的方法。
如果您关心性能，或者如果您有更复杂的情况，请查看 
aiologic
 提供的其他原语。例如，
什么原语可以用作锁
。
    
	

    

    


  

  

    
最新问题


  





  

    © www.soinside.com 2019 - 2024. All rights reserved.