标签: task-queue

嗨,
我要设置一个rails-website,在一些初始用户输入之后,进行了一些繁重的计算(通过c-extension到ruby,将使用多线程).因为这些计算几乎消耗所有cpu-time(内存也是如此),所以一次运行的计算决不应该多于一个.我也不能使用(异步)后台作业(比如延迟作业),因为rails必须显示该计算的结果,并且该站点应该在没有javascript的情况下工作.
所以我想我需要一个单独的进程,其中所有rails实例必须排队他们的计算请求并等待答案(如果队列已满,可能是错误消息),一种同步作业管理器.

有没有人知道是否有这样的功能的宝石/插件？(nanite对我来说似乎很酷,但似乎只是异步,因此rails实例不知道计算何时完成.这是正确的吗？)
另一个想法是使用分布式ruby(drb)编写我自己的,但为什么发明如果车轮已经存在,那又一次转动？

任何帮助,将不胜感激!

编辑:由于zaius的提示,我想我将能够异步执行此操作,所以我将尝试resque.

有没有办法在执行任务时处理任何软截止日期？执行10分钟后抛出DeadlineExceededError,之后几秒钟我就会做一些事情.我想在任务终止之前清理一些事情并创建一个新任务.这可能需要几秒钟.有没有办法通过捕获大约9分钟的任何异常来做到这一点.我知道我可以在9分钟后手动抛出异常.但这可以由GAE自动完成吗？

class FillMtxHandler():

def post(self,index,user,seqlen):

    try :           
        FillMtx(index,user,seqlen)

    except DeadlineExceededError:

        deferred.defer(self.post,index,user,seqlen)

以上是我的代码.index是一个列表,从0开始.它将在FillMtx中递增.一旦超过截止日期,就会抛出错误,我想继续从索引最后增加的位置开始.我收到以下错误

   The API call taskqueue.BulkAdd() was explicitly cancelled.
Traceback (most recent call last):
  File "/base/python_runtime/python_lib/versions/1/google/appengine/ext/webapp/__init__.py", line 517, in __call__
    handler.post(*groups)
  File "/base/python_runtime/python_lib/versions/1/google/appengine/ext/deferred/deferred.py", line 258, in post
    run(self.request.body)
  File "/base/python_runtime/python_lib/versions/1/google/appengine/ext/deferred/deferred.py", line 124, in run
    return func(*args, **kwds)
  File "/base/python_runtime/python_lib/versions/1/google/appengine/ext/deferred/deferred.py", line 166, in invoke_member
    return getattr(obj, membername)(*args, **kwargs)
  File "/base/data/home/apps/0xxopdp/3.347813391084738922/fillmtx.py", line 204, in post
    deferred.defer(self.post,index,user,seqlen)
  File "/base/python_runtime/python_lib/versions/1/google/appengine/ext/deferred/deferred.py", line 241, in defer
    return task.add(queue, transactional=transactional)
  File "/base/python_runtime/python_lib/versions/1/google/appengine/api/taskqueue/taskqueue.py", line 688, in add
    return …

我每天都有一个cron工作来调用API并获取一些数据.对于数据的每一行,我启动一个任务队列来处理数据(这涉及通过其他API查找数据).一旦完成所有这些,我的数据在接下来的24小时内不会改变,所以我记得它.

有没有办法知道我排队的所有任务何时完成,以便我可以缓存数据？

目前我通过安排两个这样的cron作业以一种非常混乱的方式做到这一点:

class fetchdata(webapp.RequestHandler):
def get(self):
    todaykey = str(date.today())
    memcache.delete(todaykey)
    topsyurl = 'http://otter.topsy.com/search.json?q=site:open.spotify.com/album&window=d&perpage=20'
    f = urllib.urlopen(topsyurl)
    response = f.read()
    f.close()

    d = simplejson.loads(response)
    albums = d['response']['list']
    for album in albums:
        taskqueue.add(url='/spotifyapi/', params={'url':album['url'], 'score':album['score']})

