5

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

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

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

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

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

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

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

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

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

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

4

1 回答 1

0

你已经有了 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 进程上运行广播;如果您从不同的进程(芹菜工作者)运行它,它将无法工作(通道由 django-socketio 进程在内存中引用);您可以通过将其包装在视图中来解决此问题,并且当任务完成时 celery 将调用(发出真正的 http 请求)。

于 2013-04-03T06:42:34.447 回答