使用 Celey+RabbitMQ+Redis 实现订单管理服务

Question

我已经使用zipstream 实现了将一些文件即时打包到.zip 文件中的功能，因此响应格式为mimetype='application/zip' 使用Flask。现在我想实现提交订单的服务（即生成.zip文件的任务），跟踪订单状态（例如STARTED，SUCCESS，FAILURE），撤销订单（即取消生成@987654330的任务） @文件或删除生成的.zip文件），并为当前用户下载订单（即.zip文件）。我计划使用 Celery，RabbitMQ 作为消息代理，Redis 作为结果后端。

问题来了。仅 Redis 作为结果后端就足够了吗？因为用户订单的跟踪状态似乎涉及user = ... 之类的查询，而不是通过task id

支持AsyncResult 查询

更新： 感谢@Tomáš Linhart。我使用 celery 的signal handler mechanism 遵循将结果存储在 MongoDB 中的扩展方式。这是与我创建的任务相关的代码的 sn-p。我创建task_tracker 和celery_cli 实例的方式与@Tomáš Linhart 的answer 中的tracker 和app 相同。

# Import task_tracker and celery_cli here
...

@task_tracker.track
@celery_cli.task(name='pack-up-tif')
def async_pack_up_tif(**kwargs):
    # Some processing here
    ...

def pack_up_tif(msg):
    result = async_pack_up_tif.delay(**msg)
    return result

但我还有一个问题。调用delay方法时如何拦截任务ID？因为当task_success 信号被触发时，我需要通过find_one_and_update 将该信息存储在MongoDB 集合中。

Answer 1

详细说明我发布的评论并回答您更新的问题。确定从哪里收集您想在每个信号中跟踪的各种数据有点复杂。一些信号为您提供任务本身，一些信号为您提供任务请求。

我以 task_success 信号的处理程序结束：

def _on_task_success(self, sender, result, **other_kwargs):
    if sender.name not in self.tasks:
        return

    collection = self.mongo \
        .get_database(self.config['mongodb']['database']) \
        .get_collection(self.config['mongodb']['collection'])
    collection.find_one_and_update(
        {'_id': sender.request.id},
        {
            '$setOnInsert': {
                'name': sender.name,
                'args': sender.request.args,
                'kwargs': sender.request.kwargs
            },
            '$set': {
                'status': states.SUCCESS,
                'date_done': datetime.datetime.utcnow(),
                'retries': sender.request.retries,
                'group_id': sender.request.group,
                'chord_id': sender.request.chord,
                'root_id': sender.request.root_id,
                'parent_id': sender.request.parent_id,
                'result': result
            },
            '$push': {
                'status_history': {
                    'date': datetime.datetime.utcnow(),
                    'status': states.SUCCESS
                }
            }
        },
        upsert=True,
        return_document=ReturnDocument.AFTER)