Python 异步 redis

 

现在的 Python 的异步 redis，有三种（ aredis 、aioredis、asynio_redis)

 

 

aredis 、aioredis、asynio_redis 对比

 

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 的功能更为全面一些。

 

asyncio 从 redis 取任务

asyncio 怎么与 redis 队列协同？：https://www.v2ex.com/amp/t/489042

 

 

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

 

开始使用：

更多使用示例：https://github.com/NoneGG/aredis/tree/master/examples

 

1. 单节点版

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())

 

2. 集群版

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/

 

开始使用

 

安装：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())

simple low-level interface：

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'

simple high-level interface：

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'

Connections pool：

import asyncio
import aioredis

loop = asyncio.get_event_loop()


async def go():
    pool = await aioredis.create_pool(
        ('localhost', 6379), minsize=5, maxsize=10, loop=loop
    )
    with await pool as redis:  # high-level redis API instance
        await redis.set('my-key', 'value')
        print(await redis.get('my-key'))
    # graceful shutdown
    pool.close()
    await pool.wait_closed()


loop.run_until_complete(go())

 

连接到指定 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 实例：

The password can be specified either in keyword argument or in address URI:

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

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

 

结果编码：

By default aioredis will return bytes for most Redis commands that return string replies. Redis error replies are known to be valid UTF-8 strings so error messages are decoded automatically.

If you know that data in Redis is valid string you can tell aioredis to decode result by passing keyword-only argument encoding in a command call:

示例代码：

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'

    redis.close()
    await redis.wait_closed()

asyncio.run(main())

示例代码：

import asyncio
import aioredis


async def main():
    redis = await aioredis.create_redis_pool('redis://localhost')

    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',  # note that Redis returns int as string
    }

    redis.close()
    await redis.wait_closed()

asyncio.run(main())

 

事务（ 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() method creates and returns new MultiExec object which is used for buffering commands and then executing them inside MULTI/EXEC block.

重要提示：不要在 类似 ( 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.

订阅 和 阅读 频道 示例:

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 client

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'

连接池

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)

 

示例：

import asyncio
import aioredis
 
loop = asyncio.get_event_loop()
 
 
async def go_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 go_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 go_3():
    redis_pool = await aioredis.create_pool(('localhost', 6379), minsize=5, maxsize=10, loop=loop)
    async with redis_pool.get() as conn:  # high-level redis API instance
        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(go_1())
# loop.run_until_complete(go_2())
# loop.run_until_complete(go_3())

 

 

3. asynio_redis

 

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

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

 

安装：pip install asyncio_redis

 

The connection class

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())

 

Connection pooling

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())

 

Transactions 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()

It's recommended to use a large enough poolsize. A connection will be occupied as long as there's a transaction running in there.

 

Pub / sub example

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 Scripting example

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()

 

Example using the Protocol class

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

 

 

asyncio-redis 是 Python asyncio 的 Redis 客户端 (PEP 3156)。这个 Redis 库是完全异步的，Reids 服务器非阻塞客户端，依赖于 asyncio，所以要求 Python 3.3. 以上版本。

安装：pip install asyncio-redis

创建一个redisUtils.py

class Redis:
    """
    A simple wrapper class that allows you to share a connection
    pool across your application.
    """
    _pool = None
 
    async def get_redis_pool(self):
        REDIS_CONFIG = {'host': 'localhost', 'port': 6379}
        try:
            from config import REDIS_CONFIG
        except:
            pass
        if not self._pool:
            self._pool = await asyncio_redis.Pool.create(
                host=REDIS_CONFIG['host'], port=REDIS_CONFIG['port'], password=REDIS_CONFIG.get('password'), poolsize=10
            )
 
        return self._pool
 
    async def close(self):
        if self._pool:
            self._pool.close()

再创建一个run.py

from utils.redisUtils import Redis  #引用上面的配置
import json as jsonn
 
 
redis = Redis()
r = await redis.get_redis_pool()
key = '/hushuai/product'
await r.set(key, 'value')
val = await r.get(key)
print(val)
 
key_list = '/hushuai/product_list'
product_list_size = await r.llen(key_list)
print(product_list_size)
if product_list_size != 0:
    if product_list_size > start_page:
        product_list = await r.lrange(key, start_page, end_page)
        product_list = await product_list.aslist()
        product_list_other = []
        for i in product_list:
            product_list_other.append(jsonn.loads(i.replace('\'', '\"').replace('None', '""')))
        data = [product_list_size, product_list_other]
    else:
        data = await get_items(app.pool,"product_view",re_params,with_total=with_total,pager=pager)
else:
    data = await get_items(app.pool, "product_view", re_params, with_total=with_total, pager=pager)
    data_redis = await get_items(app.pool, "product_view", re_params)
    list = []
    for product_data in data_redis:
        list.append(str(product_data))
    if list:
        await r.rpush(key, list)

主要讲的是python 使用异步redis的方式，这里只是做了redis的str和list两种类型的数据处理。