标签: message-queue

阅读常见消息队列系统（例如 RabbitMQ、ActiveMQ）的文档，似乎负载平衡总是由队列系统（循环）或由生产者使用消息上的多个队列或属性来完成。然而，在我们的应用程序中，没有人（甚至消费者）知道处理消息需要多长时间。它可以在几毫秒到几小时之间（它基本上是在任意输入 x 上计算任意函数 f）。因此，负载平衡应该由消费者执行，因为它们仅在其“负载”低于阈值时才接受消息（负载可以是正在执行的作业的数量、CPU 负载等）。理想情况下，生产者将一条消息放入队列中，该消息被“提供”给所有消费者，但只有一个消费者接受并处理该消息。甚至可能发生第一轮没有消费者接受它的情况，它一直留在队列中，直到有一个消费者有空闲资源。所以可能的问题是：对于任何常见的消息队列系统来说，这是否可能以某种方式实现，或者我们是否找错了方向？

我正在用 Celery 替换一些自制代码，但很难复制当前的行为。我期望的行为如下：

• tasks创建新用户时，应使用路由密钥将消息发布到交换器user.created。
• 此消息应触发两个 Celery 任务，即send_user_activate_email和check_spam。

我尝试通过定义user_created带有参数的任务以及和 的ignore_result=True任务来实现此目的。send_user_activate_emailcheck_spam

在我的配置中，我添加了以下路由和队列定义。当消息传递到user_created队列时，它不会传递到其他两个队列。

理想情况下，消息仅传递到send_user_activate_email和check_spam队列。当使用 vanilla RabbitMQ 时，消息被发布到交换器，队列可以绑定到交换器，但 Celery 似乎直接将消息传递到队列。

我将如何在 Celery 中实现上述行为？

CELERY_QUEUES = {
    'user_created': {'binding_key':'user.created', 'exchange': 'tasks', 'exchange_type': 'topic'},
    'send_user_activate_email': {'binding_key':'user.created', 'exchange': 'tasks', 'exchange_type': 'topic'},
    'check_spam': {'binding_key':'user.created', 'exchange': 'tasks', 'exchange_type': 'topic'},
}

