与Celery的django-socketio:异步任务在单独的进程中完成后发送到套接字

Question

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

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

理想情况下,我的Nofication模型上的post_save信号会向订阅的客户端发布一条消息(我正在使用django-socketio,这是gevent-socketio的包装器).这似乎很简单......

1. 创建一个分数
2. 在后台进程中对新的Score实例进行一些计算
3. 根据这些计算,创建通知
4. 在Notification save上,获取实例并通过套接字连接发布到订阅的客户端

但是在尝试以下后我不确定这是否可行:

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

• 在memchache中存储套接字的session_id(与Django的session_id不同),并在Celery任务进程中检索它.

• 使用Redis pubsub,因此在后台进程中创建的模型上通过post_save信号调用的方法可以简单地发布到Redis通道,但是在主应用程序进程(可以访问套接字连接)中侦听聊天通道会阻止应用程序的其余部分.

• 我还尝试为每个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）运行它，它将无法工作（通道由 django-socketio 进程在内存中引用）；您可以通过将其包装在视图中来解决此问题，并且当任务完成时，celery 将调用（发出真正的 http 请求）。