标签: message-queue

我正在尝试使用ZeroMQ的pub子消息,但客户端要求代码都是Java.我知道ZeroMQ有Java绑定但仍然依赖于ac库,因此我无法使用它.是否有可以用来连接服务器的ZeroMQ客户端,还是我自己做的简单实现？

最近我一直在研究使用Bean的Beanstalkd.我已经学到了很多,但对服务器上的设置有一些疑问等.

以下是我看到它的工作方式:

1. 我在我的Ubuntu服务器上安装了Beanstalkd和任何依赖项(例如libevent).然后我启动Beanstalkd守护程序(它应该基本上一直运行).
2. 在我的网站的某个地方(例如当用户执行某些操作等)时,任务会被添加到Beanstalkd队列中的各个管道中.

3. 我有一个bash脚本(例如下面的一个),它作为一个基本上执行PHP脚本的守护进程运行.

   #!/bin/sh
   php worker.php
   

   Run Code Online (Sandbox Code Playgroud)

4)worker脚本会有类似的东西来执行排队的任务:

while(1) {
  $job = $this->pheanstalk->watch('test')->ignore('default')->reserve();
  $job_encoded = json_decode($job->getData(), false);
  $done_jobs[] = $job_encoded;
  $this->log('job:'.print_r($job_encoded, 1));
  $this->pheanstalk->delete($job);
}

现在这里是基于上述设置的问题(如果我错了,请纠正我):

1. 假设我有将RSS源导入数据库或其他任务的任务.如果10个用户同时执行此操作,则他们将全部排在"测试"管中.但是,它们只能一次执行一个.将10个不同的管同时执行会更好吗？

2. 如果我确实需要更多电子管,那么这是否也意味着我需要10个工作脚本？除了watch()函数中的字符串文字之外,每个管都有一个基本相同的代码同时运行.

3. 如果我将该脚本作为守护进程运行,那该怎么办？它会不断执行worker.php脚本吗？该脚本在理论上循环直到队列为空,所以不应该只启动一次吗？守护进程如何决定执行worker.php的频率？这只是一个场景吗？

谢谢!

任何人都可以举例说明生产中如何使用相关ID？

我已经读过它用于请求/响应类型的消息,但我不明白我会在哪里使用它？

我可以想到的一个例子(可能是错误的)是在发布订阅场景中,我可以有5个订阅者,如果我得到5个具有相同关联ID的回复,那么我可以说我的所有订阅者都收到了它.不确定这是否正确使用它.

或者如果我发送一条简单的消息,我可以使用相关性来保证客户端收到它.

还有其他例子吗？

我目前正在开发一个项目,我需要与IBM系统进行交互,该系统通过WebSphere MQ与外部世界进行通信.我需要使用队列以"请求 - 响应"方式查询系统,我将通过队列管理器执行此操作.

但是,我无法理解这在实际工作中是如何运作的.

假设我有相同应用程序的多个实例,它们将消息放入请求队列.该消息得到了CorrelationId和MessageId在离开应用程序和ReplyToQueue属性,都会设置每封邮件,以确保该队列管理器知道哪个队列把到响应.

但是,队列管理器如何操作响应队列？关于响应的时间安排无法保证,那么正确的响应如何回到发出相应请求的应用程序实例？

我一直认为消息队列是一个FIFO队列,必须逐个拾取消息.但是,这意味着实例A可以选择针对例如B的响应.显然,这不是它的工作原理.

然后,当我看API( com.ibm.mq.MQQueue)我看到来接的消息我有机会提供CorrelationId和MessageId请求消息的.这是否意味着当我向队列管理器查询消息(具有这些ID的集合)时,队列管理器会遍历队列中的消息并返回匹配的消息？另一方面,这意味着我们不是在谈论FIFO队列？

我想知道这是否可行.我想从队列中提取任务,并且在将ack发送回RabbitMQ通知工作已完成之前,可能需要3秒或更长(可能)分钟的时间.工作由用户完成,因此这就是处理工作所需时间的原因.

