优化锁的问题

Boris-code · Boris-code · commit 15ed7326fbc4 · 2021-07-15T18:22:20.000+08:00
diff --git a/feapder/core/scheduler.py b/feapder/core/scheduler.py
@@ -263,10 +263,7 @@ def _start(self):
             if self.wait_lock:
                 # 将添加任务处加锁，防止多进程之间添加重复的任务
                 with RedisLock(
-                    key=self._spider_name,
-                    timeout=3600,
-                    wait_timeout=60,
-                    redis_cli=RedisDB().get_redis_obj(),
+                    key=self._spider_name, redis_cli=RedisDB().get_redis_obj()
                 ) as lock:
                     if lock.locked:
                         self.__add_task()
diff --git a/feapder/core/spiders/batch_spider.py b/feapder/core/spiders/batch_spider.py
@@ -946,10 +946,7 @@ def task_is_done(self):
         if is_done:  # 检查任务表中是否有没做的任务 若有则is_done 为 False
             # 比较耗时 加锁防止多进程同时查询
             with RedisLock(
-                key=self._spider_name,
-                timeout=3600,
-                wait_timeout=0,
-                redis_cli=RedisDB().get_redis_obj(),
+                key=self._spider_name, redis_cli=RedisDB().get_redis_obj()
             ) as lock:
                 if lock.locked:
                     log.info("批次表标记已完成，正在检查任务表是否有未完成的任务")
diff --git a/feapder/dedup/bloomfilter.py b/feapder/dedup/bloomfilter.py
@@ -205,6 +205,8 @@ class ScalableBloomFilter(object):
     BASE_MEMORY = BloomFilter.BASE_MEMORY
     BASE_REDIS = BloomFilter.BASE_REDIS
 
+    __redis_cli = None
+
     def __init__(
         self,
         initial_capacity: int = 100000000,
@@ -237,6 +239,13 @@ def _setup(self, initial_capacity, error_rate, name, bitarray_type, redis_url):
     def __repr__(self):
         return "<ScalableBloomFilter: {}>".format(self.filters[-1].bitarray)
 
+    @property
+    def _redis_cli(self):
+        if self.__class__.__redis_cli is None:
+            self.__class__.__redis_cli = RedisDB(url=self.redis_url).get_redis_obj()
+
+        return self.__class__.__redis_cli
+
     def create_filter(self):
         filter = BloomFilter(
             capacity=self.initial_capacity,
@@ -267,12 +276,13 @@ def check_filter_capacity(self):
 
                     self._check_capacity_time = time.time()
             else:
-                with RedisLock(
-                    key="ScalableBloomFilter",
-                    timeout=300,
-                    wait_timeout=300,
-                    redis_cli=RedisDB(url=self.redis_url).get_redis_obj(),
-                ) as lock:  # 全局锁 同一时间只有一个进程在真正的创建新的filter，等这个进程创建完，其他进程只是把刚创建的filter append进来
+                # 全局锁 同一时间只有一个进程在真正的创建新的filter，等这个进程创建完，其他进程只是把刚创建的filter append进来
+                key = (
+                    f"ScalableBloomFilter:{self.name}"
+                    if self.name
+                    else "ScalableBloomFilter"
+                )
+                with RedisLock(key=key, redis_cli=self._redis_cli) as lock:
                     if lock.locked:
                         while True:
                             if self.filters[-1].is_at_capacity:
diff --git a/feapder/utils/redis_lock.py b/feapder/utils/redis_lock.py
@@ -14,13 +14,13 @@
 
 
 class RedisLock(object):
-    def __init__(self, key, redis_cli, wait_timeout=0, lock_timeout=0):
+    def __init__(self, key, redis_cli, wait_timeout=0, lock_timeout=86400):
         """
         redis超时锁
         :param key: 存储锁的key redis_lock:[key]
         :param redis_cli: redis客户端对象
         :param wait_timeout: 等待加锁超时时间，为0时则不等待加锁，加锁失败
-        :param lock_timeout: 锁超时时间 为0时则不会超时，直到锁释放或意外退出
+        :param lock_timeout: 锁超时时间 为0时则不会超时，直到锁释放或意外退出，默认超时为1天
 
         用法示例:
         with RedisLock(key="test", redis_cli=redis_obj) as _lock:
diff --git a/tests/test_lock.py b/tests/test_lock.py
@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+"""
+Created on 2021/7/15 5:00 下午
+---------
+@summary:
+---------
+@author: Boris
+@email: boris_liu@foxmail.com
+"""
+
+from feapder.utils.redis_lock import RedisLock
+from feapder.db.redisdb import RedisDB
+import time
+
+def test_lock():
+    with RedisLock(key="test", redis_cli=RedisDB().get_redis_obj(), wait_timeout=10) as _lock:
+        if _lock.locked:
+            print(1)
+            time.sleep(100)
+
+if __name__ == '__main__':
+    test_lock()
