Tortoise ORM中捕获1062唯一键重复错误并重试，生成连续唯一产品编码

一、修复异常捕获错误 #

你当前的异常捕获语法完全错误，正确的做法是先捕获IntegrityError，再在异常块内检查MySQL的1062错误码（唯一键冲突）。以下是修正后的代码示例（以Tortoise-ORM为例，SQLAlchemy逻辑类似）：

from tortoise.exceptions import IntegrityError

product = Product(code="")

while True:
    # 用数据库原子更新代替先读再改，避免竞态条件
    updated_rows = await GlobalCount.filter(type='product_code').update(count=GlobalCount.count + 1)
    
    # 处理计数器未初始化的情况
    if updated_rows == 0:
        await GlobalCount.create(type='product_code', count=1)
        current_count = 1
    else:
        g_count = await GlobalCount.get(type='product_code')
        current_count = g_count.count
    
    product.code = f'P{current_count}'

    try:
        await product.save()
        break  # 成功生成唯一编码，退出循环
    except IntegrityError as e:
        # 检查是否是MySQL唯一键冲突错误（错误码1062）
        if hasattr(e, 'orig') and e.orig.errno == 1062:
            continue  # 重复编码，重试
        raise  # 其他完整性错误，直接抛出
    except Exception:
        raise  # 其他异常直接抛出

二、多Worker环境下的规范实践 #

1. 数据库原子化全局计数器 #

之前的get+save方式存在竞态：多个Worker可能同时读取到相同的count值，导致生成重复编码。必须使用数据库原子更新，直接在数据库层面完成自增操作，避免应用层的竞态：

-- 对应的SQL语句，ORM会自动生成
UPDATE global_count SET count = count + 1 WHERE type = 'product_code';

这种操作由数据库原子执行，不会出现并发冲突。

2. Redis分布式原子自增 #

如果是分布式多实例部署，用Redis的INCR命令实现原子自增性能更高，天生支持分布式场景：

import redis.asyncio as redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

async def generate_product_code():
    # Redis INCR是原子操作，保证每次调用返回唯一递增数值
    current_count = await redis_client.incr('product_code_counter')
    return f'P{current_count}'

# 生成并保存编码
try:
    product.code = await generate_product_code()
    await product.save()
except IntegrityError as e:
    # 极端情况Redis与数据库不一致，回减计数器并重试
    await redis_client.decr('product_code_counter')
    product.code = await generate_product_code()
    await product.save()

3. 数据库自增ID+前缀（非严格连续） #

如果可以接受编码因删除操作出现断号，直接用数据库自增主键作为编码的数字部分是最简单的方案：

class Product(Model):
    id = fields.IntField(pk=True, generated=True)  # 数据库自增主键
    code = fields.CharField(max_length=20, unique=True)

# 保存后自动生成编码
product = await Product.create()
product.code = f'P{product.id}'
await product.save()

完全依赖数据库主键的唯一性，无需额外计数器，实现成本极低。

4. 乐观锁优化全局计数器 #

给GlobalCount添加版本字段，用乐观锁确保只有第一个修改的Worker能成功，其他Worker自动重试：

class GlobalCount(Model):
    type = fields.CharField(unique=True)
    count = fields.IntField()
    version = fields.IntField(default=0)  # 乐观锁版本号

async def get_next_count():
    while True:
        g_count = await GlobalCount.get(type='product_code')
        new_count = g_count.count + 1
        # 只有版本号匹配时才更新，避免并发覆盖
        updated_rows = await GlobalCount.filter(
            type='product_code',
            version=g_count.version
        ).update(count=new_count, version=g_count.version + 1)
        if updated_rows == 1:
            return new_count
        # 更新失败，说明其他Worker已修改，重试

内容的提问来源于stack exchange，提问作者viva-glenn