Python 异步 redis_redis.asyncio

From：https://zhuanlan.zhihu.com/p/24720629

1. aioredis 要求装上 hiredis ， 而 aredis 可以不需要相关依赖地运行，速度上两者持平，且都可以使用 hiredis 来作为 parser ，用 uvloop 代替 asyncio 的 eventloop 来加速
2. asyncio_redis 使用了 Python 提供的 protocol 来进行异步通信，而 aredis 则使用 StreamReader 和 StreamWriter 来进行异步通信，在运行速度上两倍于 asyncio_redis ，附上 benchmark
3. aioredis 和 asyncio_redis 这两个客户端目前都还没有对于集群的支持，相对来说 aredis 的功能更为全面一些。

1、aredis

github 地址：https://github.com/NoneGG/aredis
aredis 官方英文文档：https://aredis.readthedocs.io/en/latest/
aredis 一个高效和用户友好的异步Redis客户端：https://www.ctolib.com/aredis.html
更多使用示例：https://github.com/NoneGG/aredis/tree/master/examples

安装：pip install aredis

单节点

import asyncio
from aredis import StrictRedis
 
 
async def example():
    client = StrictRedis(host='127.0.0.1', port=6379, db=0)
    await client.flushdb()
    await client.set('foo', 1)
    assert await client.exists('foo') is True
    await client.incr('foo', 100)
 
    assert int(await client.get('foo')) == 101
    await client.expire('foo', 1)
    await asyncio.sleep(0.1)
    await client.ttl('foo')
    await asyncio.sleep(1)
    assert not await client.exists('foo')
 
 
loop = asyncio.get_event_loop()
loop.run_until_complete(example())

集群版

import asyncio
from aredis import StrictRedisCluster
 
 
async def example():
    client = StrictRedisCluster(host='172.17.0.2', port=7001)
    await client.flushdb()
    await client.set('foo', 1)
    await client.lpush('a', 1)
    print(await client.cluster_slots())
 
    await client.rpoplpush('a', 'b')
    assert await client.rpop('b') == b'1'
 
 
loop = asyncio.get_event_loop()
loop.run_until_complete(example())

2、aioredis

github 地址：https://github.com/aio-libs/aioredis
官方文档：https://aioredis.readthedocs.io/en/v1.3.0/

从 redis.py  4.2.0rc1+ 开始，Aioredis 已经集成到 redis.py 里面，

导入：from redis import asyncio as aioredis

安装：pip install aioredis

连接 redis

import asyncio
import aioredis
 
 
async def main():
    redis = await aioredis.create_redis_pool('redis://localhost')
    await redis.set('my-key', 'value')
    value = await redis.get('my-key', encoding='utf-8')
    print(value)
 
    redis.close()
    await redis.wait_closed()
 
asyncio.run(main())

连接到指定 db

1. 指定 db 参数：redis = await aioredis.create_redis_pool('redis://localhost', db=1)
2. 在 URL 中指定 db：redis = await aioredis.create_redis_pool('redis://localhost/2')
3. 使用 select 方法：
   redis = await aioredis.create_redis_pool('redis://localhost/')
   await redis.select(3)

连接带密码的 redis

# 密码可以通过关键之指定，也可以通过 URL 指定：

redis = await aioredis.create_redis_pool('redis://localhost', password='sEcRet')

redis = await aioredis.create_redis_pool('redis://:sEcRet@localhost/')

结果编码：aioredis 默认返回字节类型，可以通过传递关键字 encoding="utf-8" 自动解码，也可以获取到字节类型后，通过 decode("utf-8") 进行解码。

示例代码：

