Django，带有递归和 Twitter API 的 Celery

Question

我正在使用 Django 1.4 和 Celery 3.0 (rabbitmq) 构建任务组合，用于获取和缓存 Twitter API 1.1 的查询。我正在尝试实现的一件事是任务链，其中最后一个基于迄今为止的响应和最近检索到的响应中的响应数据，对两个节点的任务进行递归调用。具体来说，这允许应用程序遍历用户时间线（最多 3200 条推文），考虑到任何给定的请求最多只能产生 200 条推文（Twitter API 的限制）。

可以看到我的 tasks.py 的关键组件here，但在粘贴之前，我将展示我从我的 Python shell 调用的链（但最终将通过最终 Web 应用程序中的用户输入启动)。给定：

>>request(twitter_user_id='#1010101010101#, 
  total_requested=1000, 
  max_id = random.getrandbits(128) #e.g. arbitrarily large number)

我打电话：

>> res = (twitter_getter.s(request) | 
        pre_get_tweets_for_user_id.s() | 
        get_tweets_for_user_id.s() | 
        timeline_recursor.s()).apply_async()

关键是timeline_recursor 可以启动可变数量的get_tweets_for_user_id 子任务。当timeline_recursor 处于其基本情况时，它应该返回一个响应字典，定义如下：

@task(rate_limit=None)
def timeline_recursor(request):
    previous_tweets=request.get('previous_tweets', None) #If it's the first time through, this will be None
    if not previous_tweets:
        previous_tweets = [] #so we initiate to empty array
    tweets = request.get('tweets', None) 

    twitter_user_id=request['twitter_user_id']
    previous_max_id=request['previous_max_id']
    total_requested=request['total_requested']
    pulled_in=request['pulled_in']

    remaining_requested = total_requested - pulled_in
    if previous_max_id:
        remaining_requested += 1 #this is because cursored results will always have one overlapping id

    else:
        previous_max_id = random.getrandbits(128) # for first time through loop

    new_max_id = min([tweet['id'] for tweet in tweets])
    test = lambda x, y: x<y

    if remaining_requested < 0:  #because we overshoot by requesting batches of 200
        remaining_requested = 0

    if tweets:
        previous_tweets.extend(tweets)

    if tweets and remaining_requested and (pulled_in > 1) and test(new_max_id, previous_max_id):

        request = dict(user_pk=user_pk,
                    twitter_user_id=twitter_user_id,
                    max_id = new_max_id,
                    total_requested = remaining_requested,
                    tweets=previous_tweets)

        #problem happens in this part of the logic???

        response = (twitter_getter_config.s(request) | get_tweets_for_user_id.s() | timeline_recursor.s()).apply_async()

    else: #if in base case, combine all tweets pulled in thus far and send back as "tweets" -- to be 
          #saved in db or otherwise consumed
        response = dict(
                    twitter_user_id=twitter_user_id,
                    total_requested = total_requested,
                    tweets=previous_tweets)
    return response

因此，我对 res.result 的预期响应是一个字典，其中包含 twitter 用户 ID、请求的推文数量以及在连续调用中提取的推文集。 然而，在递归任务领域，一切都不是很好。当我运行上面确定的链时，如果我在启动链后立即输入 res.status，它表示“成功”，即使在我的 celery worker 的日志视图中，我也可以看到链式递归调用到 twitter api 正在按预期使用正确的参数进行制作。即使正在执行链式任务，我也可以立即运行 result.result。 res.result 产生一个 AsyncResponse 实例 id。即使在递归链式任务完成运行之后， res.result 仍然是 AsyncResult id。

另一方面，我可以通过访问 res.result.result.result.result['tweets'] 访问我的完整推文集。我可以推断出每个连锁的连锁子任务确实正在发生，我只是不明白为什么 res.result 没有预期的结果。当 timeline_recursor 获得其基本情况时应该发生的递归返回似乎没有按预期传播。

对可以做什么有什么想法吗？ Celery 中的递归可以变得非常强大，但至少对我来说，我们应该如何考虑使用 Celery 的递归和递归函数以及这如何影响链式任务中的 return 语句的逻辑并不完全清楚。

很高兴根据需要澄清，并提前感谢您的任何建议。

Answer 1

apply_async 返回什么（如对象类型）？

我不知道 celery，但在 Twisted 和许多其他异步框架中......当任务被推迟到队列。

再次，不知道 celery ，我猜这正在发生：

你是：立即将 response 定义为异步延迟 task，但随后尝试对其采取行动，就好像结果已经到来一样

您想成为：定义一个callback 例程以在任务完成后在结果上运行并返回一个值

查看 celery 文档，apply_async 通过 link 接受回调 - 我找不到任何人试图从中捕获返回值的示例。