抽象后端的Redis消息队列后端
基于Redis Streams的生产就绪消息队列后端实现 抽象后端 项目。该后端扩展了消息队列功能以支持Redis，实现了可靠、可扩展和高性能的抽象后端事件处理。
状态和质量
CI/CD和测试
  
代码覆盖率和质量
    
代码样式和标准
   
包信息
](https://badge.fury.io/py/abe-redis) ](https://pypi.org/project/abe-redis) 
下载
](https://pepy.tech/project/abe-redis) ](https://pepy.tech/project/abe-redis) ](https://pepy.tech/project/abe-redis)
______________________________________________________________________
概述
abe redis 是一个基于Redis Streams的消息队列后端，与抽象后端的组件加载机制无缝集成。它为使用Redis作为消息队列基础设施处理抽象后端事件提供了一种可靠、可扩展的解决方案。
主要特点
- 🔌 即插即用：通过pip安装并配置环境变量
- ⚡ Redis流：面向消费者群体的现代流式消息处理
- 🚀 生产就绪：内置连接池、错误处理和重试逻辑
- 🔄 异步优先：为现代Python异步/等待模式构建
- 📦 通用兼容性：适用于使用相同组件加载机制的任何项目
- 🧪 测试良好：全面的单元和集成测试，覆盖率高
- 📚 完整记录：完整的API参考和使用示例
Python版本支持
Python 3.12、3.13
快速开始
安装
通过pip安装软件包：
pip install abe-redis配置
配置您的环境以使用Redis作为消息队列后端：
export QUEUE_BACKEND=redis
export REDIS_URL=redis://localhost:6379/0启动Redis
使用Docker（推荐）：
docker run -d --name test-redis -p 6379:6379 redis:7-alpine基本用法
import asyncio
from abe_plugin.backends.message_queue import RedisMessageQueueBackend
async def main():
# Create backend from environment variables
backend = RedisMessageQueueBackend.from_env()
# Publish a message
await backend.publish("slack:events", {"type": "message", "text": "Hello Redis!"})
# Consume messages
async for message in backend.consume():
print(f"Received: {message}")
break
await backend.close()
asyncio.run(main())建筑
Redis后端使用 Redis流 为了实现可靠的消息排队：
- 持久化存储：存储在Redis Streams中的消息具有可配置的保留期
- 消费群体：跨多个员工的分布式消费
- 自动确认：成功处理后确认的消息
- 流模式匹配：自动发现并使用
slack:*流 - 连接池：具有可配置池大小的高效连接管理
配置选项
| 环境变量 | 必填 | 默认 | 描述 |
|---|---|---|---|
QUEUE_BACKEND | Yes | - | 必须设置为 redis |
REDIS_URL | 是的 | redis://localhost:6379/0 | Redis连接URL |
REDIS_PASSWORD | 否 | 无 | Redis身份验证密码 |
REDIS_SSL | 没有 | false | 启用SSL/TLS连接 |
REDIS_MAX_CONNECTIONS | 没有 | 10 | 最大连接池大小 |
REDIS_STREAM_MAXLEN | 没有 | 10000 | 修剪的最大流长度 |
文档
有关全面的文档，包括API参考资料、示例和开发指南：
📚 全部文件
快速链接
发展
先决条件
- Python 3.12或3.13
- Redis 6.0+（推荐7.0+）
- uv或pip包管理器
设置开发环境
# Clone the repository
git clone https://github.com/Chisanan232/abe-redis.git
cd abe-redis
# Install dependencies
uv sync
# Start Redis for testing
docker run -d --name redis-dev -p 6379:6379 redis:7-alpine
# Run tests
uv run pytest
# Run tests with coverage
uv run pytest --cov=abe_plugin --cov-report=html代码质量
该项目遵循严格的代码质量标准：
运行质量检查：
# Format code
uv run black abe_plugin/ test/
# Lint code
uv run pylint abe_plugin/
# Sort imports
uv run isort abe_plugin/ test/
# Type check
uv run mypy abe_plugin/贡献
我们欢迎捐款！请查看我们的 贡献指南 了解详情。
如何贡献
- 分叉存储库
- 创建要素分支(
git checkout -b feature/amazing-feature) - 进行更改
- 运行测试和质量检查
- 提交您的更改(
git commit -m 'feat: add amazing feature') - 推到分支(
git push origin feature/amazing-feature) - 打开拉取请求
许可证
此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。
链接
- 文档: https://chisanan232.github.io/abe-redis/
- PyPI包: https://pypi.org/project/abe-redis/
- 源代码: https://github.com/Chisanan232/abe-redis
- 问题追踪: https://github.com/Chisanan232/abe-redis/issues
- 抽象后端: https://github.com/Chisanan232/abstract-backend
支持
如果您遇到任何问题或有疑问：
- 📖 检查 文档
- 🐛 通过以下方式报告错误
- 💬 在中提问
______________________________________________________________________
由以下材料制成❤️ 抽象后端生态系统