ActiveMQ：单个生产者，多个消费者

Question

我有一个使用ActiveMQ的消息队列。web请求用persistency=true将消息放入队列。现在，我有两个消费者，它们都作为单独的会话连接到这个队列。使用者1总是确认消息，但使用者2从不这样做。

JMS队列实现负载平衡器语义。一条消息将被一个使用者接收。如果在发送消息时没有可用的使用者，它将被保留，直到有可以处理消息的使用者可用为止。如果使用者接收到一条消息，但在关闭之前没有确认它，那么该消息将被重新传递给另一个使用者。一个队列可以有许多使用者，在可用的使用者之间消息负载平衡。

我从这里了解到的是，我希望所有消息最终都由消费者1处理，因为它总是确认。由于使用者2不承认，因此消息应该发送到使用者1。

但我注意到以下几点：1。当我提交一个请求时，我只看到每第二个请求来到消费者1。另一个请求不显示，它存储在ActiveMQ中。我想它去了消费者2谁不承认。那么接下来应该是消费者1吗？

我接收/处理消息的代码如下所示：

# --------------------------------------------- MODULE IMPORT ---------------------------------------------------------#
import argparse
import json
import logging
import multiprocessing as mp
import sys

import stomp
from tvpv_portal.services.msgbkr import MsgBkr
from utils import util


# --------------------------------------------- DEVELOPMENT CODE ------------------------------------------------------#
log = logging.getLogger(__name__)


class MessageProcessingListener(stomp.ConnectionListener):
    """This class is responsible for processing (consuming) the messages from ActiveMQ."""

    def __init__(self, conn, cb):
        """Initialization.

        Args:
            conn -- Connection object
            cb   -- Callback function
        """

        self._conn = conn
        self._cb = cb

    def on_error(self, headers, body):
        """When we get an error.

        Args:
            headers -- Message header
            body    -- Message body
        """

        log.error('Received error=%s', body)

    def on_message(self, headers, body):
        """When we receive a message.

        Args:
            headers -- Message header
            body    -- Message body
        """

        log.info('Received message')

        # Deserialize the message.
        item = json.loads(body)

        import pprint
        pprint.pprint(item)

        # TODO: check if msg is to be handled by this SITE. If so, acknowledge and queue it. Otherwise, ignore.

        # Put message into queue via callback (queue.put) function.
        #self._cb(item)

        # TODO: we only send acknowledge if we are supposed to process this message.
        # Send acknowledgement to ActiveMQ indicating message is consumed.
        self._conn.ack(headers['message-id'], headers['subscription'])


def worker(q):
    """Worker to retrieve item from queue and process it.

    Args:
        q -- Queue
    """

    # Run in an infinite loop. Get an item from the queue to process it. We MUST call q.task_done() to indicate
    # that item is processed to prevent deadlock.
    while True:
        try:
            item = q.get()

            # TODO: We will call external script from here to run on Netbatch in the future.
            log.info('Processed message')

        finally:
            q.task_done()


def flash_mq_rst_handler_main():
    """Main entry to the request handler."""

    # Define arguments.
    parser = argparse.ArgumentParser(description='Flash message queue request handler script',
                                     formatter_class=argparse.ArgumentDefaultsHelpFormatter,
                                     add_help=False)

    opts = parser.add_argument_group('Options')
    opts.add_argument('-h', '--help', action='help',
                      help='Show this help message and exit')
    opts.add_argument('--workers', metavar='val', type=int, default=4,
                      help='Number of worker processes')
    opts.add_argument('--log', metavar='file', type=util.get_resolved_abspath, default='flash_mq_rst_handler.log',
                      help='Log file')

    # Parse arguments.
    args = parser.parse_args()

    # Setup logger.
    util.configure_logger(args.log)
    log.info('Command line %s', ' '.join(map(str, sys.argv)))

    # Create a managed queue to store messages retrieved from message queue.
    queue = mp.Manager().JoinableQueue()

    # Instantiate consumer message broker + ensure connection.
    consumer = MsgBkr(producer=False)
    if not consumer.is_connected():
        log.critical('Unable to connect to message queue; please debug')
        sys.exit(1)

    # Register listener with consumer + queue.put as the callback function to trigger when a message is received.
    consumer.set_listener('message_processing_listener', MessageProcessingListener, cb=queue.put)

    # Run in an infinite loop to wait form messages.
    try:
        log.info('Create pool with worker=%d to process messages', args.workers)
        with mp.Pool(processes=args.workers) as pool:
            p = pool.apply_async(worker, (queue,))
            p.get()
    except KeyboardInterrupt:
        pass

    # See MsgBkr. It will close the connection during exit() so we don't have to do it.
    sys.exit(0)


if __name__ == '__main__':
    flash_mq_rst_handler_main()

Answer 1

这个问题可以通过JMS桥实现:https://activemq.apache.org/components/artemis/documentation/1.1.0/jms-bridge.html

能够让它配置创建n+1个队列。源（传入）队列是放置所有消息的地方。根据消息中的某个选择器（比如头中的'some_key':'some_value')，消息可以路由到N个目的地（传出）队列中的一个。然后，每个站点可以侦听特定队列中的消息。同一目标队列中的多个使用者将以循环方式获取消息。