python django设计流水号生成，带访问加锁，重试

python django设计流水号生成，带访问加锁，重试

老规矩，技术发于业务，努力从于项目。

1、设计一个流水号配置表如下：

# 流水号配置表
class BaseSerialNum(BaseModel):
    serial_type = models.CharField(verbose_name="流水号类型", max_length=10, unique=True)
    prefix = models.CharField(verbose_name="流水号的前缀字符", max_length=10, blank=True, null=True)
    serial = models.IntegerField(verbose_name="流水号的序列", default=0)
    serial_digit = models.IntegerField(verbose_name="流水号的位数", default=4)
    # 默认 前缀字符 + 年月日 + 流水号
    format = models.CharField(verbose_name="流水号的格式", max_length=10, blank=True, null=True)

    class Meta:
        verbose_name = '流水号配置表'
        verbose_name_plural = verbose_name
        db_table = "base_serial_num"

2、编写函数如下：

这里django版本太低了，如果稍微高一点，可以在select_for_update 这里多定义一个变量timeout

def get_serial_num(serial_type):
    # 当前时间
    now_time = datetime.now()
    today = now_time.date()
    formatted_date = now_time.strftime('%Y%m%d')

    # 获取流水号配置信息时对数据加锁  使用FOR UPDATE语句加上排它锁，以避免并发问题
    for i in range(3):  # 重试三次
        try:
            with transaction.atomic():
                # SELECT ... FOR UPDATE 语句用于加锁，避免并发问题
                # 如果锁不可用，则等待0.3秒  如果还获取不到则抛出异常
                serial_obj = BaseSerialNum.objects.select_for_update(nowait=True).get(serial_type=serial_type)

                last_update_time = serial_obj.update_time
                serial = serial_obj.serial
                if last_update_time and last_update_time.date() == today:
                    serial += 1
                else:
                    serial = 1
                result = BaseSerialNum.objects.filter(id=serial_obj.id).update(serial=serial, update_time=now_time)
                # 未更新成功重试
                if result == 0:
                    continue

                serial_number = serial_obj.prefix + formatted_date + '{:0{length}d}'.format(serial,
                                                                                            length=serial_obj.
                                                                                            serial_digit)
                return True, serial_number
        # 如果不存在抛出异常
        except ObjectDoesNotExist:
            return False, f"流水号类型参数 {serial_type} 未定义"
        # 获取不到锁异常 等待0.3秒重试
        except OperationalError:
            time.sleep(0.3)
            continue

    # 重试3次后依然失败，返回None
    return False, None

3、编写视图类如下：

# 流水号配置
class BaseSerialNumView(BaseViewSet):
    serializer_class = BaseSerialNumSerializer
    queryset = BaseSerialNum.objects.all().order_by("-id")

    # 生成流水号接口
    @action(methods=["get"], detail=False)
    def get_serial_num(self, request):
        req_data = request.query_params.copy()

        serial_type = req_data.get("serial_type")
        if not serial_type:
            return ReturnData(msg=f"缺少流水号类型参数", code=401)

        result, data = get_serial_num(serial_type)

        if not result:
            return ReturnData(msg=f"获取失败{data}", code=402)

        return ReturnData(msg=f"获取成功", code=200, data={"serial_num": data})

打完收工 芜湖~