automatic_crawler/application/common/redis/redis_helper.py

import redis

class SyncRedisHelper(object):
    _pool: redis.ConnectionPool = None
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance

    def _get_pool(self) -> redis.ConnectionPool:
        if self._pool is None:
            self._pool = redis.ConnectionPool(
                host='r-bp1mb0v08fqi4hjffu.redis.rds.aliyuncs.com',
                # host='r-bp1mb0v08fqi4hjffupd.redis.rds.aliyuncs.com',  # 外网地址
                port=6379,
                db=2,
                password='Wqsd@2019'
            )
        return self._pool

    def get_client(self) -> redis.Redis:
        pool = self._get_pool()
        client = redis.Redis(connection_pool=pool)
        return client

    def close(self):
        if self._pool:
            self._pool.disconnect(inuse_connections=True)