class flushcache(webapp.RequestHandler):
    def get(self):
        todaykey = str(date.today())
        memcache.delete(todaykey)   

然后我的cron.yaml看起来像这样:

- description: gettopsy
  url: /fetchdata/
  schedule: every day 01:00
  timezone: Europe/London

- description: flushcache
  url: /flushcache/
  schedule: every day 01:05
  timezone: Europe/London

基本上 - 我猜测我的所有任务都不会花费超过5分钟的时间来运行,所以我只需要在5分钟后刷新缓存,这样可以确保数据缓存时完成.

有没有更好的编码方式？感觉像我的解决方案不是最好的....

谢谢汤姆

（在清楚描述情况时，我将尽量使这个问题简短。如果缺少任何内容，请发表评论。）

情况

• 我正在同一数据中心中运行具有三台服务器的集群
• 为了简化部署，每个服务器运行完全相同的应用程序代码

目标

• 要由一台服务器每分钟运行一个任务（称为任务X）。

在这些条件下

• 群集保持分布式并且高度可用
• 每个服务器保持运行相同的应用程序代码。换句话说，没有诸如“将代码A部署到主服务器并将代码B部署到所有辅助服务器之类的事情。

我不希望区分服务器类型的原因是为了保持高可用性（避免所谓的主服务器宕机），冗余（分布式负载），并避免在需要部署其他服务器的地方创建复杂的部署过程应用程序到不同类型的服务器。

为什么这么难？如果我要添加每5分钟执行一次此任务的代码，则每个服务器都将执行该代码，因为每个服务器都运行相同的应用程序代码。因此，他们需要能够协调每个滴答期间哪个服务器将运行相同的服务器。

我可以使用诸如Apache Kafka或的分布式消息传递机制Redis。如果使用这种机制来协调任务，那么这种“算法”将如何工作？

我向其他人提出了这个问题，他的回答是使用任务队列。但是，这似乎并不能解决问题，因为问题仍然存在：哪个服务器要将任务添加到任务队列中？如果所有服务器都将任务添加到队列中，则将导致重复条目。此外，哪个服务器将执行队列中的下一个任务？所有这些都需要通过集群内的协调来确定，而不必区分不同类型的服务器。

很多任务都无法执行,因为它们从目标后端获得503.这是非常奇怪的,因为大多数任务都会执行,但似乎如果它得到503一次它总会得到503.

我尝试手动运行任务并且后端处于活动状态并且正在服务但是如果任务在503之前失败,那么它将在现场使用503再次失败.

任何线索？

我正在使用node.js和mongodb构建一个Web应用程序.我需要添加延迟的工作.例如,在注册后一个月向用户发送电子邮件.我没有找到除Kue之外的延迟作业的现有解决方案,但它使用Redis,我更愿意在将另一个资源添加到我的Web应用程序之前使用我现有的Mongodb.

有没有现成的解决方案？

我需要检查添加到队列(Beanstalkd)的作业是否已在Laravel(Laravel 5)中完成,如果已完成,我需要返回更新的记录(任务添加到数据库中的队列更新记录).我加入了我的composer.json:

"pda/pheanstalk": "3.*"

我以这种方式添加作业队列:

$jobId = Queue::push('App\Class', $object->toArray(), $this->getQueueName());

我用以检查作业是否已完成使用以下功能:

public function find($queueName, $jobId, $recordId)
{
    $phean = Queue::getPheanstalk();

    try {
        $phean->peek($jobId);
        $data = ['status' => 'waiting'];
    } catch (ServerException $e) {
        $message = $e->getMessage();

        if ($message == 'NOT_FOUND: Job ' . $jobId . ' does not exist.') {
            $data = ... // here I get from database data for $recordId

        } else {
            $data = ['status' => 'error'];
        }
    }

    return $data;
}

