标签: message-queue

我需要在Java中动态创建异步消息队列.我的用例是通过多个SMTP服务器发送电子邮件:我需要按顺序强制执行相同SMTP服务器的电子邮件,但可以同时处理发送到不同SMTP服务器的电子邮件.我以前使用过JMS,但据我所知它只允许编译时队列创建,而我需要在运行时创建队列(每个SMTP服务器一个队列).

我是否遗漏了有关JMS的内容,或者是否有其他工具/建议我应该查看一下？

我正在使用Grails构建一个社区网站,我想实现与stackoverflow.com完全相同的用户通知.例如,用户将在他返回网站时通过电子邮件或邮箱收到新事件("您有新评论","您最喜欢的一个已经改变了......")的通知.

我想这是一个常见的问题,我想听听你在Grails领域实现后端的建议是什么简单的解决方案.例如,Java Message Service是推荐的解决方案吗？

谢谢.

假设我有几百万的事件,在SQS队列,我希望尽快摆脱他们,但我不能只删除队列,使一个新的.

从队列中删除/排除这些事件的最快方法是什么？

我们有一个基于PHP的大型Web应用程序.此应用程序允许安排推文和墙上帖子,并且有从服务器发出的预定电子邮件.

通过'预定',我的意思是这些是计划在特定时间运行的PHP脚本cron.大约有7个PHP文件可以完成上述工作.

我听说过消息队列.任何人都可以解释消息队列是否最适合这种情况？消息队列是否执行PHP脚本？或者我们需要完全不同地配置它吗？有什么优点/缺点？

从不太酷的问题类别......

"队列式的东西"我的意思是支持以下操作:

• append(entry:Entry) - 在队列尾部添加条目
• take():Entry - 从队列头部删除条目并返回它
• promote(entry_id) - 将条目移动到靠近头部的位置; 当前占据该位置的条目将移动到旧位置
• 降级(entry_id) - 与促销相反(entry_id)

可选操作类似于:

• 提升(entry_id,amount) - 比如提升(entry_id),除非你指定了仓位数量
• 降级(entry_id,金额) - 与促销相反(entry_id,金额)
• 当然,如果我们允许数量为正数或负数,我们可以通过单一移动(entry_id,amount)方法合并推广/降级方法

如果可以以分布式方式对队列执行以下操作(多个客户端与队列交互),那将是理想的:

queue = ...

queue.append( a )
queue.append( b )
queue.append( c )

print queue
"a b c"

queue.promote( b.id )
print queue
"b a c"

queue.demote( a.id )
"b c a"

x = queue.take()
print x
"b"
print queue
"c a"

是否有任何特别适合此用例的数据存储？即使多个用户同时修改队列,队列也应始终处于一致状态.

如果不是为了促进/降级/移动要求,那就不会有太大的问题.

编辑:如果有Java和/或Python库来完成上述任务的加分点.

解决方案应该非常好地扩展

一些背景,在得到真正的问题之前:

我正在开发一个由几个不同模块组成的后端应用程序.目前,每个模块都是一个命令行java应用程序,它是"按需"运行的(稍后会详细介绍).

每个模块都是一个"步骤",是一个更大的过程的一部分,您可以将其视为数据流; 第一步从外部源收集数据文件并将其推送/加载到某些SQL数据库表中; 然后根据不同的条件和事件(时间,数据库中存在数据,通过Web服务/ Web界面完成的消息和详细说明),从(1个或多个)DB表中获取数据,处理它们,并将它们写在不同的表格上.步骤在三个不同的服务器上运行,并从三个不同的DB读取数据,但只能在一个DB中写入.目的是汇总数据,计算指标和统计数据.

目前,每个模块都是定期执行的(从第一个模块的几分钟/小时,到链中最后一个模块的几天,需要聚合更多数据,因此等待"更长时间"从它们可用),使用的cronjob.运行一个模块(当前是一个java控制台应用程序),它会检查数据库中给定日期时间窗口中新的未处理信息,并完成其工作.

问题:它有效,但是......我需要扩展和维护它,这种方法开始显示其局限性.

1. 我不喜欢依靠"民意调查"; 这是浪费,考虑到以前模块的信息足以在他们需要的信息可用时"告诉"链中的其他模块,并且他们可以继续.
2. 它"缓慢":链条上的模块延迟了几天,因为我们必须确保数据是由前面的模块到达和处理的.所以我们"停止"这些模块,直到我们确定我们拥有所有数据.新增功能需要实时(不是很难,但"尽快")计算某些指标.一个很好的例子就是在这里发生的事情,在SO上,带有徽章!:)我需要获得一些非常相似的东西.

为了解决第二个问题,我将介绍"部分"或"增量"计算:只要我有一组相关信息,我就会处理它.然后,当一些其他链接信息到达时,我计算差异并相应地更新数据,但我还需要通知其他(从属)模块.

问题

- 1)哪种方法最好？ - 2)相关:哪些是"通知"其他模块(在我的情况下是java可执行文件)相关数据可用的最佳方式？

• 将其他"非数据"表添加到数据库中,每个模块写入"嘿,我已经完成了这个并且它可用".当cronjob启动另一个模块时,它会读取表格,确定他可以计算子集xxx,然后执行.等等
• 使用Message Queues,如ZeroMQ(或Apache Camel,如@mjn建议)而不是DB表
• 使用像Redis这样的键值存储,而不是数据库表

编辑:我确信基于队列的方法是要走的路,我为完整性添加了"table + polling"选项,但现在我明白这只是一种分心(显然,每个人都会回答"是的,使用队列,民意调查是邪恶的" - 这是正确的!".因此,让我重新解释一下这个问题: 使用像Redis这样的pub/sub的键值存储使用MQ的优点/缺点是什么？

• 3)有没有任何解决方案可以帮我完全摆脱cronjobs？