import asyncio
import aioredis
 
 
async def main():
    redis = await aioredis.create_redis_pool('redis://localhost')
 
    await redis.set('key', 'string-value')
    bin_value = await redis.get('key')
    assert bin_value == b'string-value'
    str_value = await redis.get('key', encoding='utf-8')
    assert str_value == 'string-value'
 
    await redis.hmset_dict(
        'hash', key1='value1', key2='value2', key3=123
    )
    result = await redis.hgetall('hash', encoding='utf-8')
    assert result == {
        'key1': 'value1',
        'key2': 'value2',
        'key3': '123',  # 注意: redis 返回的int会作为str对待
    }
 
    redis.close()
    await redis.wait_closed()
 
 
asyncio.run(main())

简单 低级别 接口

import asyncio
import aioredis
 
loop = asyncio.get_event_loop()
 
 
async def go():
    conn = await aioredis.create_connection(('localhost', 6379), loop=loop)
    await conn.execute('set', 'my-key', 'value')
    val = await conn.execute('get', 'my-key')
    print(val)
    conn.close()
    await conn.wait_closed()
 
 
loop.run_until_complete(go())
# will print 'value'

简单 高级别 接口

import asyncio
import aioredis
 
loop = asyncio.get_event_loop()
 
 
async def go():
    redis = await aioredis.create_redis(('localhost', 6379), loop=loop)
    await redis.set('my-key', 'value')
    val = await redis.get('my-key')
    print(val)
    redis.close()
    await redis.wait_closed()
 
 
loop.run_until_complete(go())
# will print 'value'

连接池

import asyncio
import aioredis
 
loop = asyncio.get_event_loop()
 
 
async def func_1():
    conn = await aioredis.create_connection(
        ('localhost', 6379), loop=loop)
    await conn.execute('set', 'my-key', 'value')
    val = await conn.execute('get', 'my-key')
    print(val)
    conn.close()
    await conn.wait_closed()
 
 
async def func_2():
    redis = await aioredis.create_redis(('localhost', 6379), loop=loop)
    await redis.set('my-key', 'value')
    val = await redis.get('my-key')
    print(val)
    redis.close()
    await redis.wait_closed()
 
 
async def func_3():
    pool = await aioredis.create_pool(
        ('localhost', 6379),
        minsize=5, maxsize=10,
        loop=loop
    )
    async with await pool as redis:  # 高级别 redis API 实例
        await redis.set('my-key', 'value')
        print(await redis.get('my-key'))
    # 优雅的关闭
    pool.close()
    await pool.wait_closed()
 
 
async def func_4():
    redis_pool = await aioredis.create_pool(
        ('localhost', 6379), 
        minsize=5, maxsize=10, 
        loop=loop
    )
    async with redis_pool.get() as conn:  # 高级别 redis API 实例
        await conn.execute('set', 'my-key', 'value')
        print(await conn.execute('get', 'my-key'))
    # graceful shutdown
    redis_pool.close()
    await redis_pool.wait_closed()
 
 
loop.run_until_complete(func_1)

示例：

from sanic import Sanic, response
import aioredis
 
app = Sanic(__name__)
 
 
@app.route("/")
async def handle(request):
    async with request.app.redis_pool.get() as redis:
        await redis.execute('set', 'my-key', 'value')
        val = await redis.execute('get', 'my-key')
    return response.text(val.decode('utf-8'))
 
 
@app.listener('before_server_start')
async def before_server_start(app, loop):
    app.redis_pool = await aioredis.create_pool(
        ('localhost', 6379),
        minsize=5,
        maxsize=10,
        loop=loop
    )
 
 
@app.listener('after_server_stop')
async def after_server_stop(app, loop):
    app.redis_pool.close()
    await app.redis_pool.wait_closed()
 
 
if __name__ == '__main__':
    app.run(host="0.0.0.0", port=80)

事务（ Multi/Exec ）

import asyncio
import aioredis
 
 
async def main():
    redis = await aioredis.create_redis_pool('redis://localhost')
 
    tr = redis.multi_exec()
    tr.set('key1', 'value1')
    tr.set('key2', 'value2')
    ok1, ok2 = await tr.execute()
    assert ok1
    assert ok2
 
asyncio.run(main())

multi_exec() 创建和返回一个新的 MultiExec 对象用于缓冲命令，然后在 MULTI / EXEC 块中执行它们。

