작업자가 새 데이터를 가져올 수 없습니다. 플라스크 셀러리의 작업에서 mysql에 삽입하십시오.

Question

3 개의 서버에서 내 프로그램을 실행하고 각 서버에는 8 개의 셀러리 작업자가 있습니다. 즉, 각 서버의 셀러리 작업을 다른 서버에서 실행할 수 있습니다. 각 서버

: 변경 및 호출 태스크

... 
    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에 커밋하고 add_ci_bj 작업을 호출하지만 second_id으로 데이터를 가져올 수 없다는 것을 알 수 있습니다. 누구나 도움을 줄 수 있습니까?

Answer 1

다른 서버에서 호출하기 전에 세션을 닫습니다. 예 :

@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) 
    ...