CELERY_ROUTES = {
    'user_created': {
        'queue': 'user_created',
        'routing_key': 'user.created',
        'exchange': 'tasks',
        'exchange_type': 'topic',
    },
    'send_user_activate_email': {
        'queue': 'user_created',
        'routing_key': 'user.created', …

man 7 mq_overviewPOSIX 表示“...系统上的消息队列可以使用通常用于文件的命令（例如 ls(1) 和 rm(1)）来查看和操作。” 例如，我能够使用 mqd_t 作为文件描述符进行读取，如下所示：

#include <iostream>
#include <fcntl.h>
#include <mqueue.h>
#include <unistd.h>
#include <stdlib.h>

int main(int argc, char **argv) {
  if (argc != 2) {
    std::cout << "Usage: mqgetinfo </mq_name>\n";
    exit(1);
  }
  mqd_t mqd = mq_open(argv[1], O_RDONLY);

  struct mq_attr attr;
  mq_getattr(mqd, &attr);
  std::cout << argv[1] << " attributes:"
        << "\nflag: " << attr.mq_flags
        << "\nMax # of msgs: " << attr.mq_maxmsg
        << "\nMax msg size: " << attr.mq_msgsize
        << "\nmsgs now in queue: …

如何从WebSphere MQ读取消息而不从队列中删除原始消息？

我有一个 Spring 应用程序，它从 WebSphere MQ 读取消息。阅读后，我有一个处理方法，它将处理从队列中检索的数据。

步骤1：

response = jmsTemplate.receive();
//Message automatically removed from queue.

第2步：

process(response);

process 方法中有可能抛出异常。如果出现异常，我需要将消息保留在队列中。

是否可以？他们有办法仅在用户确认后删除该消息吗？

我尝试添加以下内容：

jmsTemplate.setSessionAcknowledgeMode(javax.jms.Session.CLIENT_ACKNOWLEDGE);

...但该消息仍然被删除。

JmsTemplate创建代码片段：

JndiConnectionFactorySupport connectionFactoryBean = new JndiConnectionFactorySupport();
    connectionFactoryBean.setBindingsDir(this.bindingDir);


        connectionFactoryBean
                .setConnectionFactoryName(connectionFactoryName);
        connectionFactoryBean.afterPropertiesSet();
        jmsTemplate.setConnectionFactory(connectionFactoryBean.getObject());


    JndiDestinationResolver destinationResolver = new JndiDestinationResolver();
    destinationResolver.setJndiTemplate(connectionFactoryBean
            .getJndiTemplate());

    jmsTemplate.setDestinationResolver(destinationResolver);
    jmsTemplate.setReceiveTimeout(20000);
    jmsTemplate.setDefaultDestinationName(this.defaultDestinationName);

            

尝试了jmsTemplate.execute()如下方法：

@SuppressWarnings({ "unused", "unchecked" })
        Message responseMessage = (Message) jmsTemplate.execute(
            new SessionCallback() { 
                public Object doInJms(Session session)
                        throws JMSException {
                    MessageConsumer consumer = session
                    .createConsumer(jmsTemplate.getDestinationResolver().resolveDestinationName(session, "QUEUE_NAME", false));
                    Message response = consumer.receive(1); …

因此，我有一个天蓝色函数作为队列触发器，调用内部托管的 API。

对于如何处理由于有毒以外的问题而无法处理的消息，网上似乎没有明确的答案。

一个例子：

收到我的消息并且该函数尝试调用 API。消息有效负载是正确的并且可以处理，但是 API/服务由于某种原因而关闭（这个时间可能会超过 10 分钟）。目前发生的情况是消息传递计数达到其最大值（10），然后被推送到死信队列，这又会在之后的每条消息中发生。

我需要一种方法来要么不增加交付计数，要么在达到最大值时重置它。或者，我可以放弃对消息的查看锁定，而不增加传递计数，因为我想停止处理队列上的任何消息，直到 API/服务恢复并运行。这样我就可以确保所有可以处理的消息都不会因为服务之间的连接问题而陷入死信。

关于如何实现这一目标有什么想法吗？

我正面临卡夫卡的第一英里问题。每个人都会处理分区等问题，但如何处理第一英里呢？

我的系统由许多应用程序组成，这些应用程序生成分布在节点上的事件。我需要以可靠/故障安全的方式将这些事件传递给一组充当消费者的应用程序。选择的消息系统是 Kafka（由于其日志性质），但它并不是一成不变的。

事件应该尽可能以解耦的即发即忘的方式传播。这意味着生产者应该对可靠地传递消息负全部责任。这意味着生成事件的应用程序根本不必担心事件传递。

生产者的可靠性模式必须考虑：

• 盒子连接中断- 在中断期间生产者根本无法访问网络；Kafka集群因此无法访问
• 框重新启动- 生产者和事件生成应用程序重新启动（独立）；生产者应该保留运行中的消息（在重试、批处理等期间）
• 内部 Kafka 异常-消息大小太大；序列化异常；ETC。

到目前为止，我检查过的库都没有涵盖这些情况。有建议的策略来解决这个问题吗？

我知道Producer 期间存在可重试和不可重试的错误send()。对于那些可重试的内容，库通常会在内部处理所有事情。但是，不可重试以异步回调中的异常结束......

我应该盲目地无限重播这些吗？对于网络中断，它应该可以工作，但是 Kafka 内部错误怎么样 - 比如说消息太大。可能有类似DeadLetterQueue的机制+重播。但是，如何处理消息计数......

关于持久性 - 轻量级数据库后端应该可以解决这个问题。只需创建一个持久队列，然后删除那些已经发送/确认的队列。然而，恐怕如果这么简单的话，它很早以前就已经在标准 Kafka 库中实现了。性能可能会下降。

看到像KAFKA-3686或KAFKA-1955这样的东西让我有点担心。

提前致谢。

我试图了解分布式系统中队列的用例是什么。
以及它如何扩展以及如何确保它不是系统中的单点故障？

任何直接回答或对文档的引用都值得赞赏。

1. 用例：
   我知道队列是一个消息系统。并且它解耦了彼此通信的系统。但是，这是使用队列的唯一目的吗？

2. 可扩展性：
   队列如何针对大量数据进行扩展？既读又写。

3. 可靠性：队列如何不成为系统中的单点故障？队列是否进行复制（类似于数据存储）？

我的问题没有指定到任何特定的队列服务器，如 Kafka 或 JMS。就一般而言。

我需要确保添加到队列中的同一作业在一段时间内不会重复。

是否值得D/M/Y-HH:M在我的独特中包含部分时间戳（即）jobId，以便仅在不在同一分钟内时才进行处理？

如果在 处添加一项作业，在 处添加12:01另一项作业，它仍然会重复12:09\xe2\x80\x93，它仍然会重复，或者 Bull 是否有更好的方法来做到这一点？

我想知道我对消息总线和消息队列工作原理的理解是否正确。

首先，我需要明确命名，服务总线与消息总线可以互换使用吗？这是一种发布者-订阅者类型的系统，其中消息被添加到任意数量的发布者的消息集合中，并且任意数量的订阅者都可以从中读取消息，到目前为止我是对的吗？

P1 ---                                         /``````S1
       \________ Service Bus Middleware ------+------ S2
       /           MESSAGE-COLLECTION          \______S3
P2 ---

我不明白的是

1. 订阅者如何知道它感兴趣的消息，我的意思是它显然订阅了它，但是它如何知道它应该订阅哪条消息？，它在哪里看到消息列表，它如何使用它？通过 API 或如何？

2. 订阅者如何接收消息？

3. 何时从 MESSAGE-COLLECTION 中删除消息？我可以想象的是，为每条消息保留一些计数器，该计数器代表订阅者的总数，一旦一个订阅者成功处理该消息，该计数器就会递减。

消息队列也称为消息代理，是一种推拉类型的系统。有任意数量的生产者和任意数量的消费者。每个生产者为每个消费者创建一个队列，并向其提供消息。

             --- Message Queue 1 ---- C1
           /  
P1 ------ +
           \ 
             --- Message Queue 2 ---- C2


P2 ------ + --- Message Queue 1 ---- C1

在这种情况下，一旦消费者成功处理该消息，该消息就会被删除。我对消息队列工作原理的理解是否正确？

我不确定到底是什么的另一个概念是event hub。

我的分布式应用程序由许多将任务推送到多个 FIFO 队列的生产者以及每个队列的多个消费者组成。所有这些组件都位于单个节点上，因此不涉及网络。

Python 的内置功能完全支持这种模式multiprocessing.Queue，但是当我扩展应用程序时，队列实现似乎是一个瓶颈。我没有发送大量数据，因此内存共享并不能解决问题。我需要的是每秒快速保证传送 10^4-10^5 条小消息。每条消息大约为 100 字节。

我是快速分布式计算世界的新手，我对大量的选择感到非常困惑。有RabbitMQ、Redis、Kafka等。

ZeroMQ 是一个更加专注和紧凑的替代方案，它还有 nanomsg 和 nng 等后继者。此外，在没有代理的情况下，实现诸如具有保证交付的多对多队列之类的东西似乎并不简单。

如果有人能向我指出一种使用更快的框架之一执行此类操作的“标准”方法，我将非常感激。