python+redis防止请求重复提交、提交间隔时间为3s_python limiter 单个ip每分钟只能请求一次

   在很多场景下，会出现用户不小心点击过快的情况下，请求出现重复提交的问题，为了保证系统的数据一致性，可以对在一瞬间发起多次的请求进行拦截，并提示操作频繁，请稍后重试等。
   实现原理
  1) 对于使用session的系统，我们可以将用户信息存储到session里，并指定的name为student， key 的指定方式: user_id: view_obj._ class_:func._ name_。 即: 用户id: 类名: 方法名 。
   2) 使用redis自带的 INCR(key)方法，可以将相同的key值+1 。 如果在方法里>1 ,那么就直接return 提示，如果=1,那么就继续执行原有方法: func(obj_view,request,*args,**kwargs)。

1.添加redis配置

# redis相关配置
REDIS_HOST = "localhost"
REDIS_PORT = "6379"
REDIS_MAX_CONNECTIONS = 50
REDIS_USERNAME = "root"
REDIS_PASSWORD = "123456"
1
2
3
4
5
6

2.自定义一个redis连接池

class ConnectionPoolUtils:
    # 初始化一个连接池,全局唯一
    pool = redis.ConnectionPool(host=settings.REDIS_HOST,
                                port=settings.REDIS_PORT,
                                max_connections=settings.REDIS_MAX_CONNECTIONS,

                                username=settings.REDIS_USERNAME,
                                password=settings.REDIS_PASSWORD)


# 与redis建立连接
def get_redis_connection():
    pool = ConnectionPoolUtils.pool
    conn = redis.Redis(connection_pool=pool, socket_connect_timeout=3000)
    return conn

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

3. 自定义装饰器，用来标注方法

  可以使用redis自带的incr()方法，如果有两个线程同时进入到lock_request方法，同时执行到redis_conn.incr(key)方法，两个线程会排队进入到redis, 如果count=2,那么就return。

from functools import wraps

def lock_request(func):
    @wraps(func)
    def wrapper(view_obj, request, *args, **kwargs):
        if "student" in request.session:
            student = request.session["student"]
        else:
            raise Exception("系统出错，请联系系统管理员!")
        user_id = student["id"]
        # user_id = 1
        key = '{}:{}:{}'.format(user_id, view_obj.__class__.__name__, func.__name__)
        redis_conn = get_redis_connection()
        # 如果有多个相同的请求同时进来，那么count>1。
        count = redis_conn.incr(key)
        redis_conn.expire(key, 60)
        if count > 1:
            redis_conn.close()
            return JsonResponse({'msg': '操作频繁，请稍后重试!', 'code': -2})
        else:
            res = func(view_obj, request, *args, **kwargs)
            redis_conn.delete(key)
            redis_conn.close()
            return res

    return wrapper
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

测试方法:

class RepeatView(BaseView):

    @lock_request
    def get(self, request, format=None):
        return HttpResponse("请求成功!")

    @lock_request
    def post(self, request, format=None):
        return HttpResponse("请求成功!")

1
2
3
4
5
6
7
8
9
10

   优化: 上述方法能解决在高并发短时间内同用户下的重复请求，如果需要在3s内不能重复请求，应该怎么做?

# 强制3s内不能重复提交请求
def lock_submit(func):
    @wraps(func)
    def wrapper(view_obj, request, *args, **kwargs):
        user_token = request.META.get("HTTP_USER_TOKEN")
        print("token:", user_token)
        if user_token is None:
            raise AuthException
        redis_conn = get_redis_connection()
        if not redis_conn.exists(user_token):
            redis_conn.close()
            raise AuthException
        user_id = redis_conn.get(user_token).decode("utf-8")
        key = '{}:{}:{}'.format(user_id, view_obj.__class__.__name__, func.__name__)
        # 如果有多个相同的请求同时进来，那么count>1, 如果有多个请求会阻塞,同时强制3s内不能重复
        click = redis_conn.get(key)
        ttl_key = redis_conn.execute_command("ttl", key)
        if ttl_key == -1 or ttl_key == -2:
            # 已失效的key或者不存在的key
            redis_conn.set(key, 1)
            redis_conn.expire(key, 5)
            redis_conn.close()
        else:
            click = click.decode("utf-8")
            if int(click) > 1:
                redis_conn.close()
                return JsonResponse({'message': '操作频繁，请稍后重试!', 'code': -9})
            redis_conn.incr(key)
            redis_conn.close()
        res = func(view_obj, request, *args, **kwargs)
        return res

    return wrapper
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

注: 用redis_conn.get(key)获取key时,会获取到失效的key,因此需要使用ttl key判断一下key有没有失效或key不存在， -1 时表示key已失效，-2表示Key不存在或者失效的key已被redis清除。

4. Jmeter和Postman测试

测试第一种情况

使用jmeter测试100并发，结果主要有两种情况:
1 ) 请求成功的

2. 重复请求
   可以发现有部分请求，出现操作频繁，稍后重试的返回错误。
   

测试第二种情况

  第一次访问后，再次访问， 查看在3s内、3s后的访问情况。