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

У меня есть распределенная вычислительная среда, в которой используется Celery + RABBITMQ + супервизор. В задачи моего работника входит чтение из базы данных, вычисление некоторых значений и обновление базы данных после завершения процесса. Однако, когда я пытаюсь запустить несколько рабочих распределенным образом, я продолжаю сталкиваться с ошибкой: -

(2014 г., «Команды не синхронизированы; сейчас вы не можете запустить эту команду»)

Может ли кто-нибудь предложить мне способ настроить механизм, подобный мьютексу или файлу блокировки, чтобы рабочие могли одновременно обращаться к базе данных.

Любая помощь будет оценена, спасибо, Амит

Редактировать :-

con = mdb.connect(parameters...)

def reset_table(table_name,con):
    with con:
        cur = con.cursor(mdb.cursors.DictCursor)
        cur.execute("UPDATE " + table_name + " SET active_status = 0 where last_access <     (NOW() - INTERVAL 15 MINUTE)")
        con.commit()

Трассировки стека :-

   File "/usr/local/lib/python2.7/dist-packages/celery/app/trace.py", line 238, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/celery/app/trace.py", line 416, in    __protected_call__
    return self.run(*args, **kwargs)
  File "/home/elasticsearch/celery_test/tasks.py", line 183, in download_data
    auth = get_auth(con)
  File "/home/elasticsearch/celery_test/tasks.py", line 94, in get_auth
    reset_table("auths",con)
  File "/usr/lib/python2.7/dist-packages/MySQLdb/connections.py", line 249, in __exit__
    self.rollback()
ProgrammingError: (2014, "Commands out of sync; you can't run this command now")

mysql celery locking mysql-python django-celery
person Amit Gupta    schedule 07.01.2014    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Я бы не стал спешить с применением механизма блокировки, а скорее попытался бы исправить то, как ваш клиент использует базу данных, см. документы mysql для несинхронизированных команд

Если вы получаете команды не синхронизированы; вы не можете запустить эту команду сейчас в своем клиентском коде, вы вызываете клиентские функции в неправильном порядке.

если вы решите реализовать блокировку (что опять же я бы не рекомендовал), один из хороших способов сделать это выглядит следующим образом:

import redis

have_lock = False
my_lock = redis.Redis().lock("my_key")
try:
    have_lock = my_lock.acquire(blocking=True)
    if have_lock:
        print("Got lock.")
    else:
        print("Did not acquire lock.")

finally:
    if have_lock:
        my_lock.release()

подробное объяснение см. на http://loose-bits.com/2010/10/distributed-task-locking-in-celery.html

person Guy Gavriely    schedule 07.01.2014
comment
MySQL-python реализует функции ожидания по умолчанию. Извините, но я не понимаю, как я могу заставить его работать без функции ожидания где-то в коде. - person Amit Gupta; 08.01.2014
comment
трудно сказать, не видя ваших запросов/SQL и т. д., лучше включить всю трассировку стека, а не только саму ошибку - person Guy Gavriely; 08.01.2014
comment
Понятно, а вы знаете, что на самом деле делает этот оператор with? см. stackoverflow.com/questions/8067690/ - person Guy Gavriely; 08.01.2014
comment
Извините, я до сих пор не понимаю ответа. Похоже, мне на самом деле не нужно реализовывать механизм lockfile. Мне просто нужно убедиться, что вызовы db выполняются правильно, и вызывается mysql_free_result. Однако я не уверен, как это обеспечить. - person Amit Gupta; 08.01.2014
comment
Я предполагаю, что with и con.commit () плохо сочетаются друг с другом, удалите con.commit () и посмотрите, как пойдет. - person Guy Gavriely; 08.01.2014