Python自定义拦截器

场景:同一段表单需要两个人填写,但是其中一人提交后表单则不能再次提交了,如果后端正在处理第一人提交的表单,此时第二人也提交,很大概率导致重复提交两份,或者第二人覆盖提交第一人的表单内容

解决:使用到了 redis 的 incr 原子操作,对一个键进行自增,如果自增后发现不等于1, 则表示当表单由其他人正在处理

代码实现:

from extensions import REDIS
from typing import Callable


# 拦截器
class Interceptor(object):

    def __init__(self, id: str, over_func: Callable = None, args: tuple = tuple()):
        if REDIS.incr(id) != 1:
            if over_func:
                over_func(*args)
            print('被拦截')
            return
        self.id = id

    def __enter__(self):
        pass

    def __exit__(self, *args, **kwargs):

        if hasattr(self, 'id'):
            # 正常执行完成删除自增键
            REDIS.delete(self.id)

测试: 使用多线程测试

import threading
import time


def ad():
    with Interceptor(id='abc') as a:
        time.sleep(0.1)
        return


task_list = []
for one in range(20):
    t = threading.Thread(target=ad)
    task_list.append(t)

for one in task_list:
    one.start()

for one in task_list:
    one.join()