0

我有三台服务器运行我的程序，每台服务器有八个芹菜工人从 redis 获取任务。也就是说每个服务器的 celery 任务都可以被另一个服务器执行。

在每个服务器中： 提交更改并调用任务

    ...
    try:
        db.session.commit()
    except Exception as e:
        current_app.logger.error(str(e))
        db.session.rollback()
        if not ci_existed:  # only add
            self.delete(ci.ci_id)
        return abort(500, "add CI error")
    his_manager = CIAttributeHistoryManger()
    his_manager.add(ci.ci_id, histories)
    ci_cache.apply_async([ci.ci_id], queue="async")
    # add bj ci
    add_ci_bj.apply_async([ci_type.type_name, None, ci.ci_id], queue="async")
    return ci.ci_id

任务功能

@celery.task(name="xxxxxxx", queue="async")
def add_ci_bj(ci_type, first_id, second_id):
    param, status = lib.ci.CIManager().get_relations(first_id, second_id, is_async=True)
    ...

任务中的功能

def get_relations(self, first_id, second_id, is_async=False):
    start = time.clock()
    try:
        second = self.get_ci_by_id(second_id, need_children=False)
    except Exception as e:
    return None, "get ci by id error: first %s, second %s, e %s, is_async:%s" % \
           (first_id, second_id, e, is_async)
    ...

我插入数据并提交到 MySQL 并调用 task add_ci_bj，但我无法获取数据second_id，我无法弄清楚，任何人都可以提供帮助吗？

4

1 回答 1

0

得到答案，在调用另一台服务器之前关闭会话。如：

@celery.task(name="xxxxxxx", queue="async")
def add_ci_bj(ci_type, first_id, second_id):
    db.session.Close()
    param, status = lib.ci.CIManager().get_relations(first_id, second_id, is_async=True)
    ...
于 2016-11-15T13:40:01.783 回答