Совместное использование соединения базы данных Oracle между одновременными задачами Celery

Я работаю с Python2.7, Celery и cx_Oracle для доступа к базе данных Oracle.

Я создаю много задач. Каждая задача запускает запрос через cx_Oracle. Многие из этих задач будут выполняться одновременно. Все задачи должны иметь одно и то же соединение с базой данных.

Если я запускаю только одну задачу, запрос запускается корректно. Однако, если я запускаю несколько запросов, я начинаю получать это сообщение об ошибке:

[2016-04-04 17:12:43,846: ERROR/MainProcess] Task tasks.run_query[574a6e7f-f58e-4b74-bc84-af4555af97d6] raised unexpected: DatabaseError('<cx_Oracle._Error object at 0x7f9976635580>',) Traceback (most recent call last): File "/usr/local/lib/python2.7/dist-packages/celery/app/trace.py", line 240, in trace_task R = retval = fun(*args, **kwargs) File "/usr/local/lib/python2.7/dist-packages/celery/app/trace.py", line 438, in __protected_call__ return self.run(*args, **kwargs) File "/home/ric/workspace/dbw_celery/tasks.py", line 39, in run_query column_names = get_column_names(oracle_conn, table_info["table_name"]) File "/home/ric/workspace/dbw_celery/utilities.py", line 87, in get_column_names cursor.execute(query_str) DatabaseError: <cx_Oracle._Error object at 0x7f9976635580> 

Давайте посмотрим на мой код.

Это мой файл tasks.py , где я создаю соединение с базой данных Oracle, экземпляр Celery и определяю мои задачи, которые будут tasks.py пользовательское соединение с базой данных:

 # tasks.py import celeryconfig from celery import Celery from utilities import connect_to_db, get_new_rows, write_output_rows # Define a Celery instance dbwapp = Celery('tasks') dbwapp.config_from_object(celeryconfig) dbwapp.conf["CELERYBEAT_SCHEDULE"] = {} # Define an Oracle connection as a global variable to be used by all tasks oracle_conn = connect_to_db(db_user, db_pass, db_host, db_port, db_name) # Define the task function that each Celery worker will run @dbwapp.task() def run_query(table_info, output_description): """Run a query on a given table. Writes found rows to output file.""" global oracle_conn column_names = get_column_names(oracle_conn, table_info["table_name"]) new_rows, last_check_timestamp = get_new_rows(oracle_conn, table_info) write_result_to_output_file(output_file, new_rows) def load_celerybeat_schedule(): """Loads the CELERYBEAT_SCHEDULE dictionary with the tasks to run.""" new_task_dict = { "task": "tasks.run_query", "schedule": timedelta(seconds=table_config["check_interval"]), "args": (table_config, output_description) } new_task_name = "task-" + table_config["table_name"] dbwapp.conf["CELERYBEAT_SCHEDULE"][new_task_name] = new_task_dict 

Вот как я подключаюсь к базе данных в файле utilities.py :

 # utilities.py def connect_to_db(db_user, db_password, db_host, db_port, db_name): """Connect to DB.""" connection_str = "%s/%s@%s:%s/%s" % (db_user, db_password, db_host, db_port, db_name) try: db_connection = cx_Oracle.connect(connection_str) except cx_Oracle.DatabaseError: logger.error("Couldn't connect to DB %s" % db_name) return None logging.info("Succesfully connected to the DB: %s" % db_name) return db_connection 

Это get_new_rows_function определенная в другом файле, где запрос фактически запускается:

 #utilities.py def get_new_rows(db_connection, table_info): """Return new rows inserted in a given table since last check.""" cursor = db_connection.cursor() query_str = "SELECT * FROM {0}".format(table_info["table_name"]) cursor.execute(query_str) new_rows = cursor.fetchall() cursor.close() return new_rows 

Я запускаю свой код следующим образом: celery -A tasks worker -B

Я попытался упростить свой код, чтобы упростить его понимание.

Я боюсь, что ошибка, которую я получаю, вызвана различными задачами, выполняемыми одновременно и совместным использованием одного и того же соединения с базой данных. Их одновременное исполнение «перепутано» или что-то в этом роде.

Каков правильный способ совместного использования базы данных между различными задачами Celery?

Кто-нибудь знает, что я делаю неправильно?

• Как назначить рабочие дни в сельском сельском хозяйстве
• Сельдерей умирает с DBPageNotFoundError
• Python Celery socket.error: Соединение отклонено
• Как демонировать периодическую задачу django celery на сервере ubuntu?
• Результаты Django Celery задают идентификатор задачи для чего-то читаемого человеком?

 conn = cx_Oracle.connect(connection_str, threaded = True)