与不使用 celery 相比，Celery 任务的运行过程需要更长的时间答案

【问题标题】：Celery tasks taking longer to run process than if using without celery与不使用 celery 相比，Celery 任务的运行过程需要更长的时间
【发布时间】：2022-06-10 16:56:52
【问题描述】：

我正在使用 celery 处理一个函数，由于使用 networkx 图查找所有路径（特别是 all_simple_paths 方法），该函数有时可能需要很长时间才能处理。

在使用 celery 之前，有一些模拟需要 1-30 秒，所以这对 nginx 不会超时来说很好，但是还有一些模拟可能在 30 秒到 2m 之间。

我选择使用 celery 并轮询状态直到完成，这一切都很好，但是现在最初需要 1-30 秒的模拟现在可以占用 2m。我知道代理/后端会增加一些开销，但肯定不会这么多？有没有办法让这些模拟以正常性能运行？

我使用 dash 构建了轮询器，我使用了两个不同的版本，因为我最初使用的是 dash v1。

dash v1 ex

@app.callback(
    Output("analysis-button", "disabled"),
    Output("analysis-button", "children"),
    Output("cancel-button", "disabled"),
    Output("future-content", "data"),
    Output("result-tuple", "data"),
    Output("poller", "max_intervals"),
    Input("analysis-button", "n_clicks"),
    State("param", "value"),
    State("param_one", "value"),
    State("param_two", "value"),
    prevent_initial_call=True,
)
def start_celery_process(
    n_clicks: bool,
    param: str,
    param_one: str,
    param_two: str
):
    logger.info("Button pressed")
    content_for_polling = short_paths(param, param_one, param_two)
    cel_status = find_all_paths.delay(param, param_one, param_two) # this can take a long time or short time
    return True, "Analyzing...", False, content_for_polling, cel_status.as_tuple(), -1

@app.callback(
    [
        Output("analysis-button", "disabled"),
        Output("analysis-button", "children"),
        Output("cancel-button", "disabled"),
        Output("ret-one", "data"),
        Output("ret-two", "data"),
        Output("ret-three", "data"),
        Output("poller", "max_intervals"),
    ],
    [Input("poller", "n_intervals"), Input("future-content", "data")],
    [State("result-tuple", "data")],
    prevent_initial_call=True,
)
def poll_result(n_intervals, future_content, data):
    logger.info("in poll")
    result = result_from_tuple(data, app=celery_app)
    logger.info(result)
    if not result.ready():
        logger.info("prevented updated poll")
        raise PreventUpdate()
    logger.info("gets into update")
    para_fut, para_fut_one, para_fut_two = future_content
    param_res, param_res_one, param_res_two = result.get()
    ret_param = para_fut.append(para_fut)
    ret_param_one = para_fut_two.append(para_fut_one)
    ret_param_two = para_fut_two.append(para_fut_two)
    
    return (
        False,
        "Analyze",
        True,
        ret_param,
        ret_param_one,
        ret_param_two,
        0
    )

然后我看到 v2 中的破折号有一个 long callback 将 celery 集成到其中，所以我更新了我的版本并尝试实现它，但仍然有同样的问题。

celery_app = Celery(__name__, broker="redis://localhost:6379/0"),
                    backend="redis://localhost:6379/1"))
long_callback_manager = CeleryLongCallbackManager(celery_app)


@app.long_callback(
    Output("analysis-button", "disabled"),
    Output("analysis-button", "children"),
    Output("cancel-button", "disabled"),
    Output("ret-one", "data"),
    Output("ret-two", "data"),
    Output("ret-three", "data"),
    Input("analysis-button", "n_clicks"),
    State("param", "value"),
    State("param_one", "value"),
    State("param_two", "value"),
    prevent_initial_call=True,
    manager=long_callback_manager
)
def start_celery_process(
    n_clicks: bool,
    param: str,
    param_one: str,
    param_two: str
):
    logger.info("Button pressed")
    para_fut, para_fut_one, para_fut_two = short_paths(param, param_one, param_two)
    param_res, param_res_one, param_res_two = find_all_paths(param, param_one, param_two) # this can take a long time or short time
    ret_param = para_fut.append(para_fut)
    ret_param_one = para_fut_two.append(para_fut_one)
    ret_param_two = para_fut_two.append(para_fut_two)
    
    return (
        False,
        "Analyze",
        True,
        ret_param,
        ret_param_one,
        ret_param_two,
    )

就在写这篇文章的时候，我现在的想法是，由于 pickle 序列化程序需要很长时间，因为我有这样的配置：

celery_app.conf.task_serializer = 'pickle'
celery_app.conf.result_serializer = 'pickle'
celery_app.conf.accept_content = ['application/json', 'application/x-python-serialize']

但我删除了它，它没有做任何事情。

find_all_paths 方法如下所示：

def find_all_paths(
        source_ID,
        target_ID,
        min_path,
):
    logger.info("starts find all paths")
    source = index_a[source_ID]
    target = index_a[target_ID]
    start = time.time()
    for path in nx.all_simple_paths(G, source, target, 60): # where G is the network x graph
        paths, paths_weights, weights_map = transform_path(source, target, inv_index_a) 

        elapsed = time.time() - start
        if elapsed > limit:
            break

    return (
        paths,
        paths_weights,
        weights_map,

    )

更新：

我运行了sys.getsizeof 来获取每个对象的大小，以字节为单位返回，如下所示：

sys.getsizeof(my_obj))

三个返回对象的值分别是：

336
336
360

这是使用带有要返回的数据的版本 2 long_callback 完成的。这个特殊的模拟在没有 celery 的情况下需要 30 秒，并且在 nginx 服务器上运行良好。使用长回调（celery 实现）只需 1 分钟多一点。

【问题讨论】：

您的数据有多大？除非真的很大，否则 celery 增加的开销应该可以忽略不计
@emher 刚刚添加了尺寸。

标签： python celery plotly-dash