使用asyncio运行并发mongoengine查询

我有4个基本上构建查询并执行它们的函数。我想让它们使用asyncio同时运行。我的asyncio实现似乎是正确的，因为非MongoDB任务运行时应该如此（例如asyncio.sleep（））。以下是代码：使用asyncio运行并发mongoengine查询

loop = asyncio.new_event_loop() 
asyncio.set_event_loop(loop) 
tasks = [ 
       service.async_get_associate_opportunity_count_by_user(me, criteria), 
       service.get_new_associate_opportunity_count_by_user(me, criteria), 
       service.async_get_associate_favorites_count(me, criteria=dict()), 
       service.get_group_matched_opportunities_count_by_user(me, criteria) 
      ] 

available, new, favorites, group_matched = loop.run_until_complete(asyncio.gather(*tasks)) 

stats['opportunities']['available'] = available 
stats['opportunities']['new'] = new 
stats['opportunities']['favorites'] = favorites 
stats['opportunities']['group_matched'] = group_matched 

loop.close() 


# functions written in other file 

@asyncio.coroutine 
def async_get_ass(self, user, criteria=None, **kwargs): 
    start_time = time.time() 
    query = **query that gets built from some other functions** 

    opportunities = Opportunity.objects(query).count() 
    run_time = time.time() - start_time 
    print("runtime of available: {}".format(run_time)) 
    yield from asyncio.sleep(2) 
    return opportunities 

@asyncio.coroutine 
def get_new_associate_opportunity_count_by_user(self, user, criteria=None, **kwargs): 
    start_time = time.time() 
    query = **query that gets built from some other functions** 
    opportunities = Opportunity.objects(query).count() 
    run_time = time.time() - start_time 
    print("runtime of new: {}".format(run_time)) 
    yield from asyncio.sleep(2) 
    return opportunities 

@asyncio.coroutine 
def async_get_associate_favorites_count(self, user, criteria={}, **kwargs): 
    start_time = time.time() 
    query = **query that gets built from some other functions** 
    favorites = Opportunity.objects(query).count() 
    run_time = time.time() - start_time 
    print("runtime of favorites: {}".format(run_time)) 
    yield from asyncio.sleep(2) 
    return favorites 

@asyncio.coroutine 
def get_group_matched_opportunities_count_by_user(self, user, criteria=None, **kwargs): 
    start_time = time.time() 
    query = **query that gets built from some other functions** 
    opportunities = Opportunity.objects(query).count() 
    run_time = time.time() - start_time 
    print("runtime of group matched: {}".format(run_time)) 
    yield from asyncio.sleep(2) 
    return opportunities

yield from asyncio.sleep(2)只是表明函数是异步运行的。这里是在终端上的输出：

组的运行时间相匹配：喜好0.11431598663330078 运行时：0.0029871463775634766 时间戳功能的运行时间：新0.0004897117614746094 运行时：0.13006806373596191 总运行时间：可用0.15225648880004883 运行时2403.2700061798096 根据我的理解，除了由于睡眠功能而增加到总运行时间的2000ms之外，由于所有功能中的最大运行时间都是该值，所以它不应该超过155-160ms。

我目前正在研究motorengine（mongoengine 0.9.0的一个端口），它显然可以启用异步mongodb查询，但我认为我将无法使用它，因为我的模型已经使用mongoengine定义。有没有解决这个问题的方法？

来源

2015-09-10 Aleem Saadullah

您的查询不是并行运行的原因是因为无论何时在您的协同程序中运行Opportunity.objects(query).count()，整个事件循环都会阻塞，因为这些方法正在阻止IO。

所以你需要一个可以做异步/非阻塞IO的mongodb驱动程序。您尝试使用motorengine的方法正确，但据我所知它是为Tornado异步框架编写的。要使它与asyncio一起工作，您将不得不连接 Tornado和asycnio。见，http://tornado.readthedocs.org/en/latest/asyncio.html关于如何做到这一点。

另一种选择是使用asyncio-mongo，但它没有mongoengine兼容ORM，因此您可能必须重写大部分代码。

来源

2015-09-10 18:05:11

使用asyncio运行并发mongoengine查询

回答

相关问题