重要提示：不要在 类似 ( tr.set('foo', '123') ) 上 使用 await buffered 命令， 因为它将被永远阻塞。

下面的代码将会给永远阻塞：

tr = redis.multi_exec()
await tr.incr('foo')   # that's all. we've stuck!

发布 和 订阅 模式

aioredis 提供了对 Redis 的 发布/订阅（Publish / Subscribe） 消息的支持。

To start listening for messages you must call either subscribe() or psubscribe() method. Both methods return list of Channel objects representing subscribed channels.

Right after that the channel will receive and store messages (the Channel object is basically a wrapper around asyncio.Queue). To read messages from channel you need to use get() or get_json() coroutines.

要开始监听消息，必须调用 subscribe() 或 psubscribe() 方法。这两个方法都返回一个列表，列表中的元素是 "订阅的 Channel(通道) 对象"。在此之后，Channel(通道) 将接收并存储消息 ( "Channel(通道) " 基本上是 asyncio.Queue 的包装器)。要从 channel 中读取消息，需要使用get()或get_json()协程。

订阅 和 阅读 频道 示例:

import asyncio
import aioredis
 
 
async def main():
    redis = await aioredis.create_redis_pool('redis://localhost')
 
    ch1, ch2 = await redis.subscribe('channel:1', 'channel:2')
    assert isinstance(ch1, aioredis.Channel)
    assert isinstance(ch2, aioredis.Channel)
 
    async def reader(channel):
        async for message in channel.iter():
            print("Got message:", message)
    asyncio.get_running_loop().create_task(reader(ch1))
    asyncio.get_running_loop().create_task(reader(ch2))
 
    await redis.publish('channel:1', 'Hello')
    await redis.publish('channel:2', 'World')
 
    redis.close()
    await redis.wait_closed()
 
asyncio.run(main())

订阅 和 阅读 模式:

import asyncio
import aioredis
 
 
async def main():
    redis = await aioredis.create_redis_pool('redis://localhost')
 
    ch, = await redis.psubscribe('channel:*')
    assert isinstance(ch, aioredis.Channel)
 
    async def reader(channel):
        async for ch, message in channel.iter():
            print("Got message in channel:", ch, ":", message)
    asyncio.get_running_loop().create_task(reader(ch))
 
    await redis.publish('channel:1', 'Hello')
    await redis.publish('channel:2', 'World')
 
    redis.close()
    await redis.wait_closed()
 
asyncio.run(main())

Sentinel ( 哨兵 )

Redis(主从复制、哨兵模式、集群)：https://blog.csdn.net/Bilson99/article/details/118732296

哨兵的核心功能：在主从复制的基础上，哨兵引入了主节点的自动故障转移

哨兵模式的原理：哨兵(sentinel) 是一个分布式系统，用于对主从结构中的每台服务器进行监控，当出现故障时通过投票机制选择新的 Master 并将所有 Slave 连接到新的 Master。所以整个运行哨兵的集群的数量不得少于3个节点。

哨兵模式的作用

• 监控：哨兵会不断地检查主节点和从节点是否运作正常。
• 自动故障转移：当主节点不能正常工作时，哨兵会开始自动故障转移操作，它会将失效主节点的其中一个从节点升级为新的主节点，并让其他从节点改为复制新的主节点。
• 通知（提醒）：哨兵可以将故障转移的结果发送给客户端。

哨兵结构由两部分组成，哨兵节点和数据节点：

• 哨兵节点：哨兵系统由一个或多个哨兵节点组成，哨兵节点是特殊的redis节点，不存储数据。
• 数据节点：主节点和从节点都是数据节点。

哨兵的启动依赖于主从模式，所以须把主从模式安装好的情况下再去做哨兵模式，所有节点上都需要部署哨兵模式，哨兵模式会监控所有的 Redis 工作节点是否正常，当 Master 出现问题的时候，因为其他节点与主节点失去联系，因此会投票，投票过半就认为这个 Master 的确出现问题，然后会通知哨兵间，然后从 Slaves 中选取一个作为新的 Master。

