时间戳

首页 > 解决方案 > django celery 和 asyncio - 循环参数必须大约每 3 分钟与 Future 一致

python - django celery 和 asyncio - 循环参数必须大约每 3 分钟与 Future 一致

问题描述

我使用 django celery 和 celery beat 来运行周期性任务。我每分钟运行一次任务以通过 SNMP 获取一些数据。

我的函数使用 asyncio，如下所示。我已经检查了代码以检查循环是否关闭并创建一个新的。

但似乎正在发生的是每隔几个任务，我就会失败，并且在 Django-tasks-results 数据库中我有以下回溯。似乎每3分钟就有一次失败，但每分钟都有一次成功，没有失败

错误：

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/celery/app/trace.py", line 374, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/celery/app/trace.py", line 629, in __protected_call__
    return self.run(*args, **kwargs)
  File "/itapp/itapp/monitoring/tasks.py", line 32, in link_data
    return get_link_data()
  File "/itapp/itapp/monitoring/jobs/link_monitoring.py", line 209, in get_link_data
    done, pending = loop.run_until_complete(asyncio.wait(tasks))
  File "/usr/local/lib/python3.6/asyncio/base_events.py", line 468, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.6/asyncio/tasks.py", line 311, in wait
    fs = {ensure_future(f, loop=loop) for f in set(fs)}
  File "/usr/local/lib/python3.6/asyncio/tasks.py", line 311, in <setcomp>
    fs = {ensure_future(f, loop=loop) for f in set(fs)}
  File "/usr/local/lib/python3.6/asyncio/tasks.py", line 514, in ensure_future
    raise ValueError('loop argument must agree with Future')
ValueError: loop argument must agree with Future

功能：

async def retrieve_data(link):
    poll_interval = 60
    results = []
    # credentials:
    link_mgmt_ip = link.mgmt_ip
    link_index = link.interface_index
    snmp_user = link.device_circuit_subnet.device.snmp_data.name
    snmp_auth = link.device_circuit_subnet.device.snmp_data.auth
    snmp_priv = link.device_circuit_subnet.device.snmp_data.priv
    hostname = link.device_circuit_subnet.device.hostname
    print('polling data for {} on {}'.format(hostname,link_mgmt_ip))

    # first poll for speeds
    download_speed_data_poll1 = snmp_get(link_mgmt_ip, down_speed_oid % link_index ,snmp_user, snmp_auth, snmp_priv)

    # check we were able to poll
    if 'timeout' in str(get_snmp_value(download_speed_data_poll1)).lower():
        return 'timeout trying to poll {} - {}'.format(hostname ,link_mgmt_ip)
    upload_speed_data_poll1 = snmp_get(link_mgmt_ip, up_speed_oid % link_index, snmp_user, snmp_auth, snmp_priv) 

    # wait for poll interval
    await asyncio.sleep(poll_interval)

    # second poll for speeds
    download_speed_data_poll2 = snmp_get(link_mgmt_ip, down_speed_oid % link_index, snmp_user, snmp_auth, snmp_priv)
    upload_speed_data_poll2 = snmp_get(link_mgmt_ip, up_speed_oid % link_index, snmp_user, snmp_auth, snmp_priv)    

    # create deltas for speed
    down_delta = int(get_snmp_value(download_speed_data_poll2)) - int(get_snmp_value(download_speed_data_poll1))
    up_delta = int(get_snmp_value(upload_speed_data_poll2)) - int(get_snmp_value(upload_speed_data_poll1))

    # set speed results
    download_speed = round((down_delta * 8 / poll_interval) / 1048576)
    upload_speed = round((up_delta * 8 / poll_interval) / 1048576)

    # get description and interface state
    int_desc = snmp_get(link_mgmt_ip, int_desc_oid % link_index, snmp_user, snmp_auth, snmp_priv)   
    int_state = snmp_get(link_mgmt_ip, int_state_oid % link_index, snmp_user, snmp_auth, snmp_priv)

    ...
    return results

def get_link_data():  
    mgmt_ip = Subquery(
        DeviceCircuitSubnets.objects.filter(device_id=OuterRef('device_circuit_subnet__device_id'),subnet__subnet_type__poll=True).values('subnet__subnet')[:1])
    link_data = LinkTargets.objects.all() \
                .select_related('device_circuit_subnet') \
                .select_related('device_circuit_subnet__device') \
                .select_related('device_circuit_subnet__device__snmp_data') \
                .select_related('device_circuit_subnet__subnet') \
                .select_related('device_circuit_subnet__circuit') \
                .annotate(mgmt_ip=mgmt_ip) 
    tasks = []
    loop = asyncio.get_event_loop()
    if asyncio.get_event_loop().is_closed():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(asyncio.new_event_loop())

    for link in link_data:
        tasks.append(asyncio.ensure_future(retrieve_data(link)))

    if tasks:
        start = time.time()  
        done, pending = loop.run_until_complete(asyncio.wait(tasks))
        loop.close()  

        results = []
        for completed_task in done:
            results.append(completed_task.result()[0])

        end = time.time() 
        print("Poll time: {}".format(end - start))
        return 'Link data updated for {}'.format(' \n '.join(results))
    else:
        return 'no tasks defined'

标签： pythondjangocelerypython-asyncio

解决方案

来自 user4815162342 建议的这些 url

https://medium.freecodecamp.org/a-guide-to-asynchronous-programming-in-python-with-asyncio-232e2afa44f6

何时使用，何时不使用 Python 3.5 `await`？

运行异步函数时，任何输入输出操作都需要异步兼容，但在内存中运行的函数除外。（在我的例子中是一个正则表达式查询）

即任何需要从另一个不兼容异步的源（在我的示例中为 django 查询）收集数据的函数必须在执行程序中运行。

我想我现在已经通过在执行程序中运行所有 django DB 调用来解决我的问题，从那以后我就没有遇到过运行临时脚本的问题。

但是我有 celery 和 async 的兼容性问题（因为 celery 还不兼容 asyncio 这会引发一些错误，但不是我之前看到的错误）

推荐阅读