在python代码中实时查看celery队列不太方便，只能通过管理工具来查看(目前暂未找到其他较好方案，如有请留言告诉我)，本文介绍一种利用redis的list结构来模拟任务队列状态的方案。

首先我们学习下redis的基本操作。

python中redis基本操作

 # 创建连接，默认数据库参数
 r = redis.Redis()

 # 可以使用连接池
 pool = redis.ConnectionPool()
 r = redis.Redis(connection_pool=pool)
 r.set('name', 'my name') # 添加
 print(r.get('name')) # b'my name'

 # 批量操作
 r.mset({"key1": 'value1', "key2": 'value2'})
 print(r.mget(["key1", "key2"])) # [b'value1', b'value2']

 # String操作
 r.set('t', '测试')
 print((r.get('t').decode('utf-8'))) # 测试

 # hash测试
 dic = {"a1": "aa", "b1": "bb"}
 r.hmset("dic_name", dic)
 print(r.hmget("dic_name", ["a1", "b1"])) # [b'aa', b'bb']
 print(r.hget("dic_name", "b1")) # b'bb'

 # list测试
 # 在name对应的list中添加元素，每个新的元素都添加到列表的最左边
 r.lpush("list_name", 2)
 r.lpush("list_name", 3, 4, 5) # 保存在列表中的顺序为5，4，3，2
 # name对应的list元素的个数
 print(r.llen("list_name")) # 4
 # 对list中的某一个索引位置重新赋值
 r.lset("list_name", 0, "bbb") # b'bbb'
 # 移除列表的左侧第一个元素，返回值则是第一个元素
 print(r.lpop("list_name"))
 # 根据索引获取列表内元素
 print(r.lindex("list_name", 1)) # b'3'
 # 删除name对应的list中的指定值
 r.lrem("list_name", value=3, count=0)

 # set测试
 # 给name对应的集合中添加元素,元素不重复
 r.sadd("set_name", "aa")
 r.sadd("set_name", "aa", "bb")

 # 设置过期
 r.setex(name='a', value='b', time=10) # 设置过期时间（秒）
 print(r.get('a')) # b'b'
 time.sleep(10)
 print(r.get('a')) # None, 已经过期了

利用list结构模拟监控celery任务队列

1. 定义异步任务
tasks.py 中定义任务

@app.task
def task_list_test():
 pool = redis.ConnectionPool()
 r = redis.Redis(connection_pool=pool)
 # 将本任务添加到队列，任务名为task_name_001,状态为false
 r.rpush('task_name_list', "task_name_001")
 r.set('task_name_001', "false")

 # 这里设置任务为访问网络
 print(requests.request('GET', 'http://www.baidu.com').content)
 time.sleep(10)

 # 设置任务状态为完成
 r.set('task_name_001', 'true')

 print('task finish')

在这个例子中，我们在redis中设置任务队列名为task_name_list，并添加任务task_name_001，状态为"false"，执行网络访问操作，结束后，设置任务状态为"true"。

2. 执行任务，并查看运行状态
在另外一个文件worker.py中执行任务

 # 异步执行任务
 task_list_test.delay()
 print(r.get('task_name_001')) # None
 print(r.lrange("task_name_list", 0, -1)) # []
 time.sleep(20)
 print(r.get('task_name_001')) # 'true'
 print(r.lrange("task_name_list", 0, -1)) # [b'task_name_001']

我们首先执行异步任务，可以看到异步任务未执行时，任务队列为空，等待20s后，再次查看队列，发现有任务队列task_name_list，且其中任务名task_name_001执行状态为true，表明任务已经完成。