需要特别注意的是，客观下线是主节点才有的概念；如果从节点和哨兵节点发生故障，被哨兵主观下线后，不会再有后续的客观下线和故障转移操作。

import asyncio
import aioredis
 
 
async def main():
    sentinel = await aioredis.create_sentinel(
        ['redis://localhost:26379', 'redis://sentinel2:26379'])
    redis = sentinel.master_for('mymaster')
 
    ok = await redis.set('key', 'value')
    assert ok
    val = await redis.get('key', encoding='utf-8')
    assert val == 'value'
 
asyncio.run(main())

Sentinel 客户端需要一个 Redis Sentinel 地址列表，来连接并开始发现服务。

调用 master_for() 或 slave_for() 方法 将返回连接到 Sentinel 监视的指定服务的 Redis 客户端。

Sentinel 客户端将自动检测故障转移并重新连接 Redis 客户端。

import asyncio
import aioredis
 
loop = asyncio.get_event_loop()
 
async def go():
    conn = await aioredis.create_connection(
        ('localhost', 6379), loop=loop)
    await conn.execute('set', 'my-key', 'value')
    val = await conn.execute('get', 'my-key')
    print(val)
    conn.close()
    await conn.wait_closed()
loop.run_until_complete(go())
# will print 'value'

示例

import uuid
import time
import json
import datetime
import uvicorn
from pathlib import Path
from fastapi import FastAPI
from typing import Optional
import redis
from redis import asyncio as aioredis
from concurrent.futures import ThreadPoolExecutor, wait
 
 
redis_config = {
    'host': '172.16.30.180',
    'port': 6379,
    'db': 1,
}
 
redis_db_yibu = aioredis.StrictRedis(**redis_config)
redis_db_tongbu = redis.StrictRedis(**redis_config)
app = FastAPI()
 
 
@app.get("/api_test")
async def func_handle_request(q: Optional[str] = None):
    """和 Flask 不同，Flask 是使用 <>，而 FastAPI 使用 {}"""
    print(f'q ---> {q}')
    current_timestamp = datetime.datetime.now().timestamp()
    req_uuid = str(uuid.uuid5(uuid.NAMESPACE_URL, f'{current_timestamp}{q}'))
    await redis_db_yibu.hset('api_test_req', req_uuid, q)
 
    while True:
        result = await redis_db_yibu.hget('api_test_resp', req_uuid)
        if not result:
            time.sleep(0.1)
            print('睡眠 100ms 继续监听')
            continue
        break
 
    return {"result": result}
 
 
def http_server():
    """
    :return:
    """
    print(f'{Path(__file__).stem}:app')
    uvicorn.run(f'{Path(__file__).stem}:app', host="0.0.0.0", port=9999)
 
 
def func_consumer(task_string=None):
    task_dict = json.loads(task_string)
    data = {'resp': str(datetime.datetime.now())}
    for k, v in task_dict.items():
        redis_db_tongbu.hset('api_test_resp', k, json.dumps(data, ensure_ascii=False))
        redis_db_tongbu.hdel('api_test_req', k)
        print(f'请求 [{k}] ---> 处理成功')
 
 
def func_producer():
    worker_count = 50
    with ThreadPoolExecutor(max_workers=worker_count) as pool:
        while True:
            task_list = redis_db_tongbu.hgetall('api_test_req')
            if task_list:
                for k, v in task_list.items():
                    task_dict = {k.decode('utf-8'): v.decode('utf-8')}
                    task_string = json.dumps(task_dict, ensure_ascii=False)
                    pool.submit(func_consumer, task_string)
                pass
            else:
                # print('task 为空，睡100ms继续')
                time.sleep(0.1)
    pass
 
 
def main():
    with ThreadPoolExecutor(max_workers=2) as pool:
        pool.submit(http_server)
        pool.submit(func_producer)
 
 
