java - 一旦消息数量超过并发消费者数量，消费者就不会分配给消息

Question

我想了解消息被消费后，消费者如何为并发消费者工作。消费者 bean 是被丢弃并创建新的，还是被重用？我可以看到，在设置rabbit mq服务器后，当我发送消息时，会实例化一个消费者bean。

我的问题是，一旦消息数量等于并发消费者数量，消息监听器中的消费者对于下一条消息将为空。我可以在日志中看到之前消息的 Ack/Nack，并且预取设置为 1。

XML 配置:

<rabbit:admin connection-factory="cachingConnectionFactory" />

<rabbit:template id="template" connection-factory="cachingConnectionFactory"
    retry-template="retryTemplate" />

<!-- CachingConnectionFactory -->
<bean id="cachingConnectionFactory"
    class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
    <constructor-arg ref="clientConnectionFactorySSL" />
    <property name="username" value="${QM_USERNAME}" />
    <property name="password" value="${QM_PASSWORD}" />
    <property name="virtualHost" value="${QM_VIRTUALHOST}" />
    <property name="host" value="${QM_HOST}" />
    <property name="port" value="${QM_PORT}" />
</bean>

<!-- SimpleRabbitListenerContainerFactory -->
        <bean id="rabbitListenerContainerFactory"
            class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
            <property name="connectionFactory">
                <ref bean="cachingConnectionFactory" />
            </property>
            <property name="concurrentConsumers" value="5" />
            <property name="maxConcurrentConsumers" value="8" />
             <property name="startConsumerMinInterval" value="3000" />
        </bean>

<!-- Message listener container -->
        <bean id="springListenerContainerRABC"          class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
            <property name="connectionFactory" ref="cachingConnectionFactory" />
            <property name="queues" ref="rabbitDestinationRABC" />
            <property name="messageListener" ref="rabbitMessageListenerRABC" />
            **<property name="concurrentConsumers" value="4" />**
            <property name="maxConcurrentConsumers" value="5" />
        </bean>

<!-- Message Listener -->
    <bean id="rabbitMessageListenerRABC" class="com.queue.rabbitmessaging.RabbitMQListener">
        **<property name="consumerType" value="XYZConsumer" />**
        <property name="responseSender" ref="XYZResponseSender" />
        <property name="taskExecutor" ref="taskExecutor" />
        <property name="encoding" value="UTF-16BE" />
    </bean>

<!-- Consumers -->
<bean id="XYZConsumer" class="com.msg.consumer.XYZConsumer"
    scope="prototype" />

以下是我在日志中看到的内容:

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - **Received message for consumer: XYZConsumer**

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Message :: test1

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Received message for consumer: XYZConsumer

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Message :: test2

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Received message for consumer: XYZConsumer

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Message :: test3

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Received message for consumer: XYZConsumer

00:17:49.725 [springListenerContainerRABC-1] INFO RabbitMQListener - Message :: test4

[org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1] INFO RabbitMQListener - **Received message for consumer: null**

[org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1] INFO RabbitMQListener - Message :: test5

消费者代码如下:

@EnableRabbit
@Service 
public class RabbitMQListener implements MessageListener {
    @Autowired
    private ApplicationContext applicationContext;

    protected static Logger logger = LoggerFactory.getLogger(RabbitMQListener.class);

    private TaskExecutor taskExecutor;
    private RabbitSender responseSender;
    private String consumerType;
    private String encoding;

     public TaskExecutor getTaskExecutor() {
        return taskExecutor;
    }

    public void setTaskExecutor(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    public void setResponseSender(RabbitSender responseSender) {
        this.responseSender = responseSender;
    }

    public void setConsumerType(String consumerType) {
        this.consumerType = consumerType;
    }

    public void setEncoding(String encoding) {
        this.encoding = encoding;
    }

    @Override
    @RabbitListener(containerFactory = Constants.CONTAINER_FACTORY, queues = "#{'${QM_QUEUES}'.split(',')}")
    public void onMessage(final Message arg0) {

        if (arg0 != null) {
            try {
                String message = new String(arg0.getBody());
                logger.info("Message Properties class: " + arg0.getClass());
                logger.info("Message Properties: " + arg0.getMessageProperties());
                logger.info("Received message for consumer: " + this.consumerType);
                logger.info("Message :: " + message);
                Object bean = applicationContext.getBean(this.consumerType);
                if (bean instanceof Consumer) {
                    logger.debug("Processing message for consumer: " + this.consumerType);
                    Consumer consumer = (Consumer)bean;
                    consumer.setMessage(message);
                    consumer.setResponseSender(responseSender);
                    logger.debug("Executing consumer: " + this.consumerType);
                    taskExecutor.execute(consumer);
                } else {
                    logger.info("Could not process message due to invalid consumer.");
                }

            } 
            catch (AmqpException e1) {
                logger.error("Spring AMQP error processing message.", e1.getMessage());
            }
            catch (Exception e2) {
                logger.error("General error processing message.", e2.getMessage());
            }
        }
        else {
            logger.debug("message received is NULL");
        }

    }


}

Answer 1

所描述的行为与收到消息后不确认消息一致。 RabbitMQ 可以配置为 automatically acknowledge交货时的一条消息。如果这种情况没有发生，或者没有配置，RabbitMQ 将假定消费者正忙，并且不会向其传递额外的消息。您必须在成功处理消息后发送 ack 才能接收队列中的下一条消息。