标签: task-queue

我一直在学习RabbitMQ各种拓扑,但是,我找不到任何对生产者发出的动态队列创建(也就是Declare Queue)的引用.想法是根据特定事件(例如HTTP请求)动态创建队列.队列将是临时的,具有TTL集并以事件ID命名.然后,消费者可以订阅主题"event.*"并合并与其相关的所有消息.

例:

1. 收到HTTP POST"创建用户"
2. producer创建一个队列user.ID
3. 在队列中推送有关用户的所有后续消息(例如"添加用户名","添加电子邮件"......)
4. worker被分配到随机队列"user.*"并将所有内容合并到一个用户帐户中
5. TTL过期后会自动删除队列

现在,RabbitMQ的这种情况是否可行？

考虑一个PHP Web应用程序,其目的是接受用户启动通用异步作业的请求,然后创建一个工作进程/线程来运行该作业.作业不是特别占用CPU或内存,但预计会经常阻塞I/O调用.每秒应该启动不超过一个或两个作业,但由于运行时间长,可能会同时运行许多作业.

因此,工作并行运行至关重要.此外,每个作业都必须由负责杀死挂起的工作人员,根据用户请求中止工作人员等的经理守护程序进行监控.

实现这样的系统的最佳方法是什么？我可以看到:

1. 从经理那里分配一名工人 - 这似乎是最低级别的选择,我必须自己实施一个监控系统.Apache是​​Web服务器,因此看起来该选项需要通过FastCGI启动任何PHP工作程序.
2. 使用某种作业/消息队列.(gearman,beanstalkd,RabbitMQ等) - 最初,这似乎是明显的选择.经过一些研究,我对所有选项感到困惑.例如,Gearman看起来像是为庞大的分布式系统而设计的,那里有一个固定的工人池......所以我不知道它是否适合我需要的东西(每个工作一个工人).

我目前正在设计一个Web应用程序,允许用户安排将针对HTTP API(代表他们)执行的任务.任务可以重复进行,可用于安排的最小时间分辨率将是一分钟.由于任务的性质,我认为以异步方式执行它们是有意义的.但是,这部分的架构应该如何？

我考虑过使用任务队列来创建Web应用程序的任务,并让它们由工作人员执行.在这种情况下,我有几个问题:

• 如何处理重复性任务？
• 如何轻松保存任务结果？
• 是否可以轻松地使队列"持久"？
• 工人应该直接与数据库交互吗？
• 我应该手动排队重复执行的任务吗？

还有什么可以考虑的？既然我认为我不是唯一考虑过这种Web应用程序架构的人,那么有没有"最佳实践"？是一个任务队列的方式去？

我将几个任务放入任务队列,并想知道特定任务何时完成.我没有在API中找到关于回叫或检查任务状态的任何内容,所以我想我会看到其他人做了什么,或者是否有一种解决方法(或官方).我不关心个人任务,如果它有帮助,我将6个不同的任务放进去,并想知道所有6个完成的时间.

谢谢!

我正在开发一个Perl项目,该项目需要一个FIFO消息队列,用于在一台机器上的多个进程之间分配任务(UNIX).队列大小可能会增加到1M个作业.

我已经尝试过了IPC::DirQueue,但是当大约5万个工作入队时,它变得非常缓慢.这个模块的哪些好的替代品可以在Perl中使用？

问题在哪里:

我们使用RabbitMQ作为任务队列.其中一项具体任务 - 向Vkontakte社交网络发送通知.他们api限制每秒请求和此限制根据您的应用程序大小.只需3个电话就可以获得少于10万人的应用程序,依此类推.所以我们需要人为地限制对他们的服务请求.现在这个逻辑基于应用程序.这很简单,虽然每个队列只能使用一个工作人员,但只需设置睡眠(300毫秒)并保持冷静.但是当你应该使用N个工作者时,这种同步变得微不足道.

如何使用RabbitMQ限制吞吐量？

基于上面的故事.如果可能的话,设置预取大小不仅基于消息,而且基于此逻辑的时间可以非常简单.例如,"每次获取的qos到1条消息不会快于1次,以秒为单位"等等.

1. 有这样的事吗？
2. 可能是关于此的其他策略？

我正在使用Laravel 5创建一个网站.我正在使用带有2个管的beanstalkd队列.我有10个由daemontools监控的工人.5名工人加工A管和B管.另外5名工人只加工B管.

事情大部分时间都很有效,但偶尔管B似乎锁定了.日志显示工人正在抛出异常.