if __name__ == '__main__':
    main()
    pass

3、asynio_redis

GitHub 地址：https://github.com/jonathanslenders/asyncio-redis

官方英文文档：https://asyncio-redis.readthedocs.io/en/latest/

安装：pip install asyncio_redis

asynio_redis(下划线) 和 asyncio-redis(中划线）都已不再更新，推荐 aioredis

示例：Connection 类

A asyncio_redis.Connection instance will take care of the connection and will automatically reconnect, using a new transport when the connection drops. This connection class also acts as a proxy to a asyncio_redis.RedisProtocol instance; any Redis command of the protocol can be called directly at the connection.

import asyncio
import asyncio_redis
 
 
async def example():
    # Create Redis connection
    connection = await asyncio_redis.Connection.create(host='127.0.0.1', port=6379)    
 
    # Set a key
    await connection.set('my_key', 'my_value')
 
    # When finished, close the connection.
    connection.close()
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(example())

示例：连接池

Requests will automatically be distributed among all connections in a pool. If a connection is blocking because of --for instance-- a blocking rpop, another connection will be used for new commands.

import asyncio
import asyncio_redis
 
 
async def example():
    # Create Redis connection
    connection = await asyncio_redis.Pool.create(host='127.0.0.1', port=6379, poolsize=10)
 
    # Set a key
    await connection.set('my_key', 'my_value')
 
    # When finished, close the connection pool.
    connection.close()
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(example())

示例

import asyncio
import asyncio_redis
 
 
async def example():
    # Create Redis connection
    connection = await asyncio_redis.Pool.create(
        host='127.0.0.1', port=6379, poolsize=10
    )
 
    # Create transaction
    transaction = await connection.multi()
 
    # Run commands in transaction (they return future objects)
    f1 = await transaction.set('key', 'value')
    f2 = await transaction.set('another_key', 'another_value')
 
    # Commit transaction
    await transaction.exec()
 
    # Retrieve results
    result1 = await f1
    result2 = await f2
 
    # When finished, close the connection pool.
    connection.close()

只要有事务在其中运行，连接就会被占用。建议使用足够大的池大小。

示例：发布 / 订阅

Pub / sub 

import asyncio
import asyncio_redis
 
 
async def example():
    # Create connection
    connection = await asyncio_redis.Connection.create(host='127.0.0.1', port=6379)
 
    # Create subscriber.
    subscriber = await connection.start_subscribe()
 
    # Subscribe to channel.
    await subscriber.subscribe([ 'our-channel' ])
 
    # Inside a while loop, wait for incoming events.
    while True:
        reply = await subscriber.next_published()
        print('Received: ', repr(reply.value), 'on channel', reply.channel)
 
    # When finished, close the connection.
    connection.close()

示例：LUA 脚本

import asyncio
import asyncio_redis
 
code = """
local value = redis.call('GET', KEYS[1])
value = tonumber(value)
return value * ARGV[1]
"""
 
 
async def example():
    connection = await asyncio_redis.Connection.create(host='127.0.0.1', port=6379)
 
    # Set a key
    await connection.set('my_key', '2')
 
    # Register script
    multiply = await connection.register_script(code)
 
    # Run script
    script_reply = await multiply.run(keys=['my_key'], args=['5'])
    result = await script_reply.return_value()
    print(result)  # prints 2 * 5
 
    # When finished, close the connection.
    connection.close()

示例：使用 Protocol类

import asyncio
import asyncio_redis
 
 
async def example():
    loop = asyncio.get_event_loop()
 
    # Create Redis connection
    transport, protocol = await loop.create_connection(
        asyncio_redis.RedisProtocol, '127.0.0.1', 6379
    )
 
    # Set a key
    await protocol.set('my_key', 'my_value')
 
    # Get a key
    result = await protocol.get('my_key')
    print(result)
 
    # Close transport when finished.
    transport.close()
 
 
if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(example())
    pass