标签: message-queue

是否有更快的方法来更新匹配特定条件的MySQL表的最旧行,而不是使用ORDER BY id LIMIT 1以下查询中的情况？

UPDATE mytable SET field1 = '1' WHERE field1 = 0 ORDER BY id LIMIT 1;

注意:

• 假设主键是,id并且还有一个索引field1.
• 我们正在更新一行.
• 我们没有严格更新最旧的行,我们正在更新与条件匹配的最旧行.
• 我们想要更新最旧的匹配行,即最低的id,即FIFO队列的头部.

问题:

• 有ORDER BY id必要吗？MySQL如何默认订购？

现实世界的例子

我们有一个DB表用于电子邮件队列.当我们要将电子邮件排队以发送给我们的用户时,会添加行.行由cron作业删除,每分钟运行一次,在该分钟内尽可能多地处理并每行发送1封电子邮件.

我们计划放弃这种方法,并使用Gearman或Resque之类的东西来处理我们的电子邮件队列.但与此同时,我有一个问题,我们如何有效地标记队列中最旧的项目进行处理,即具有最低ID的行.此查询完成工作:

mysql_query("UPDATE email_queue SET processingID = '1' WHERE processingID = 0 ORDER BY id LIMIT 1");

但是,由于扩展问题,它出现在mysql慢日志中很多.当表有500,000行时,查询可能需要10秒以上.问题是这个表自首次引入以来已经大量增长,现在有时有50万行,开销为133.9 MiB.例如,我们每天INSERT 6000个新行可能180次,并且删除大致相同的数字.

为了阻止查询出现在慢速日志中,我们删除了ORDER BY id以阻止整个表格的大规模.即

mysql_query("UPDATE email_queue SET processingID = …

我正在设计一个系统,它将使用jms和一些消息传递软件(我倾向于使用ActiveMQ)作为中间件.将有少于100个代理,每个代理每天最多推送5000条消息.

每条消息的有效负载大约为100字节.我希望大约有一半(2500)的消息聚集在午夜左右,而另一半则在白天有些均匀分布.上面给出的数字都是我所期望的更高端.(是的,我可能会在不久的将来吃掉那个声明).

有一种类型的消息,其中有效载荷将相当大,例如在5-50mb的范围内.这些消息每天只会从每个代理发送几次.

我的问题是: 这会以任何方式引起我的问​​题,还是通过消息队列发送大量数据是完全正常的？

例如,它会在处理较大的消息时减少吞吐量(较小的消息排队)吗？

或者消息队列会阻塞更大的消息？

或者我应该以不同的方式处理这个问题,比如通过jms发送数据的位置,让终端接收器在其他地方获取数据？(我希望不会因为耦合,安全问题和额外配置而导致特殊情况).

我对jms的实用细节完全不了解,所以请告诉我是否需要提供更多详细信息.

编辑: 我接受了Andres真正棒极了的回答.继续发布建议和意见,我会保持一切有用的东西.

我试图弄清楚通过wcf使用rabbitmq集群的最佳解决方案是什么.

目前的设置:

1. 2个IIS Web服务器(充当消息生成并通过amqp wcf客户端将消息发布到队列).
2. 带有rabbitmq代理的2台服务器(与镜像队列,rabbit1和rabbit2集群)
3. 具有托管amqp wcf服务的Windows服务(worker),用于侦听传入消息.

Web角色将消息发布到rabbit1节点,worker也监听rabbit1节点.如果rabbit1节点失败,系统(web和worker)都应该切换到rabbit2.这就是问题,如何以更优雅的方式实现它,而不是处理应用程序代码中的连接失败.

首先,我现在看到的唯一方法是使用wcf4路由备份端点功能.这种方式只解决了客户端(Web角色)的问题,但没有解决wcf服务端(worker角色)的问题.

消息总线体系结构的每个描述似乎都描述了用于路由消息的集中式中介:

每个应用程序都与消息总线有一个连接,而不是与每个其他应用程序的多个专用连接.

http://msdn.microsoft.com/en-us/library/ff647328.aspx

将这些应用程序之间的连接中间件构建为消息总线,使它们能够使用消息传递协同工作.

http://www.eaipatterns.com/MessageBus.html

Service Bus是位于应用程序之间的消息传递基础结构,允许它们以松散耦合的方式交换消息,以提高规模和弹性.

http://www.windowsazure.com/en-us/home/features/messaging/

描述看起来与Message Broker体系结构相同; 让我相信以下几点:

术语"消息总线"实际上与"MoM"相同 - 一个用于引用消息传递基础结构而没有明确提及拓扑的通用术语.即Message Broker 是一种消息总线.

这似乎是一个有争议的话题,虽然在SO上有一些类似的问题,但没有一个(IMO)有明确的答案.

Ps我知道企业服务总线完全不同,实际上是一个"功能丰富"的MoM.

gcc 4.7.2
c89

我正在寻找一种跨平台（Linux，Windows）的轻量级消息队列（IPC）。

我用过msgget，msgsnd和msgrcv。但这不是跨平台的，因此无法在Windows上使用。

我不需要任何套接字层（tcp，udp）。由于所有消息传递都将在同一服务器上。

基本上，我有一个事件循环，并且我将阻塞（独立线程），直到从消息队列（msgrcv）获得消息为止。然后处理该事件。这些消息将来自我创建的另一个共享库，该库将监视事件循环中的事件。

我只希望能够将消息发送到消息队列（msgsnd），然后在我的事件循环中接收它们（msgrcv）。

我想知道是否有某种方法可以延迟处理akka消息？

我的用例: 对于我的每一个请求,我都需要做少量的工作,然后我需要在两小时后完成额外的工作.

有没有简单的方法来延迟AKKA中的消息处理？我知道我可以设置一个外部分布式队列,例如ActiveMQ,RabbitMQ,它可能有这个功能,但我不愿意.

我知道我需要使邮箱持久耐用,以便它可以在重新启动或崩溃时存活.我们已经有了mongo设置,所以我可能会使用MongoBasedMailbox来提高耐用性.

我正在尝试根据我的要求找出最佳的MQ选项.我需要能够以高可靠性在地理上不同的数据中心内和跨地域传输文本和二进制消息.快速很好,但缩放也是一种选择.像RabbitMQ一样支持很好.

以下是一些假设:

1. 使用联合或铲除消息在数据中心之间推送相同的消息.
2. 使用AMQP传输二进制消息,因为我们是.Net/Python商店.

我想确保我的假设是有效的,并需要帮助选择哪个MQ.我过去使用过ActiveMQ + MySQL,但我喜欢Mnesia的选项,用于持久的消息传递.另外,使用AMQP 0.9而不是1.0是否可以.通过插件看起来像RabbitMQ支持1.0.

感谢我能得到的任何其他建议.

一个EventHandler(DatabaseConsumer)Disruptor调用存储过程在数据库中,这是如此之慢,以至于它会阻塞Disruptor一段时间.

因为我需要Disruptor保持运行而不会阻塞.我正在考虑添加一个额外的队列,以便EventHandler可以作为Producer另一个新创建的线程可以Consumer用来处理数据库的工作,这可以是异步的,而不会影响Disruptor

这是一些约束:

1. 该对象Disruptor传递给EventHandler周围30KB这个对象的数量约为40万.理论上,需要处理的对象的总大小约为30KBX400K = 12GB.所以额外的队列应该足够了.
2. 由于性能很重要,因此应避免GC暂停.
3. Java程序的堆大小只有2GB.

我认为文本文件是一个选项.EventHandler(生产者)将对象写入文件并Consumer从中读取并调用存储过程.问题是如何处理它到达文件末尾的情况以及如何知道新的即将到来的行.

有谁解决过这种情况吗？有什么建议？

我想使用Spring Messaging为我的webapp为已登录用户创建一个实时通知系统.

我定义AbstractWebSocketMessageBrokerConfigurer如下:

@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/notifications").withSockJS()
        .setSessionCookieNeeded(true)
        .setWebSocketEnabled(true);
}

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.enableSimpleBroker("/topic/", "/queue/");
}

