AI导哈

Redis消息队列后端

一个基于Redis Streams的生产就绪消息队列后端实现，为Abstract Backend项目提供可靠、可扩展且高性能的事件处理能力。

  • by Chisanan232
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Chisanan232/MCP-BackEnd-Message-Queue-Redis

使用说明

抽象后端的Redis消息队列后端

基于Redis Streams的生产就绪消息队列后端实现 抽象后端 项目。该后端扩展了消息队列功能以支持Redis，实现了可靠、可扩展和高性能的抽象后端事件处理。

状态和质量

CI/CD和测试

代码覆盖率和质量

代码样式和标准

包信息

下载

概述

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/

贡献

我们欢迎捐款！请查看我们的 贡献指南 了解详情。

如何贡献

  1. 分叉存储库
  2. 创建要素分支(git checkout -b feature/amazing-feature)
  3. 进行更改
  4. 运行测试和质量检查
  5. 提交您的更改(git commit -m 'feat: add amazing feature')
  6. 推到分支(git push origin feature/amazing-feature)
  7. 打开拉取请求

许可证

此项目根据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

支持

如果您遇到任何问题或有疑问：

由以下材料制成❤️ 抽象后端生态系统

相关 MCP