Metadata-Version: 2.4
Name: redis-xsync
Version: 1.0.3
Summary: Sync/Async unified Redis client with distributed lock and rate limiter
Author: Jay
License: MIT
Project-URL: Homepage, https://github.com/jaygarage/RedisXsync
Project-URL: Repository, https://github.com/jaygarage/RedisXsync
Project-URL: Issues, https://github.com/jaygarage/RedisXsync/issues
Keywords: redis,async,sync,distributed-lock,redis-xsync,rate-limit
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.0
Requires-Dist: redis>=5.0.0

# RedisXsync

![PyPI](https://img.shields.io/pypi/v/redis-xsync) ![Python](https://img.shields.io/pypi/pyversions/redis-xsync) ![License](https://img.shields.io/badge/license-MIT-green)

RedisXsync 是一个 **同步 / 异步统一 Redis 客户端**，专为 **分布式锁** 和 **爬虫账号池限流** 场景设计，支持多实例、多
DB，线程安全和协程安全。

```bash
# 安装
pip install redis-xsync

# 导包
from redis_xsync import register_redis, resolve_redis, redisXsync, RedisXsync
```

---

## ✨ 核心特性

| 特性          | 描述                                             |
|-------------|------------------------------------------------|
| 同步 / 异步统一接口 | 同一套 API 支持 `with` 和 `async with`               |
| 多实例 / 多 DB  | 轻松管理多个 Redis 实例和数据库                            |
| 分布式锁        | 支持单键锁、组合多键锁，阻塞/非阻塞模式                           |
| 账号池限流       | 基于 Redis + RediSearch + Lua 脚本，实现高并发、安全的分布式令牌池 |
| 线程 & 协程安全   | 避免多线程/协程下的连接冲突                                 |
| 自动连接管理      | Context manager 自动获取和释放连接                      |

---

## 🏗 架构概览

| RedisXsync               |
|--------------------------|
| Sync / Async Interfaces  |
| Distributed Lock Manager |
| Rate Limiter / TokenPool |

- 分布式锁：通过单键或多键原子操作保证并发安全
- 限流/账号池：Lua 脚本保证跨进程/跨机器的原子性，精确到毫秒

---

## 📦 安装

```bash
pip install redis-xsync

---

## 🚀 快速开始

### 1️⃣ 注册 Redis

​```python
from redis_xsync import register_redis, resolve_redis, redisXsync

redis1 = register_redis(
    label="redis_xsync",
    host="127.0.0.1",
    port=6379,
    db=0,
    password=None,
    decode_responses=True
)

# 通过 label 获取
redis2 = resolve_redis("redis_xsync")


# 使用方式说明：
# 1️⃣ 全局默认实例 redisXsync
#    - 在首次调用 register_redis 注册任意 Redis 时，首次注册会自动赋值给 redisXsync，非并发安全
#    - 后续可以直接使用 redisXsync，无需通过 label 获取
#
# 2️⃣ 通过 label 显式管理 Redis
#    - register_redis(label=..., ...) 注册 Redis 实例
#    - resolve_redis(label) 根据 label 获取对应 Redis 实例
#
# 3️⃣ 独立实例
#    - 可以直接创建 RedisXsync() 实例，不必参与 register_redis 、resolve_redis、 redisXsync 全局管理
#    - 适合临时或隔离使用场景
```

## 🔄 同步 / 异步使用

### ✅ 同步

```python
with redis(db=1) as r:
    r.set("key", "value")
    print(r.get("key"))
```

---

### ✅ 异步

```python
import asyncio


async def main():
    async with redis(db=1) as r:
        await r.set("key", "value")
        print(await r.get("key"))


asyncio.run(main())
```

---

## 🔐 分布式锁（支持组合锁）

### 特性

* 多 key 原子加锁
* 所有 key 释放才算解锁
* 支持阻塞等待

---

### 示例（多线程 + 协程混合）

```python
import threading
import asyncio
from redis_xsync import register_redis

cx = 0
redis = register_redis(label="redis_xsync", host="127.0.0.1", port=6379)
Lock = redis.AsyncRedisAtomicMultiLock


def sync_worker():
    global cx
    for _ in range(5):
        with Lock("lock_key", db=0, ttl_ms=10, blocking=True):
            with redis(db=2) as r:
                r.setnx("db2", str(cx))
                cx += 1
                print(f"[SYNC] {cx=}")


async def async_worker():
    global cx
    for _ in range(5):
        with Lock("lock_key", db=0, ttl_ms=10, blocking=True):
            async with redis(db=3) as r:
                await r.setnx("db3", str(cx))
                cx += 1
                print(f"[ASYNC] {cx=}")
        await asyncio.sleep(0.2)


# 启动线程
threading.Thread(target=sync_worker).start()

# 启动协程
asyncio.run(async_worker())
```

---

## 🚦 爬虫账号池限流

基于 RediSearch + Lua，实现**分布式限流 + 自动等待可用账号**

📦 Limited — RediSearch 容器操作方法对照表

| 方法                                                      | 异步 | 阻塞 | 遵循限流 | 数量 | 描述                       |
|---------------------------------------------------------|----|----|------|----|--------------------------|
| `set_available_containers(*containers)`                 | ✅  | ❌  | ✅    | N  | 存储/更新容器配置（生产者）           |
| `ask_available_containers(quantity)`                    | ✅  | ❌  | ✅    | N  | 获取 N 个可用容器，若无返回 `None`   |
| `ask_available_container()`                             | ✅  | ❌  | ✅    | 1  | 获取单个可用容器，若无返回 `None`     |
| `wait_ask_available_containers(quantity, timeout)`      | ✅  | ✅  | ✅    | N  | 阻塞等待 N 个可用容器             |
| `wait_ask_available_container(timeout)`                 | ✅  | ✅  | ✅    | 1  | 阻塞等待单个可用容器               |
| `acquire_random_containers(quantity)`                   | ✅  | ❌  | ❌    | N  | 随机获取 N 个容器，忽略限流          |
| `acquire_random_container()`                            | ✅  | ❌  | ❌    | 1  | 随机获取单个容器，忽略限流            |
| `wait_acquire_random_containers(quantity, timeout)`     | ✅  | ✅  | ❌    | N  | 乐观锁：阻塞等待 N 个随机容器，忽略限流    |
| `wait_acquire_random_container(timeout)`                | ✅  | ✅  | ❌    | 1  | 乐观锁：阻塞等待单个随机容器，忽略限流      |
| `acquire_available_container(timeout, lock_release_ms)` | ✅  | ✅  | ✅    | 1  | 悲观锁推荐：获取并自动加锁，上下文管理器自动释放 |

---

### 写入账号（生产者）

## LimitedModel 说明

`LimitedModel` 用于表示受限访问（rate-limited）的容器条目，每条数据包含唯一标识、时间戳、TTL、访问间隔和访问计数等信息。

### 字段说明

| 字段                    | 类型    | 默认值       | 描述                  |
|-----------------------|-------|-----------|---------------------|
| `id`                  | `str` | —         | 数据唯一标识符，必须提供        |
| `ct`                  | `int` | 当前时间戳（毫秒） | 记录条目的收集时间，无需设置      |
| `ttl_ms`              | `int` | `0`       | 生存时间（0 表示永不过期）      |
| `interval_ms`         | `int` | `0`       | 访问间隔（0 表示不限速）       |
| `next_time_available` | `int` | 当前时间戳（毫秒） | 下次可访问时间，用于限流控制，无需设置 |
| `containers`          | `str` | `None`    | 实际存储的容器数据（JSON字符串等） |
| `usage_count`         | `int` | `0`       | 已使用次数（系统自动递增）       |
| `locked_until`        | `int` | `0`       | 临时锁定截至时间（系统维护）      |

### 使用说明

- `ct` 和 `next_time_available` 都以毫秒为单位，便于高精度限流。
- `interval_ms > 0` 时，访问条目会更新 `next_time_available` 为当前时间 + 间隔。
- 每次访问条目时，都会通过 `usage_count` 记录被访问次数，以统计和调度。
- `ttl_ms` 可配合 Redis 等存储设置过期时间，实现自动清理。

```python
import random
from redis_xsync import register_redis
from redis_xsync.rtypes import LimitedModel

redis = register_redis(label="redis_xsync", host="127.0.0.1", port=6379)
Limited = redis.Limited


async def produce_container():
    async with Limited(redis_key="crawler:identity:flow_limit") as lt:
        for i in range(10):
            token = LimitedModel(
                id=f"user_{i}",
                ttl_ms=random.randint(30000, 60000),
                interval_ms=random.randint(1000, 6000),
                containers="cookies"
            )
            await lt.set_available_containers(token)
```

---

### 获取账号（消费者，阻塞等待）

```python
# 普通等待方式（不使用上下文管理器）
async def consume_token():
    async with Limited(redis_key="crawler:identity:flow_limit") as lt:
        # 等待可用 token，超时 10 秒
        token = await lt.wait_ask_available_containers(quantity=1, timeout=10)
        print("获取到账号:", token)


# 消费者（推荐写法：使用上下文管理器自动加锁释放）
async def consume():
    async with Limited(redis_key="crawler:identity:flow_limit") as lt:
        async with lt.acquire_available_container(timeout=8000, lock_release_ms=5000) as container:
            if container:
                print("获取到容器:", container["id"])
                # 使用 container["containers"] 中的数据进行请求
                # 退出 with 块时自动释放锁并更新 next_time_available
```

---

### 同步调用

```python
# 普通等待方式（不使用上下文管理器）
def consume_sync():
    with Limited(redis_key="crawler:identity:flow_limit") as lt:
        token = lt.wait_ask_available_containers(quantity=1, timeout=10)
        print("获取到账号:", token)


# 消费者（推荐写法：使用上下文管理器自动加锁释放）
with Limited(redis_key="crawler:identity:flow_limit") as lt:
    with lt.acquire_available_container(timeout=8000, lock_release_ms=5000) as container:
        print(container)
```

## ⚡ 说明：

- produce_container 写入 token 到 Redis，带过期时间和间隔限流
- consume_token 或 consume_sync 从 Redis 中获取可用 token，如果当前没有可用，会等待直到 timeout
- 支持异步和同步，且保证跨进程 / 跨机器原子性

---

## 🧠 设计说明

### 🔐 分布式锁

* 基于 Redis 原子操作
* 支持多 key（组合锁）
* 防止并发冲突

---

### 🚦 限流模型

* 每个 token：

    * 独立 TTL
    * 独立间隔（interval）
* Lua 保证原子性
* 支持多进程 / 多机器共享

---

## ⚠️ 注意事项

### Windows 建议

```python
import asyncio

asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
```

### 🧩 Redis 版本兼容性

RedisXsync 基于标准 Redis 协议（RESP）实现，理论上兼容大多数 Redis 版本（6.x / 7.x / 8.x）。
但在实际生产环境中，建议使用较新版本以获得更好的稳定性与性能。

#### ✅ 推荐版本

推荐版本： Redis ≥ 6.0
最佳体验： Redis 7.x 及以上
不推荐： Redis ≤ 5.x

#### 📌 原因说明

- 官方支持策略
  Redis 官方逐步停止对旧版本的维护（EOL），不再提供安全更新与问题修复。
- 功能与性能差异
  Redis 7.x / 8.x 在性能、命令能力以及 Lua 执行方面更加完善。
- Lua 脚本依赖（关键）
  RedisXsync 在分布式锁与限流中大量依赖 Lua 原子操作，
  旧版本 Redis 在脚本执行效率与行为一致性上可能存在问题。

#### ⚠️ 实践建议

```
✔ 推荐组合：
- Redis 7.x + RedisXsync   ← 最稳定

✔ 可用组合：
- Redis 6.x + RedisXsync   ← 兼容良好

❌ 不推荐：
- Redis 5.x 及以下        ← 可能存在兼容或性能问题
```

#### 📎 补充说明

- RedisXsync 不依赖 Redis Cluster，单机 Redis 即可运行
- 完整支持以下能力：
  1.分布式锁
  2.限流（基于 Lua）
  3.多实例 / 多 DB 管理

* 使用 acquire_available_container() 上下文管理器是最安全、推荐的方式，它会自动处理等待、加锁和释放。

👉 在生产环境中，建议统一使用较新版本 Redis，以避免潜在兼容问题并获得最佳性能。
高并发建议
---

## 📌 适用场景

* 爬虫账号 / Token / 代理池管理
* API 接口限流
* 分布式任务调度
* 多服务并发控制
* 需要精确毫秒级限流的场景

---

## 🤝 贡献

欢迎 Issue / PR！

---

## 📄 License

MIT License
