一、TaskTiger 库概述
TaskTiger 是基于 Redis 开发的 Python 分布式任务队列库，专注于异步任务处理、定时任务与重试机制，采用生产者-消费者模型，通过 Redis 存储任务与状态。优点为轻量易用、依赖少、支持任务重试/定时/优先级，缺点是功能不如 Celery 丰富，生态较小。基于 MIT License 开源，可自由商用与修改。
二、TaskTiger 安装与环境准备
在使用 TaskTiger 之前，必须先完成环境配置，它强依赖 Redis，所以我们要先安装 Redis 服务，再安装 TaskTiger 本身。
2.1 安装 Redis
TaskTiger 使用 Redis 作为消息代理和任务存储后端，因此需要先安装并启动 Redis：
- Linux：可通过 apt、yum 等包管理器安装
- macOS：使用 Homebrew 安装
brew install redis - Windows：可下载 Redis 安装包或使用 WSL 安装
安装完成后，启动 Redis 服务：
redis-server默认情况下，Redis 会在本地的 6379 端口运行，这也是 TaskTiger 默认使用的地址。
2.2 安装 TaskTiger
使用 pip 即可直接安装最新版本的 TaskTiger，命令如下：
pip install tasktiger安装完成后，我们可以在 Python 环境中导入 tasktiger 来验证是否安装成功：
import tasktiger
print(tasktiger.__version__)如果能够正常输出版本号，就说明 TaskTiger 已经成功安装，可以开始使用。
2.3 基础连接配置
TaskTiger 默认连接本地的 Redis（localhost:6379），如果你的 Redis 有密码、端口不同，或者是远程 Redis 服务，可以在初始化时进行配置：
from tasktiger import TaskTiger
import redis
# 自定义 Redis 连接
redis_client = redis.Redis(
host='localhost',
port=6379,
password='your_redis_password', # 有密码则填写
db=0
)
# 初始化 TaskTiger
tiger = TaskTiger(connection=redis_client)这是最基础的配置方式，绝大多数场景下，使用默认连接就足够使用。
三、TaskTiger 核心使用方式与基础示例
TaskTiger 的核心使用流程非常清晰：定义任务 → 提交任务 → 启动 worker 执行任务，我们从最简单的异步任务开始演示。
3.1 定义并执行第一个异步任务
我们先创建一个 Python 文件，命名为 tasks.py，在里面编写一个简单的任务函数，并用 TaskTiger 装饰器将其注册为任务。
tasks.py
from tasktiger import TaskTiger
# 初始化 TaskTiger 实例
tiger = TaskTiger()
# 使用装饰器注册任务
@tiger.task
def add(a, b):
"""简单的加法任务，模拟基础计算任务"""
result = a + b
print(f"任务执行：{a} + {b} = {result}")
return result上面代码中，@tiger.task 装饰器是核心，它会把普通函数变成可以被 TaskTiger 调度的异步任务。
接下来，我们再创建一个文件 producer.py，用来提交任务到队列：
producer.py
from tasks import add
# 异步提交任务，不会阻塞当前程序
add.delay(10, 20)
add.delay(30, 40)
print("任务已提交到队列，等待 worker 执行")这里的 delay() 方法是 TaskTiger 任务对象的核心方法，作用是将函数调用封装成任务，发送到 Redis 队列中，而不是立即执行。
现在我们有了任务定义和任务提交代码，还需要启动 worker 进程来消费并执行队列中的任务。
打开命令行，进入项目目录，执行以下命令：
tasktiger -w tasks参数 -w 后面跟着的是任务所在的模块名，也就是我们的 tasks.py，去掉 .py 后缀。
执行后，worker 会持续运行，监听 Redis 队列，一旦有新任务就会立即执行，你会在控制台看到：
任务执行：10 + 20 = 30
任务执行：30 + 40 = 70这就是 TaskTiger 最基础的异步任务流程，适用于发送邮件、生成日志、简单计算等不需要立即返回结果的场景。
3.2 带参数与关键字参数的任务
TaskTiger 支持位置参数、关键字参数，使用方式和普通函数完全一致，非常贴合 Python 原生语法。
我们在 tasks.py 中新增一个任务：
@tiger.task
def send_message(user_id, content, priority="normal"):
"""模拟发送消息任务"""
print(f"向用户 {user_id} 发送消息：{content}，优先级：{priority}")在 producer.py 中提交任务：
from tasks import send_message
# 位置参数
send_message.delay(1001, "你好，欢迎使用 TaskTiger")
# 关键字参数
send_message.delay(1002, "异步任务执行成功", priority="high")启动 worker 后，任务会按照提交顺序依次执行，参数传递完全兼容普通函数的用法，对开发者非常友好，不需要额外学习复杂的参数规则。
四、TaskTiger 高级功能使用
TaskTiger 不止能执行基础异步任务，还提供了任务重试、定时任务、任务唯一、任务优先级、异常处理等高级功能，满足生产环境需求。
4.1 任务失败自动重试
在实际项目中，任务可能因为网络波动、第三方接口异常而执行失败，TaskTiger 支持自动重试机制，可以指定重试次数、重试间隔。
我们在 tasks.py 中添加一个会随机失败的任务，模拟接口调用失败：
import random
@tiger.task(retry=3, retry_delay=5)
def request_api():
"""模拟请求第三方API，随机失败"""
if random.choice([True, False]):
raise Exception("API 请求失败")
print("API 请求成功")参数说明：
retry=3：任务失败后最多重试 3 次retry_delay=5：每次重试间隔 5 秒
当任务抛出异常时，TaskTiger 会自动将任务重新放入队列，等待指定时间后再次执行，直到执行成功或达到最大重试次数。
4.2 定时任务与延时任务
很多场景需要任务在指定时间后执行，或者每天、每小时定时执行，TaskTiger 提供了便捷的延时执行方法。
4.2.1 固定延时执行任务
在提交任务时使用 execute_after() 方法，可以指定任务在多少秒后执行：
from tasks import add
from datetime import timedelta
# 10 秒后执行
add.execute_after(timedelta(seconds=10), 5, 8)
# 也可以直接传入秒数
add.execute_after(10, 5, 8)这个功能非常适合用于订单超时未支付取消、延时发送通知等场景。
4.2.2 指定时间点执行任务
除了延时执行，还可以指定具体的时间点执行任务：
from datetime import datetime
# 在 2026-01-01 00:00:00 执行
run_time = datetime(2026, 1, 1, 0, 0, 0)
add.execute_at(run_time, 1, 1)这种方式适合节日定时推送、定时报表生成等场景。
4.3 唯一任务：避免重复提交
在高并发场景下，可能会出现重复提交相同任务的情况，比如重复发送短信、重复创建订单，TaskTiger 可以通过 unique 参数设置任务唯一。
@tiger.task(unique=True)
def send_sms(phone, code):
"""给同一个手机号发送验证码，防止重复发送"""
print(f"向 {phone} 发送验证码：{code}")当 unique=True 时，TaskTiger 会根据函数名和参数生成唯一标识，在任务执行完成前，相同参数的任务不会重复加入队列，从根源上避免重复执行。
4.4 任务优先级控制
当队列中有大量任务时，我们希望重要任务优先执行，比如支付、通知任务，TaskTiger 支持设置任务优先级。
# 定义高优先级任务
@tiger.task(priority=10)
def high_priority_task():
print("高优先级任务执行")
# 定义低优先级任务
@tiger.task(priority=1)
def low_priority_task():
print("低优先级任务执行")数字越大，优先级越高，worker 会优先从高优先级队列中取任务执行。
4.5 任务状态与结果查看
TaskTiger 会在 Redis 中记录任务的状态，包括等待、执行中、成功、失败、重试等，我们可以在代码中查看任务状态。
from tasks import add
# 提交任务并获取任务对象
task = add.delay(10, 20)
# 查看任务 ID
print("任务 ID:", task.id)
# 查看任务状态
print("任务状态:", task.status)
# 查看任务结果（执行完成后才有）
print("任务结果:", task.result)常见任务状态：
queued：已进入队列，等待执行active：正在执行done：执行成功failed：执行失败retrying：等待重试
五、生产环境常用配置与多 worker 部署
在实际生产项目中，单 worker 无法满足高并发需求，TaskTiger 支持启动多个 worker 进程，同时监听队列，提升任务处理效率。
5.1 启动多个 worker
命令行中可以使用 -w 参数指定 worker 数量，例如启动 4 个 worker：
tasktiger -w tasks -n 4参数 -n 4 表示启动 4 个 worker 进程，并行处理任务，充分利用服务器 CPU 资源。
5.2 指定队列执行
TaskTiger 支持多队列隔离，不同类型的任务放入不同队列，避免任务互相干扰。
定义任务时指定队列：
@tiger.task(queue="email")
def send_email(email):
print(f"发送邮件到 {email}")
@tiger.task(queue="sms")
def send_sms(phone):
print(f"发送短信到 {phone}")启动 worker 时只监听指定队列：
# 只处理 email 队列任务
tasktiger -w tasks -q email
# 只处理 sms 队列任务
tasktiger -w tasks -q sms这种方式适合大型项目，按业务模块拆分任务队列，方便维护和扩容。
5.3 后台运行 worker
生产环境中，我们需要让 worker 在后台持续运行，可以使用 nohup 或 supervisor 托管进程。
使用 nohup 后台启动：
nohup tasktiger -w tasks -n 4 > tasktiger.log 2>&1 &日志会输出到 tasktiger.log 文件，方便排查问题。
六、实际项目综合案例：用户注册异步流程
我们模拟一个完整的用户注册业务流程，用户注册成功后，异步发送欢迎邮件、记录日志、统计注册数量，使用 TaskTiger 实现全流程解耦。
项目结构
project/
├── tasks.py # 任务定义
├── user_service.py # 用户注册逻辑
└── run_worker.sh # worker 启动脚本6.1 编写任务文件 tasks.py
from tasktiger import TaskTiger
import time
tiger = TaskTiger()
# 发送欢迎邮件
@tiger.task(queue="email", retry=2, retry_delay=3)
def send_welcome_email(email):
time.sleep(1) # 模拟发送邮件耗时
print(f"【邮件】已向 {email} 发送欢迎邮件")
# 记录用户注册日志
@tiger.task(queue="log")
def write_register_log(user_id, username):
log_info = f"用户 {user_id} - {username} 注册成功"
print(f"【日志】{log_info}")
# 实际项目可写入数据库或文件
# 统计注册人数
@tiger.task(queue="stat")
def update_register_count():
print("【统计】更新系统注册人数 +1")6.2 编写用户注册服务 user_service.py
from tasks import send_welcome_email, write_register_log, update_register_count
def register_user(username, email):
"""模拟用户注册"""
# 1. 模拟数据库保存用户信息
user_id = 10001
print(f"用户 {username} 注册成功，用户ID：{user_id}")
# 2. 提交异步任务，不阻塞注册流程
send_welcome_email.delay(email)
write_register_log.delay(user_id, username)
update_register_count.delay()
return {"user_id": user_id, "username": username}
# 模拟用户注册
if __name__ == "__main__":
result = register_user("test_user", "[email protected]")
print("注册接口返回结果：", result)6.3 启动 worker 并执行
- 启动 TaskTiger worker：
tasktiger -w tasks -n 3- 运行用户注册脚本：
python user_service.py控制台输出：
用户 test_user 注册成功，用户ID：10001
注册接口返回结果： {'user_id': 10001, 'username': 'test_user'}
【邮件】已向 [email protected] 发送欢迎邮件
【日志】用户 10001 - test_user 注册成功
【统计】更新系统注册人数 +1可以看到，用户注册接口快速返回结果，后续的邮件、日志、统计操作都由 TaskTiger 异步执行，极大提升了接口响应速度。
七、相关资源
- Pypi 地址：https://pypi.org/project/tasktiger/
- Github 地址：https://github.com/closeio/tasktiger
- 官方文档地址：https://tasktiger.readthedocs.io/
关注我，每天分享一个实用的Python自动化工具。
