在我目前正在進行的項目中，我不允許使用 ORM，所以我自己做了它工作得很好，但我在 Celery 和它的并發性方面遇到了問題。有一段時間，我將其設置為1(using --concurrency=1)，但我添加了新任務，這些任務的處理時間比使用 celerybeat 運行所需的時間要長，這會導致任務大量積壓。當我將 celery 的并發設置為 > 1 時，會發生以下情況（pastebin 因為它很大）：https://pastebin.com/M4HZXTDC關于如何在其他進程上實現某種鎖定/等待以使不同的工作人員不會相互交叉的任何想法？編輯：這是我設置PyMySQL 實例以及如何處理打開和關閉的位置
1 回答
鴻蒙傳說
TA貢獻1865條經驗 獲得超7個贊
PyMSQL不允許線程共享同一個連接（模塊可以共享，但線程不能共享連接）。您的模型類在各處重用相同的連接。
因此，當不同的工作人員調用模型進行查詢時，他們使用相同的連接對象，從而導致沖突。
確保您的連接對象是線程本地的。不要使用db類屬性，而是考慮一種檢索線程本地連接對象的方法，而不是重用可能在不同線程中創建的連接對象。
例如，在任務中創建連接。
現在，您在每個模型的任何地方都使用全局連接。
# Connect to the database
connection = pymysql.connect(**database_config)
class Model(object):
? ? """
? ? Base Model class, all other Models will inherit from this
? ? """
? ? db = connection
為了避免這種情況，您可以在方法中創建數據庫__init__......
class Model(object):
? ? """
? ? Base Model class, all other Models will inherit from this
? ? """
? ? def __init__(self, *args, **kwargs):
? ? ? ? self.db = pymysql.connect(**database_config)
但是，這可能不高效/不實用，因為 db 對象的每個實例都會創建一個會話。
為了改進這一點，您可以使用一種方法threading.local來將連接保持在線程本地。
class Model(object):
? ? """
? ? Base Model class, all other Models will inherit from this
? ? """
? ? _conn = threading.local()
? ? @property
? ? def db(self):
? ? ? ? if not hasattr(self._conn, 'db'):
? ? ? ? ? ? self._conn.db = pymysql.connect(**database_config)
? ? ? ? return self._conn.db
請注意，假設您使用線程并發模型，線程本地解決方案就可以工作。另請注意，celery 默認情況下使用多個進程（prefork）。這可能是問題，也可能不是問題。
添加回答
舉報
0/150
提交
取消