3

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

我的 tasks.py 的关键组件可以在这里看到，但在粘贴之前，我将展示我从 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 的递归和递归函数以及这如何影响链式任务中返回语句的逻辑并不完全清楚。

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

4

1 回答 1

1

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

我不知道 celery，但是在 Twisted 和许多其他异步框架中......True当任务被推迟到队列中时，对类似的东西的调用会立即返回（通常或者可能是一个可以跟踪状态的对象）。

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

你是：response立即定义为 async deferred task，但随后尝试对其采取行动，就好像结果已经进来一样

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

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

于 2013-02-22T21:35:57.837 回答