标签: rabbitmq

这些技术套件适用于多用户在线游戏项目..项目要求:1.能够在任何给定时间处理2k-5k用户.2.用于iphone和android的客户端库(原生,没有javascript).3.用于Microsoft Windows的客户端库(最重要的),也用于mac os x和linux.4.专门用于mod开发的良好文档5.项目不是开源的.因此,只能使用具有适当许可证的库.

我也可以在erlang和java中编程,编程语言不是问题.

我正在研究如Openfire,Tigase,ejabberd和RabbitMQ等服务器技术.所有这些对我的项目都有好处,但我想了解更多关于我需要的套件,AMQP或XMPP.

AMQP专门为实时在线游戏提供什么.那么xmpp是更好的选择吗？

我想了解什么是scenario/usecases/RabbitMQ之类的消息可以帮助消费者Web应用程序.

是否有任何具体的资源可供学习？

目前哪些Web应用程序正在使用此类消息传递方案以及如何使用？

当RabbitMQ有消息时,RabbitMQ是否为消费者调用回调函数,或者消费者是否必须轮询RabbitMQ客户端？

所以在消费者方面,如果有PHP脚本,RabbitMQ可以调用它并将消息/参数传递给它.例如,如果在shard 1上提交了rating并且aggregateRating表在shard 2上,那么在shard 2上的RabbitMQ使用者会触发脚本说aggRating.php并传递插入到shard 1中的参数吗？

我正在使用Pika处理来自RabbitMQ的数据.由于我似乎遇到了不同类型的问题,我决定编写一个小型测试应用程序来查看如何处理断开连接.

我写了这个测试应用程序,其中包括:

1. 连接到Broker,重试直到成功
2. 连接时创建队列.
3. 使用此队列并将结果放入python Queue.Queue(0)
4. 从Queue.Queue(0)获取项目并将其生成回代理队列.

我注意到的是2个问题:

1. 当我从一台连接到另一台主机上的rabbitmq的主机(在vm内)运行我的脚本时,这些脚本随机退出而不会产生错误.
2. 当我在安装RabbitMQ的同一主机上运行我的脚本时,它运行正常并继续运行.

这可能是因为网络问题,数据包丢失,但我发现连接不是很强大.

当脚本在RabbitMQ服务器上本地运行并且我杀死RabbitMQ时,脚本退出时出现错误:"ERROR pika SelectConnection:3:104上的套接字错误"

所以看起来我不能让重新连接策略按原样运行.有人可以查看代码,看看我做错了什么？

谢谢,

松鸦

#!/bin/python
import logging
import threading
import Queue
import pika
from pika.reconnection_strategies import SimpleReconnectionStrategy
from pika.adapters import SelectConnection
import time
from threading import Lock

class Broker(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.logging = logging.getLogger(__name__)
        self.to_broker = Queue.Queue(0)
        self.from_broker = Queue.Queue(0)
        self.parameters = pika.ConnectionParameters(host='sandbox',heartbeat=True)
        self.srs = SimpleReconnectionStrategy()
        self.properties = pika.BasicProperties(delivery_mode=2)

        self.connection = None
        while True:
            try:
                self.connection = SelectConnection(self.parameters, self.on_connected,  reconnection_strategy=self.srs)
                break
            except Exception as err:
                self.logging.warning('Cant …

本来喜欢使用Amazon SQS,如果它提供了一些相似的FIFO访问,但序列似乎是完全随机的.

是否有什么能够为我提供FIFO排队作为云服务以及SQS的高可用性？

如果这要求太多 - 在EC2中将上述要求放在一起的最简单方法是什么？或者换句话说,在EC2中最简单的高可用性排队解决方案是什么？

感谢您的见解!

我在rabbitmq上使用芹菜.我已经向队列发送了数千条消息,并且它们正在被成功处理,一切正常.但是,几个rabbitmq队列中的消息数量正在增长(队列中有数十万个项目).队列命名celeryev.[...](见下面的截图).这是恰当的行为吗？这些队列的目的是什么,不应该定期清除它们？有没有办法更频繁地清除它们,我认为它们占用了相当多的磁盘空间.

有没有办法计算一个工作被重新排队的时间(通过拒绝或Nak)而无需手动重新计算工作？我需要在'n'时间重试一份工作,然后在'n'时间之后放弃它.

ps:目前我手动重新排队作业(删除旧作业,创建具有确切内容的新作业,如果计数器不存在或者值小于'n',则额外的Counter标题)

场景(我简化了一些事情):

• 许多最终用户可以从前端Web应用程序(生产者)开始工作(繁重的工作,例如渲染大型PDF).
• 作业将发送到单个持久的RabbitMQ队列.
• 许多工作者应用程序(使用者)处理这些作业并将结果写回数据存储区.

这个相当标准的模式工作正常.

问题是:如果用户在同一分钟内启动了10个作业,并且在一天中的那个时间只有10个工作者应用程序启动,则该最终用户有效地接管了他自己的所有计算时间.

问题:如何确保每个最终用户只能处理一个作业？(奖励:一些最终用户(例如管理员)不得受到限制)

此外,我不希望前端应用程序阻止最终用户启动并发作业.我只希望最终用户等待他们的并发作业一次完成一个.

解决方案？:我应该为每个最终用户动态创建一个自动删除独占队列吗？如果是,我如何告诉worker应用程序开始使用此队列？如何确保一个(并且只有一个)工作者将从此队列中消耗？

RabbitMQ .NET客户端是否有任何异步支持？我希望能够异步连接和使用消息,但到目前为止还没有找到办法.

(对于消费消息,我可以使用EventingBasicConsumer,但这不是一个完整的解决方案.)

为了给出一些上下文,这是我现在如何使用RabbitMQ的一个例子(代码取自我的博客):

var factory = new ConnectionFactory() { HostName = "localhost" };

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        channel.QueueDeclare("testqueue", true, false, false, null);

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += Consumer_Received;
        channel.BasicConsume("testqueue", true, consumer);

        Console.ReadLine();
    }
}

码:

RabbitMQListener:

@Component
public class ServerThroughRabbitMQ implements ServerThroughAMQPBroker {
    private static final AtomicLong ID_COUNTER=new AtomicLong();
    private final long instanceId=ID_COUNTER.incrementAndGet();


    @Autowired
    public ServerThroughRabbitMQ( UserService userService,LoginService loginService....){
....
    }

    @Override
    @RabbitListener(queues = "#{registerQueue.name}")
    public String registerUserAndLogin(String json) {
       .....
    }

SERVERCONFIG:

@Configuration
public class ServerConfig {
    @Value("${amqp.broker.exchange-name}")
    private String exchangeName;
    @Value("${amqp.broker.host}")
    private String ampqBrokerHost;
    @Value("${amqp.broker.quidco.queue.postfix}")
    private String quidcoQueuePostfix;
    @Value("${amqp.broker.quidco.queue.durability:true}")
    private boolean quidcoQueueDurability;
    @Value("${amqp.broker.quidco.queue.autodelete:false}")
    private boolean quidcoQueueAutodelete;

    private String registerAndLoginQuequName;


    @PostConstruct
    public void init() {
        registerAndLoginQuequName = REGISTER_AND_LOGIN_ROUTING_KEY + quidcoQueuePostfix;
    public String getRegisterAndLoginQueueName() …