pip install aioredis

export APP_ENV=development

export REDIS_URL="192.168.70.130/"
export REDIS_USER=
export REDIS_PASSWORD=
export REDIS_HOST="192.168.70.130"
export REDIS_PORT=6379

import os
from dotenv import load_dotenv
import aioredis
import asyncio

load_dotenv()

class Redis():
? ? def __init__(self):
? ? ? ? """initialize ?connection """
? ? ? ? self.REDIS_URL = os.environ['REDIS_URL']
? ? ? ? self.REDIS_PASSWORD = os.environ['REDIS_PASSWORD']
? ? ? ? self.REDIS_USER = os.environ['REDIS_USER']
? ? ? ? self.connection_url = f"redis://{self.REDIS_USER}:{self.REDIS_PASSWORD}@{self.REDIS_URL}"
? ? ? ? self.REDIS_HOST = os.environ['REDIS_HOST']
? ? ? ? self.REDIS_PORT = os.environ['REDIS_PORT']
? ? ? ??
? ? async def create_connection(self):
? ? ? ? self.connection = aioredis.from_url(
? ? ? ? ? ? self.connection_url, db=0)

? ? ? ? return self.connection


class Producer:
? ? def __init__(self, redis_client):
? ? ? ? self.redis_client = redis_client

? ? async def add_to_stream(self, ?data: dict, stream_channel):
? ? ? ? """將一條數(shù)據(jù)添加到隊列

? ? ? ? Args:
? ? ? ? ? ? data (dict): _description_
? ? ? ? ? ? stream_channel (_type_): _description_

? ? ? ? Returns:
? ? ? ? ? ? _type_: _description_
? ? ? ? """
? ? ? ? try:
? ? ? ? ? ? msg_id = await self.redis_client.xadd(name=stream_channel, id="*", fields=data)
? ? ? ? ? ? print(f"Message id {msg_id} added to {stream_channel} stream")
? ? ? ? ? ? return msg_id

? ? ? ? except Exception as e:
? ? ? ? ? ? raise Exception(f"Error sending msg to stream => {e}")

class StreamConsumer:
? ? def __init__(self, redis_client):
? ? ? ? self.redis_client = redis_client

? ? async def consume_stream(self, count: int, block: int, ?stream_channel):
? ? ? ? """讀取隊列中的消息，但是并不刪除

? ? ? ? Args:
? ? ? ? ? ? count (int): _description_
? ? ? ? ? ? block (int): _description_
? ? ? ? ? ? stream_channel (_type_): _description_

? ? ? ? Returns:
? ? ? ? ? ? _type_: _description_
? ? ? ? """
? ? ? ? response = await self.redis_client.xread(
? ? ? ? ? ? streams={stream_channel: ?'0-0'}, count=count, block=block)

? ? ? ? return response

? ? async def delete_message(self, stream_channel, message_id):
? ? ? ? """成功消費數(shù)據(jù)后，調(diào)用此函數(shù)刪除隊列數(shù)據(jù)

? ? ? ? Args:
? ? ? ? ? ? stream_channel (_type_): _description_
? ? ? ? ? ? message_id (_type_): _description_
? ? ? ? """
? ? ? ? await self.redis_client.xdel(stream_channel, message_id)


async def main():
? ? redis_conn = await Redis().create_connection()
? ? produce = Producer(redis_conn)
? ? consumer = StreamConsumer(redis_conn)
? ? # 添加一個消息到隊列中
? ? data = {'xiaoming4':123}
? ? await produce.add_to_stream(data=data,stream_channel='message_channel')
? ??
? ? # 從隊列中拿出最新的1條數(shù)據(jù)
? ? data = await consumer.consume_stream(1,block=0,stream_channel='message_channel')
? ? print(data)
? ??
? ? # 輪詢等待隊列中的新消息
? ? response = await consumer.consume_stream(stream_channel="message_channel", count=1, block=0)
? ? if response:
? ? ? ? for stream, messagees in response:
? ? ? ? ? ? print('stream:',stream)
? ? ? ? ? ? for message in messagees:
? ? ? ? ? ? ? ? print('message: ',message)
? ? ? ? ? ? ? ? message_id = message[0]
? ? ? ? ? ? ? ? print('message_id: ',message_id)
? ? ? ? ? ? ? ? message_content = message[1]
? ? ? ? ? ? ? ? print('message_content: ',message_content)
? ? ? ? ? ? ? ? print('注意里面的鍵、值都變成了byte類型，需要進行解碼:')
? ? ? ? ? ? ? ? message_content:dict
? ? ? ? ? ? ? ? print('message_content_decode: ',{k.decode('utf-8'):v.decode('utf-8') for k,v in message_content.items()})

? ? # 消費成功后刪除隊列中的消息
? ? await consumer.delete_message(
? ? ? ? stream_channel='message_channel',message_id=message_id
? ? ) ? ?

if __name__ == '__main__':
? ? asyncio.run(main())