标签: message-queue

我正在开发一个多人游戏,它需要一个消息队列(即消息输入,消息输出,没有重复或删除消息,假设没有意外的高速缓存驱逐).以下是我所知道的基于memcache的队列:

• MemcacheQ:http://memcachedb.org/memcacheq/
• Starling:http://rubyforge.org/projects/starling/
• Depcached:http://www.marcworrell.com/article-2287-en.html
• 麻雀:http://code.google.com/p/sparrow/

我从这篇博文中了解了memcache队列的概念:

所有消息都以整数作为键保存.有一个密钥具有下一个密钥,另一个密钥具有队列中最旧的消息的密钥.要访问这些,增量/减量方法用作其原子,因此有两个键充当锁.它们会递增,如果返回值为1,则进程具有锁定,否则它将继续递增.一旦完成该过程,它将值设置回0.简单但有效.需要注意的是整数将溢出,因此有一些逻辑可以在我们接近该限制时将使用的密钥设置为1.由于增量操作是原子的,只有在使用两个或更多个内存缓存(用于冗余)时才需要锁定,以保持这些内存同步.

我的问题是,是否有可以在App Engine上运行的基于memcache的消息队列服务？

我目前正在评估消息队列系统,RabbitMq似乎是一个很好的候选者,所以我正在深入研究它.

为了给出一些上下文,我希望有一个类似于一个交换负载的东西来平衡消息发布到多个队列.我不想复制邮件,因此不能选择扇出交换.

另外我之所以考虑拥有多个队列而不是一个队列来处理与消费者的循环,我不希望我们的单点故障处于队列级别.

听起来我可以在发布者端添加一些逻辑来通过编辑路由键并具有适当的绑定来模拟该行为.但这种被动方法不会考虑每个队列上的消息消耗速度,如果该队列的消费者应用程序已经死亡,可能会导致填满一个队列.

我正在寻找一种来自交换实体方面的更主动的方式,它将根据每个队列大小或某种性质决定在何处发送下一条消息.

我读到了Alice和可用的RESTful API,但这似乎是实现快速路由决策的重要解决方案.

任何人都知道如果在交换队列之间循环是否可行,那么RabbitMQ呢？谢谢.

我在SQS队列中有一小组消息,即使发送到AWS端点的删除请求返回200响应,也不会删除这些消息.我的应用程序处理的消息很好,删除请求也很好.

我正在使用Java AWS SDK 1.3.6.

还有其他人遇到过这个问题吗？

我正在涉及过程到过程的沟通; 目标是让工作进程执行一些计算并将结果传回控制进程.我安装了zeromq.node并在coffeescript中设置了一个简单的请求者和响应者.

请求者:

# requester.coffee

zmq                       = require 'zmq'
context                   = new zmq.Context()
socket                    = zmq.socket 'req'

socket.bind 'tcp://127.0.0.1:5555', ( error ) =>
  throw error if error?
  console.log 'Requesting writer bound to port 5555'
  setInterval ( -> socket.send 'helo world' ), 1


response_count  = 0
t0              = new Date() / 1000

socket.on 'message', ( message ) ->
  response_count += 1
  # x = message.toString 'utf-8'
  if response_count % 1000 is 0
    t1              = new Date() / 1000
    console.log …

我在这里看到(关于IPC的维基百科文章)这些是两个不同的东西,但即使阅读每一个专用的维基百科页面,我也不明白它的区别是什么.

有人能解释一下吗？

我们正在重建我们的消息队列系统.在浏览RabbitMQ交换类型时,我注意到有两种可能的解决方案来实现路由消息的多播特性.

1. 主题交流.通过设置主题交换和具有特定模式的路由密钥,消息将被路由到指定的队列.IE产品.*.根据AMQP规范,这通常是实现Pub/Sub模式的交换类型.

2. 标题交换.所谓的"类固醇直接交换".对于多播消息更加灵活,因为路由密钥被忽略,而每条消息都有"x-match"标头来表示消息应该传递给哪些队列.每条消息都可以以不同方式动态路由.但是,这种交换类型似乎与Message Queue设计更紧密地结合在一起,因为消费者/生产者必须更多地了解目标队列.

所以问题是,有没有人有这两种交换类型的经验,并分享上述两种类型的优点/缺点的更多特征？谢谢!

参考文献[1]:https://www.rabbitmq.com/tutorials/amqp-concepts.html

我在最近的老式Apple MacBook Pro上使用Python 2.7.5,它有四个硬件和八个逻辑CPU; 即,sysctl实用程序提供:

$ sysctl hw.physicalcpu
hw.physicalcpu: 4
$ sysctl hw.logicalcpu
hw.logicalcpu: 8

我需要在大型1-D列表或数组上执行一些相当复杂的处理,然后将结果保存为中间输出,稍后将在我的应用程序的后续计算中再次使用.我的问题的结构很自然地适用于并行化,所以我认为我会尝试使用Python的多处理模块将1D阵列细分为几个部分(4件或8件,我还不确定哪个),执行并行计算,然后将结果输出重新组合成最终格式.我正在尝试决定是使用multiprocessing.Queue()(消息队列)还是multiprocessing.Array()(共享内存)作为我的首选机制,用于将子进程的结果计算传递回主父进程,我一直在尝试使用几个"玩具"模型,以确保我理解多处理模块实际上是如何工作的.然而,我遇到了一个相当意外的结果:在为同一个问题创建两个本质上等效的解决方案时,使用共享内存进行进程间通信的版本似乎比使用消息的版本需要更多的执行时间(比多30倍!)队列.下面,我为"玩具"问题提供了两个不同版本的示例源代码,它使用并行进程生成一长串随机数,并以两种不同的方式将聚集结果传回父进程:首先使用消息队列,第二次使用共享内存.

以下是使用消息队列的版本:

import random
import multiprocessing
import datetime

def genRandom(count, id, q):

    print("Now starting process {0}".format(id))
    output = []
    # Generate a list of random numbers, of length "count"
    for i in xrange(count):
        output.append(random.random())
    # Write the output to a queue, to be read by the calling process 
    q.put(output)

if __name__ == "__main__":
    # Number of random numbers to be generated …

我正在编写一些代码来替换带有POSIX消息队列的TCP套接字.有时程序会崩溃(仍在开发中),并且不会删除创建的队列(不执行:mq_close()+ mq_unlink()).当我再次运行代码时,这会导致问题.

有没有办法使用命令行删除/删除这些队列？我试过用:ipcs -q.这无法列出任何队列.

我试过了:lsof | grep queue-name.他们确实出现在这里.

理想情况下,我想使用:ipcrm.

从我看到的情况来看,如果Node中的事件需要"长时间"调度,Node会创建某种"事件队列",并且会尽快逐个触发它们.

这个队列可以有多长时间？

我有一个应用程序,我需要将一些数据存储在数据库(例如mysql)中,然后在消息队列中发布一些数据.我的问题是:如果应用程序在数据库中存储后崩溃,我的数据将永远不会写入消息队列然后丢失(因此我的系统的最终一致性将无法保证).我怎么解决这个问题 ？