带 celery 的 django-socketio : send to socket after async task completes in separate process

Question

如何在我的主要 Django 应用程序进程中访问 Celery 任务的结果？或者，如何从单独的进程发布到现有套接字连接？

我有一个用户可以收到分数的应用程序。当记录分数时，会进行计算(目标进展等)，并根据这些计算向感兴趣的用户发送通知。计算可能需要 30 秒以上，因此为了避免 UI 缓慢，这些操作通过 Celery 任务在后台进程中执行，由我的 Score 模型的 post_save 信号调用。

理想情况下，我的 Nofication 模型上的 post_save 信号会向订阅的客户端发布消息(我正在使用 django-socketio，gevent-socketio 的包装器)。这看起来很简单......

1. 创建乐谱
2. 在后台进程中对新的 Score 实例进行一些计算
3. 根据这些计算，创建通知
4. 在通知保存时，抓取实例并通过套接字连接发布到订阅的客户端

但是在尝试了以下之后我不确定这是可能的:

• 将 gevent 的 SocketIOServer 实例传递给任务调用的回调方法，但这需要 pickle 传递的对象，这是不可能的

• 将套接字的 session_id(不同于 Django 的 session_id)存储在 memchache 中，并在 Celery 任务进程中检索它。

• 使用 Redis pubsub，因此在后台进程中创建的模型上由 post_save 信号调用的方法可以简单地发布到 Redis channel ，但在主应用程序进程(可以访问套接字连接)中监听聊天 channel 会阻塞应用程序的其余部分。

• 我还尝试为每个 Redis 客户端生成新线程，这些线程是为每个套接字订阅者创建的。据我所知，这需要产生一个新的 gevent.greenlets.Greenlet，并且 gevent 不能在多线程中使用

当然，这是一个已解决的问题。我错过了什么？

Answer 1

你已经有了django-socketio，用redis写一个pub/sub会很可惜:)

客户端:

var socket = new io.Socket();
socket.connect();
socket.on('connect', function() {
    socket.subscribe({{ score_update_channel }});
});

服务器端:

from django_socketio import broadcast_channel
def user_score_update(user):
    return 'score_updates_user_%s' % user.pk

channel = user_score_update(user)
broadcast_channel(score_result_data, channel)

需要在django-socketio进程上运行广播；如果您从不同的进程(celery worker)运行它，它将无法工作( channel 在内存中被 django-socketio 进程引用)；您可以通过将其包装在 View 中来解决此问题，当任务完成时 celery 将调用(发出真正的 http 请求)。