我不想在弹出队列后立即确认消息,因为如果没有收到确认,我希望消息重新排队.任何人都可以给我任何有关如何解决我的问题的见解？

我将在每天超过十亿次请求的高流量环境中使用kafka.每个请求都将与kafka集群建立连接以发送消息.因此,每秒都会有如此多的连接.这可能会导致套接字超时等问题.生产者正在进行所有非持久性连接.因此在这种情况下可能存在套接字超时或端口耗尽问题.

大多数生态系统都在php中,所以我必须使用php库来实现kafka.现在如何有效地使用kafka生产者来缓解这种连接争用？

我想到了一个可以传递消息的守护进程,然后它会批量发送这些消息到kafka集群.另外一方面是连接数量有限.缺点是这种服务的响应延迟会妨碍应用程序.此外,我必须使用一些中间存储来保存消息.

现在我知道有很多极高容量的应用程序/站点使用kafka来直接传输消息.你们中的任何一个人都可以指导我如何解决这些问题吗？在这种情况下持久连接可以帮助吗？在如此高容量的环境中使用kafka-producer的php库本身是个坏主意吗？

我想知道是否有可以在队列之间移动消息的工具或库？目前,我正在做类似下面的事情

public static void ProcessQueueMessage([QueueTrigger("myqueue-poison")] string message, TextWriter log)
{
    CloudStorageAccount storageAccount = CloudStorageAccount.Parse(connString);
    CloudQueueClient queueClient = storageAccount.CreateCloudQueueClient();
    CloudQueue queue = queueClient.GetQueueReference("myqueue");
    queue.CreateIfNotExists();

    var messageData = JsonConvert.SerializeObject(data, new JsonSerializerSettings { ContractResolver = new CamelCasePropertyNamesContractResolver() });
    queue.AddMessage(new CloudQueueMessage(messageData));
}

我们的要求非常简单.向订阅主题的用户发送消息.我们需要我们的消息传递系统能够近乎实时地支持数百万个主题,并且可能支持数百万订阅者.我们的应用程序是用Java构建的.

我们几乎决定使用RabbitMQ,因为社区支持,文档和功能(可能会提供我们需要的一切).但我非常倾向于使用Redis,因为它看起来很有前途和轻量级.老实说,我对Redis作为一个消息传递系统的理解有限,但是看着越来越多的公司使用它作为排队(使用Ruby Resque),我想知道是否有像Resque in Java这样的产品,有什么优势或者使用Redis作为RabbitMQ上的MQ的缺点.

我正在开发一个涉及多个C++程序的项目,每个程序都接受输入并生成输出.数据(数十到数百个字节,可能是JSON)基本上(一个方向)流动(异步),并且程序需要位于LAN周围的不同Linux计算机上.

由于数据只在一个方向流动,我不相信我需要像HTTP那样的事务模型.我认为消息队列模型(fire and forget)最有意义,应该简化每个程序的逻辑.仅注意消息已成功添加到远程队列可能就足够了.

我正在寻找的是如何在C或C++中实现此消息队列的建议.似乎POSIX和Boost消息队列仅限于单个主机,而RabbitMQ似乎具有较弱的C/C++支持,并且MQ4CPP似乎对业务关键角色的支持不足.我错了吗？那么Boost ASIO或ACE或自己编写套接字代码怎么样？我期待着你的建议.

我只想以异步方式接收我的消息!它冻结了我的UI

    public async void ProcessMessages()
    {
        MessageQueue MyMessageQueue = new MessageQueue(@".\private$\MyTransactionalQueue");
        MyMessageQueue.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });

        while (true)
        {
            MessageQueueTransaction MessageQueueTransaction = new MessageQueueTransaction();
            MessageQueueTransaction.Begin();

            ContainError = false;
            ProcessPanel.SetWaiting();

            string Body = MyMessageQueue.Receive(MessageQueueTransaction).Body.ToString();

            //Do some process with body string.

            MessageQueueTransaction.Commit();
        }
    }

我只是像任何常规方法一样调用方法,并且它的工作正常!当我使用BackgroundWorkers而不是async/await时,此代码曾经工作

想法？