并且,根据文档:

应用程序可以发送针对特定用户的邮件.Spring的STOMP支持识别前缀为"/ user /"的目的地.例如,客户端可能订阅目标"/ user/queue/position-updates".该目的地将由UserDestinationMessageHandler处理并转换为用户会话唯一的目的地,例如"/ queue/position-updates-user123".这提供了订阅一般命名的目的地的便利性,同时确保不与订阅相同目的地的其他用户发生冲突,使得每个用户可以接收唯一的库存位置更新.

在发送方,消息可以发送到目的地,例如"/ user/{username}/queue/position-updates",然后由UserDestinationMessageHandler将其转换为一个或多个目的地,每个目的地对应一个目的地.用户.这允许应用程序中的任何组件发送针对特定用户的消息,而不必知道除其名称和通用目标之外的任何内容.通过注释和消息传递模板也支持此功能.

通过发送消息/user/{username}/queue/something,它将仅传递给标识的特定用户{username}.

现在,我正在寻找一种解决方案,允许我使用外部Message Broker(例如,RabbitMQ),Spring就像Broker Relay一样:

registry.enableStompBrokerRelay("/topic/", "/queue/");

在Spring中配置外部消息代理之后:

• 是否可以使用as channel在Message Broker上发送消息/user/{username/}/queue/something？如果有,怎么样？
• 通过使用as channel在Message Broker上发送消息/user/{username/}/queue/something,Spring是否只能{username}根据当前的消息发送该消息Principal？

我讨论了RabbitMQ设计与我之间的同事Server A和Server B,这里是流程图如下.

Server A将消息发送到Server B通过exchange A和Queue B,以及从接收消息Server B通过Queue A.反之亦然Server B.

Server A实现了两个类C++; 一个是sender,另一个是receiver.相同的代码结构用于Server B实现JavaScript.

对于sender和receiver在Server A.我的同事的想法是:

• 发件人中的初始化:
  • 创建连接 rabbitmq
  • 宣布 exchange A
  • 宣布 Queue B
  • 将密钥绑定到 Queue B
• Receiver中的初始化:
  • 创建连接 rabbitmq
  • 宣布 exchange B
  • 宣布 Queue A
  • 将密钥绑定到 Queue A

同样的逻辑Server B …