问题是 - 是否可靠的方法来检查工作是否已完成？我在这里比较我从异常中得到的消息.我还没有找到任何其他方法来检查工作是否已完成.

我在queue.yaml中定义了一个任务队列,如下所示:

- name: queue-name
  mode: push
  max_concurrent_requests: 10
  rate: 10/h
  retry_parameters:
    min_backoff_seconds: 5
    max_backoff_seconds: 120
    task_retry_limit: 10
    task_age_limit: 1d

当我在部署后在开发人员控制台中查看它时,我看到一条错误消息:

This queue has been disabled, and removed from queue.yaml / queue.xml

我不确定问题是什么.如果我将速率更改为10/s,则错误消失.有任何想法吗？

我们有一个cron作业,它在后端模块上每小时运行一次并创建任务.cron作业在Cloud SQL数据库上运行查询,并且任务对其他服务器进行HTTP调用并更新数据库.通常它们运行得很好,即使创建了数千个任务,但有时它会"卡住",并且日志中没有任何内容可以解释这种情况.例如,昨天我们监控了cron作业,同时它创建了几十个任务,然后停止了,还有8个任务也被卡在了队列中.当很明显没有发生任何事情时,我们再多次运行该过程并且每次都成功完成.

一天之后,原始任务被DeadlineExceededException杀死,然后显然在同一实例中运行的其他8个任务被以下消息杀死:处理此请求的进程遇到问题,导致它退出.这可能会导致新进程用于您的应用程序的下一个请求.如果经常看到此消息,则可能会在应用程序初始化期间抛出异常.(错误代码104)

在进程被杀死之前,我们在日志中看不到它们的记录,现在我们看到它们在DeadlineExceededException之前没有日志记录,所以我们不知道它们在什么时候被卡住了.我们怀疑数据库中存在一些锁定,但我们在以下链接中看到查询有10分钟的限制,因此这会导致进程在一天之内发生故障:https://cloud.google. COM/AppEngine上/文档/ JAVA /云的SQL /#Java_Size_and_access_limits

我们模块的类和扩展配置是:

<instance-class>B4</instance-class>
<basic-scaling>
    <max-instances>11</max-instances>
    <idle-timeout>10m</idle-timeout>
</basic-scaling>

队列的配置是:

<rate>5/s</rate>
<max-concurrent-requests>100</max-concurrent-requests>
<mode>push</mode>
<retry-parameters>
    <task-retry-limit>5</task-retry-limit>
    <min-backoff-seconds>10</min-backoff-seconds>
    <max-backoff-seconds>200</max-backoff-seconds>
</retry-parameters>

我上传了一些cron作业的跟踪数据图片:http: //imgur.com/a/H5wGG.这包括跟踪摘要以及时间轴的开始/结束.8个已终止的任务没有跟踪数据.

可能是什么原因以及我们如何进一步调查？

我正在为Laravel移动应用程序开发API.

方法将向其他API发出请求,组合和过滤数据,更改其结构等.

应用程序的要求之一是响应时间不超过30秒,或者根本不响应.所以,我必须尽可能多地重复请求.我试图通过Laravel Queues实现这一点,并且目前在我的Job类中有类似的东西:

private $apiActionName;

public function __construct($apiActionName)
{
    $this->apiActionName = $apiActionName;
}

public function handle(SomeService $someService)
{
    return $someService->{$this->apiActionName}();
}

而这个动作代码在控制器中:

public function someAction()
{ 
    $data = $this->dispatch(new MyJob($apiActionName));
    return response()->json($data);
}

是的,我知道从工作中返回价值是不错的主意,但我希望这是可能的.但是$ this-> dispatch()只返回排队的作业ID,而不是handle方法的结果.

TL; DR:如何从排队的作业中返回数据,而不将其保存在任何地方,即使它在队列中有多次尝试？如果乔布斯不适合这个,也许有人知道其他方法.任何建议将被认真考虑.

提前致谢!