create fast set type

.travis.yml

@@ -5,6 +5,9 @@ python:
   - "3.4"
   - "3.6"
   - "3.7"
+  - "3.8"
+  - "3.9"
+  - "3.10"
 services: redis-server
 install: pip install .
 script: python setup.py test

hot_redis/fast_set.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+
+import random
+import time
+
+from redis import Redis
+
+
+class DelayButFastSet:
+    """
+    this class will read data from redis periodly and keep data in memory
+    usage:
+
+        WATCHING_USERS = DelayBuyFastSet(Redis(), key="WATCHING_USERS", timeout=5)
+        "123" in WATCHING_USERS  # False
+        WATCHING_USERS.add("123")
+        "123" in WATCHING_USERS  # True
+
+        WATCHING_USERS = DelayBuyFastSet(Redis(), key="WATCHING_USERS", timeout=5)
+        "123" in WATCHING_USERS  # True
+    """
+
+    def __init__(self, redis_client, key, timeout):
+        assert redis_client.get_encoder().decode_responses is True
+
+        self.redis_client = redis_client
+        self.value_key = f"{key}:value"
+        self.version_key = f"{key}:version"
+
+        self.timeout = timeout
+        self.expire_at = time.perf_counter() + random.random() * timeout
+
+        self._value = set()
+        self.version = 0
+        self.refresh()
+
+    def __contains__(self, value):
+        self.refresh_in_need()
+        return str(value) in self._value
+
+    def refresh_in_need(self):
+        if time.perf_counter() < self.expire_at:
+            return
+        self.expire_at = time.perf_counter() + self.timeout
+        if int(self.redis_client.get(self.version_key) or 0) == self.version:
+            return
+        self.refresh()
+
+    def refresh(self):
+        self._value = self.redis_client.smembers(self.value_key)
+        self.version = self.redis_client.incr(self.version_key)
+
+    def add(self, value: str):
+        assert isinstance(value, str)
+        self._value.add(value)
+        added, version = self.redis_client.pipeline()\
+                .sadd(self.value_key, value)\
+                .incr(self.version_key)\
+                .execute()
+
+    def discard(self, value: str):
+        assert isinstance(value, str)
+        self._value.discard(value)
+        self.redis_client.pipeline()\
+                .srem(self.value_key, value)\
+                .incr(self.version_key)\
+                .execute()

hot_redis/types.py

@@ -916,4 +916,4 @@ def most_common(self, n=None):
             values = values[:n]
         return values
 
-collections.MutableMapping.register(MultiSet)
+collections.abc.MutableMapping.register(MultiSet)

tests/test_types.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+# Xiang Wang <ramwin@qq.com>
+
+
+import time
+import unittest
+import logging
+
+from hot_redis.fast_set import DelayButFastSet
+from redis import Redis
+
+from multiprocessing import Pool
+
+
+logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler()])
+LOGGER = logging.getLogger(__name__)
+
+
+class Test(unittest.TestCase):
+
+    @staticmethod
+    def inner_test_performance(taskid: int):
+        size = 10000
+        Redis().delete("set_fastkey:value")
+        Redis().delete("set_fastkey:version")
+        Redis().delete("set_direct")
+        a = DelayButFastSet(redis_client=Redis(decode_responses=True), key="set_fastkey", timeout=5)
+        for i in range(0, size, 2):
+            a.add(str(i))
+
+        start = time.perf_counter()
+        b = DelayButFastSet(redis_client=Redis(decode_responses=True), key="set_fastkey", timeout=5)
+        cnt = 0
+        for j in range(0, size, 3):
+            if str(j) in b:
+                cnt += 1
+        end = time.perf_counter()
+        LOGGER.info("task %d use DelayButFastSet: %f, result: %d", taskid, end - start, cnt)
+
+        client = Redis(decode_responses=True)
+        for i in range(0, size, 2):
+            client.sadd("set_direct", str(i))
+
+        start = time.perf_counter()
+        cnt = 0
+        for j in range(0, size, 3):
+            if client.sismember("set_direct", str(j)):
+                cnt += 1
+        end = time.perf_counter()
+        LOGGER.info("task %d user redis directly: %f, result: %d", taskid, end - start, cnt)
+
+    def test_performance(self):
+        task_cnt = 4
+        with Pool(task_cnt) as p:
+            p.map(self.inner_test_performance, range(task_cnt))