编辑:特别是,在可能的情况下,它意味着:在某些MQ和/或键值存储中是否有一种机制可以让我发布带有"时间"的消息？比如"在1天内交付"？显然,持久性和"几乎一次"交付保证

• 4)我应该将此消息(基于事件？)的解决方案构建为集中式服务,并将其作为其中一个服务器上的守护程序/服务运行吗？
• 5)我是否应该放弃按需启动订阅者的想法,让每个模块作为守护进程/服务连续运行？
• 6)哪些是赞成和缺点(可靠性,单点故障与资源使用和复杂性......)？

编辑:这是我最关心的一点:我想"排队"自己根据队列中的消息激活"模块",类似于MSMQ激活.这是个好主意吗？Java世界中有什么东西可以实现它,我应该自己实现它(通过MQ还是通过Redis),还是应该将每个模块作为守护进程运行？(即使某些计算通常是在突发中发生,两小时处理然后是两天的空转？)

注意:我不能使用重型容器/ EJB(No Glassfish或类似产品)

编辑:骆驼对我来说似乎有点太重了.无论是在资源还是开发的复杂性方面,我都在寻找一些非常轻松的东西

我必须建立一个运动检测服务.动态检测不对视频进行操作,而是对静止图像进行操作.

这个微服务需要能够无序地接收图像(带有时间戳),并确定图像是否与之前拍摄的图像不同(时间戳早于它).需要有多个运动检测工作人员.

因此,关键要求似乎是:

1. 不按顺序接收图像的Web服务能够将它们分组为上一对和下一对,然后计算图像是否具有与其先前图像相比的运动.
2. 许多图像制作者 - 平均每秒大约有100张图像
3. 许多运动检测消费者
4. 优先考虑吞吐量的延迟.
5. 不易单独使用的任务.

我在考虑使用单个消息队列.生产者将图像文档推送到队列中.然后,运动检测工作者从该队列中读取,并向该文档添加"diff_percentage"字段,并在数据库中更新该记录.

给定队列中的任务,工作人员可以单独操作该任务,方法是直接从数据库中取出图像,然后比较它,然后更新数据库中的记录.不幸的是,虽然这种方法运作得很好,但速度却非常慢.我想我需要减少数据库的读取.理想情况下,我希望这个"队列"能够缓冲,直到它具有给定作业所需的图像.类似于...当一个worker从队列中读取时,检查它需要比较的图像是否在队列中,如果没有,则转到数据库.

谁能指出我正确的方向？也许队列不是我想要的？也许是一个队列,还有某种缓存桶？

我的系统在Amazon自动缩放组上运行,一个功能允许用户进行用户消息传递,我有以下用例要解决.

1. 用户之间在我的应用程序中发送新消息.
2. 通过电子邮件通知用户的消息将被丢弃到延迟60秒的队列中.此延迟允许实时聊天客户端(faye/angularjs)有时间查看消息并将其标记为已查看.
3. 在延迟之后,拾取消息,检查"读取"状态,如果客户端尚未读取,则发送电子邮件.

最初我打算在每个应用程序服务器上使用cronjob轮询消息队列,但是对我来说,使用SNS调用某种电子邮件发送端点(可能在Lambda中)会更有效.

我看不出有任何方法可以进行SNS调查SQS,但有人可以建议如何做到这一点吗？基本上我希望SNS有延迟,这样我就不会在电子邮件警报的"实时"聊天中向某人发送垃圾邮件.

谢谢

使用Spring云监听AWS SQS队列,如下所示:

@SqsListener(value = "${queue.name}", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void queueListener(String message, @Headers Map<String, Object> sqsHeaders) {
    // code
}

Spring配置:

<aws-messaging:annotation-driven-queue-listener
    max-number-of-messages="10" wait-time-out="20" visibility-timeout="3600"
    amazon-sqs="awsSqsClient" />

AwsSqsClient:

@Bean
public com.amazonaws.services.sqs.AmazonSQSAsyncClient awsSqsClient() {
    ExecutorService executorService = Executors.newFixedThreadPool(10);
    return new AmazonSQSAsyncClient(new DefaultAWSCredentialsProviderChain(), executorService);
}

这很好用.

在SQS客户端中配置了10个线程来处理这些消息,如上所示.这也工作正常,在任何时间点处理最多10条消息.

问题是,我无法弄清楚控制轮询间隔的方法.默认情况下,一旦所有线程都空闲,弹出轮询.

即考虑以下示例

1. 大约有3条消息被发送到Queue
2. Spring轮询队列并获得3条消息
3. 处理每条消息的3条消息大约需要20分钟

与此同时,大约有25条消息被送到队列.在完成之前传递的所有3条消息之前,Spring不会轮询队列.从上面的例子来看,只有20分钟之后春季民意调查,尽管有7个线程仍然免费!

知道我们如何控制这种民意调查吗？即如果有任何线程空闲,则应该开始轮询,并且不应该等到所有线程都空闲

我试图理解像卡夫卡这样的东西和像骆驼这样的东西.据我所知,Camel将为开发人员提供更多抽象,而不必担心在某种程度上更改协议/系统.Kafka如何能够处理Camel现在可以做的大部分事情？我正在阅读文档,看起来Kafka已经更新/升级到足以略微脱离成为消息代理.我想我的问题真的可以归结为卡夫卡在未来打样系统方面与Camel的比较,以及卡夫卡在哪些方面没有骆驼？我的印象是Kafka的扩展性不如系统增长.

编辑:这完全基于消息.围绕Camel的文档非常清楚它基于企业集成模式,但我深入研究Kafka文档可以实现相同的模式.我错过了什么吗？