多线程更新同一条父记录报错：could not serialize access due to concurrent update

在Odoo中，我需要在stock.move完成后执行自定义操作，由于拣货验证存在性能问题，因此采用多线程在后台运行该操作，但更新同一行数据时遇到如下报错：

ERROR: could not serialize access due to concurrent update once trying to update data to the same records of parent in threaded function

代码实现如下：

def _action_done(self, cancel_backorder=False):
    res = super(StockMove, self)._action_done(cancel_backorder=cancel_backorder)
    if res:
        done_moves = res.filtered(lambda x: x.state == 'done')
        if done_moves:
            threaded_calculation = threading.Thread(
                            target=self.sync_fulfillment_stats,
                            args=(done_moves.ids,))
            threaded_calculation.setDaemon(True)
            threaded_calculation.start()

def sync_fulfillment_stats(self, move_ids):
    # 休眠确保已保存完成数量
    time2.sleep(3)
    with api.Environment.manage(), self.pool.cursor() as new_cr:
        self = self.with_env(self.env(cr=new_cr))
        moves = self.env['stock.move'].search([('id', 'in', move_ids)])
        for move in moves:
            move.sync = True   # 其他自定义操作

1. 替换不可靠的休眠，改用事务安全的任务触发 #

原代码里的3秒休眠完全依赖运气，无法保证主线程的事务已经提交。如果可以安装Odoo官方的queue_job模块，直接用它的延迟任务机制是最优解——它会自动等当前事务提交后再执行后台任务，从根源避免并发冲突：

# 先安装queue_job模块
from odoo.addons.queue_job.job import job

@job
def sync_fulfillment_stats(self, move_ids):
    with api.Environment.manage(), self.pool.cursor() as new_cr:
        self = self.with_env(self.env(cr=new_cr))
        moves = self.env['stock.move'].search([('id', 'in', move_ids)])
        for move in moves:
            move.sync = True   # 执行你的自定义操作

def _action_done(self, cancel_backorder=False):
    res = super(StockMove, self)._action_done(cancel_backorder=cancel_backorder)
    done_moves = res.filtered(lambda x: x.state == 'done')
    if done_moves:
        # 用delay()替代多线程，自动处理事务提交时机
        self.sync_fulfillment_stats.delay(done_moves.ids)

2. 如果不能用queue_job，添加锁冲突处理逻辑 #

如果必须用多线程，要给查询加锁并处理冲突：

• 在搜索记录时加上for_update=True和skip_locked=True（PostgreSQL支持），跳过已被主线程锁定的记录，稍后再处理
• 给更新操作加重试机制，遇到锁冲突时自动重试几次
• 用批量写入替代循环单条更新，提升性能同时减少锁冲突概率

修改后的sync_fulfillment_stats：

import time
from logging import getLogger

_logger = getLogger(__name__)

def sync_fulfillment_stats(self, move_ids):
    max_retries = 3
    retry_interval = 2
    with api.Environment.manage(), self.pool.cursor() as new_cr:
        self = self.with_env(self.env(cr=new_cr))
        for attempt in range(max_retries):
            # 跳过已被锁定的未同步记录
            moves = self.env['stock.move'].search(
                [('id', 'in', move_ids), ('sync', '=', False)],
                for_update=True,
                skip_locked=True
            )
            if not moves:
                break
            try:
                moves.write({'sync': True})
                break
            except Exception as e:
                _logger.warning("第%d次更新失败，错误: %s", attempt+1, str(e))
                time.sleep(retry_interval)
        else:
            _logger.error("多次尝试后仍无法更新move记录: %s", str(move_ids))

3. 避免在主事务未提交时启动线程 #

原代码中_action_done运行在主事务内，此时启动线程会导致后台线程读取未提交的数据，或和主事务提交时产生冲突。如果不用queue_job，手动提交事务风险较高，优先推荐queue_job方案。

内容的提问来源于stack exchange，提问作者Phong Vy