标签: message-queue

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

我正在开发一个由几个不同模块组成的后端应用程序.目前,每个模块都是一个命令行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或类似产品)

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

我期待在单个进程多线程应用程序上使用POSIX消息队列.mqueues将用于在线程之间共享数据.

我对它们在Linux内核中的工作方式感到有些困惑.是否所有消息都通过内核空间然后返回到接收上的用户空间？又来自用户空间线程,我做一个mq_send,消息最终在内核空间中,然后在接收时,它是从内核空间获取消息的另一个系统调用.如果是这样,对于高使用率的消息队列来说这不是非常低效吗？

我读了一篇关于多线程程序设计http://drdobbs.com/architecture-and-design/215900465的文章,它说这是"用异步消息替换共享数据的最佳实践.尽可能地保留每个线程的数据"隔离(非共享),让线程通过传递数据副本的异步消息进行通信".

令我困惑的是,我没有看到使用共享数据和消息队列之间的区别.我现在正在开发一个关于Windows的非gui项目,所以让我们使用windows的消息队列.并以传统的生产者 - 消费者问题为例.

使用共享数据,将有一个共享容器和一个锁定生产者线程和消费者线程之间容器的锁.当生产者输出产品时,它首先等待锁定然后向容器写入内容然后释放锁定.

使用消息队列,生产者可以简单地PostThreadMessage而不用阻塞.这是异步消息的优势.但我认为必须存在一些锁定两个线程之间的消息队列,否则数据肯定会被破坏.PostThreadMessage调用只是隐藏细节.我不知道我的猜测是否正确但是如果它是真的,优势似乎不再存在,因为两种方法都做同样的事情,唯一的区别是系统在使用消息队列时隐藏细节.

PS.也许消息队列使用非阻塞包含器,但我也可以使用前一种方式的并发容器.我想知道消息队列是如何实现的,这两种方式之间是否有任何性能差异？

更新:如果消息队列操作仍然在其他地方被阻止,我仍然没有得到异步消息的概念.如果我猜错了,请纠正我:当我们使用共享容器和锁时,我们将在我们自己的线程中阻塞.但是当使用消息队列时,我自己的线程立即返回,并将阻塞工作留给某个系统线程.

数据库事务是一个熟悉的概念.

try {
  ...
  ..
  updateDB()
  ..
  ...
  commit();
} catch error {
  rollback();
}

如果发生任何错误,则将丢弃updateDB所做的任何更改.

我想知道消息队列事务回滚将撤消什么.

try{
  ...
  ...
  //EDIT: swapped the order of receive and send
  Message m = queue1.receiveMessage(..)
  ..
  ..
  queue2.sendMessage(..)
  ..
  ..
  commit();
} catch error {
  rollback();
}

具体来说,回滚会做什么

1. 取消发送消息
2. 取消接收消息,即将收到的消息放回队列

或者我是将数据库tx拉得太过分了.

谢谢

编辑:我并不是暗示发送和接收操作是相关的.我只是想说有两个操作可以改变消息代理的状态 - 接收将从队列中取出一条消息,如果有的话,它将对其他消费者不可用.

我似乎无法在任何地方找到这些信息,或者我可能不理解它.在RabbitMQ的发布/订阅模式中,当生产者生成消息时,消费者如何知道队列中有新消息？

消费者是否经常轮询队列以检查是否有新消息或交换"推送"通知消费者是否有新消息？

我需要将在Heroku上运行的Node JS服务器移动到消息队列架构.目前,服务器接收HTTP请求,进行一些处理并进行响应.问题是处理需要一些时间,特别是当有大量请求时.这个漫长的处理时间会导致服务器超时,过载和崩溃!我的阅读告诉我需要后台工作人员来处理.

我对消息队列和后台工作者没有经验,我正在寻找一个非常简单的示例来开始.任何人都可以建议一个简单,易懂的模块或示例来开始？

我发现了一些例子,但它们很复杂,我迷路了!我想要一个我可以构建的准系统示例.

我的项目将Apache Kafka视为基于JMS的老化消息传递方法的潜在替代品.为了使这种转换尽可能顺利,如果替换排队系统(Kafka)具有异步订阅机制,这将是理想的,类似于我们当前项目使用MessageListener和MessageConsumer订阅主题并接收异步通知的JMS机制.如果Kafka没有严格遵守JMS API,我并不在乎,但相反,如果我不需要,我宁愿不重新设计我们的整套发布 - 订阅 - 通知类.

我可以找到各种KafkaConsumer 轮询 示例,但到目前为止还没有找到任何客户端通过异步通知通知新消息的示例.

有谁知道Kafka的当前版本(截至本文发布时为0.10.2)是否提供了这样的API,还是我一直试图使用轮询重写我的遗留代码？

谁能解释一下Android中的消息队列是什么？它是正在运行的进程列表吗？我找不到很好的来源来解释它。

我问这个问题是因为我正在阅读有关视图类的方法帖子。

邮政

添加到 API 级别 1 布尔值帖子（可运行操作）

导致 Runnable 添加到消息队列中。可运行程序将在用户界面线程上运行。

先感谢您。

当我查看机器人操作系统（ROS）的教程时，我发现大多数示例代码将发布者的队列大小设置为较大的值，例如1000。我认为这会导致节点失去实时响应。

人们出于什么目的将其设置为那么大的值？

我有一个简单的消息配置：

framework:
    messenger:
        failure_transport: failed
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 2
                    delay: 10000
            failed: 'doctrine://default?queue_name=failed'

        routing:
            someInterface: async

我想知道路由到失败传输的每条消息，当消息重定向时，系统应该向我发送电子邮件通知。我唯一的想法是定期检查messenger_messages包含存储在失败传输中的消息的表，但我认为这不是 symfony 方式的解决方案。您知道针对此功能的更聪明的解决方案吗？

我正在使用 symfony/messenger 版本 5.*