[2015-04-24 07:09:36] local.ERROR: exception 'Pheanstalk\Exception\ServerException' with message 'Cannot delete job 273812: NOT_FOUND' in /x/website/vendor/pda/pheanstalk/src/Command/DeleteCommand.php:44
Stack trace:
#0 /x/website/vendor/pda/pheanstalk/src/Connection.php(121): Pheanstalk\Command\DeleteCommand->parseResponse('NOT_FOUND', NULL)
#1 /x/website/vendor/pda/pheanstalk/src/Pheanstalk.php(384): Pheanstalk\Connection->dispatchCommand(Object(Pheanstalk\Command\DeleteCommand))
#2 /x/website/vendor/pda/pheanstalk/src/Pheanstalk.php(67): Pheanstalk\Pheanstalk->_dispatch(Object(Pheanstalk\Command\DeleteCommand))
#3 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Jobs/BeanstalkdJob.php(73): Pheanstalk\Pheanstalk->delete(Object(Pheanstalk\Job))
#4 /x/website/vendor/laravel/framework/src/Illuminate/Queue/CallQueuedHandler.php(46): Illuminate\Queue\Jobs\BeanstalkdJob->delete()
#5 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Jobs/Job.php(126): Illuminate\Queue\CallQueuedHandler->call(Object(Illuminate\Queue\Jobs\BeanstalkdJob), Array)
#6 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Jobs/BeanstalkdJob.php(51): Illuminate\Queue\Jobs\Job->resolveAndFire(Array)
#7 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Worker.php(207): Illuminate\Queue\Jobs\BeanstalkdJob->fire()
#8 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Worker.php(159): Illuminate\Queue\Worker->process('beanstalkd', Object(Illuminate\Queue\Jobs\BeanstalkdJob), '20', '120')
#9 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Worker.php(113): Illuminate\Queue\Worker->pop(NULL, NULL, '120', 3, '20')
#10 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Worker.php(85): Illuminate\Queue\Worker->runNextJobForDaemon(NULL, NULL, '120', 3, '20')
#11 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Console/WorkCommand.php(101): Illuminate\Queue\Worker->daemon(NULL, NULL, '120', 128, 3, '20')
#12 /x/website/vendor/laravel/framework/src/Illuminate/Queue/Console/WorkCommand.php(67): Illuminate\Queue\Console\WorkCommand->runWorker(NULL, NULL, '120', 128, true) …

在 Postgres 12 数据库中，我在一个函数中有多个查询（SELECT、UPDATE、 ...），总共需要大约 20 分钟才能完成。我在顶部有一个检查，它执行UPDATEifstatus未运行：

create or replace function aaa.fnc_work() returns varchar as 
$body$
    begin
        if (select count(*) from aaa.monitor where id='invoicing' and status='running')=0 then
           return 'running';
        else
           update aaa.monitor set status='running' where id='invoicing';
        end if;
        --- rest of code ---
        --finally
        update aaa.monitor set status='idle' where id='invoicing';
        return '';
    exception when others then
         return SQLERRM::varchar;
    end
$body$
language plpgsql;

这个想法是为了防止其他用户执行直到--- rest of code ---空闲status。

然而，其他人（调用相同的函数）似乎看不到更新的状态，他们也继续并开始执行 …

关于Bucket_size参数的Google文档只提供了一个简短的定义,一个Wikipedia 链接和一个示例:

bucket_size

限制队列处理的突发性,
即更高的桶大小允许队列执行速率更大的峰值.例如,考虑一个速率为5/s且桶大小为10的队列.如果该队列已处于非活动状态一段时间(允许其"令牌桶"填满),并且20个任务突然排队,它将被允许立即执行10个任务.但是在接下来的第二步中,只能执行5个以上的任务,因为令牌桶已经耗尽并且以指定的5/s速率重新填充.

能否请您提供Bucket_size参数说明以及该参数在哪些方面有用？

我有一个有向无环图networkx。每个节点代表一个任务，节点的前驱是任务依赖项（给定任务在其依赖项执行之前无法执行）。

我想在异步任务队列中“执行”图表，类似于提供的celery服务（以便我可以轮询作业的状态、检索结果等）。Celery 不提供创建 DAG 的能力（据我所知），并且在所有依赖项完成后立即转移到 a 的能力task将是至关重要的（DAG 可能有多个路径，即使一个任务很慢/阻塞） ，可能会继续执行其他任务等）。

有没有任何简单的例子说明我如何实现这一目标，或者甚至networkx与集成celery？