【RabbitMQ分析】01 SimpleMessageListenerContainer原理分析

//后置处理器，接收到的消息都添加了Header请求头
container.setAfterReceivePostProcessors(message -> {
    message.getMessageProperties().getHeaders().put("desc",10);
    return message;
});
container.setMessageListener((MessageListener) message -> {
    System.out.println("====接收到消息=====");
    System.out.println(message.getMessageProperties());
    System.out.println(new String(message.getBody()));
});

container.setConsumerTagStrategy(queue -> "order_queue_"+(++count));
//设置消费者的Arguments
Map<String, Object> args = new HashMap<>();
args.put("module","订单模块");
args.put("fun","发送消息");
container.setConsumerArguments(args);

  @Bean
   public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
       SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
       container.setConnectionFactory(connectionFactory);
       container.setQueueNames("zhihao.miao.order");
       container.setConcurrentConsumers(5);
       container.setMaxConcurrentConsumers(10);
       container.setMessageListener((MessageListener) message -> {
           System.out.println("====接收到消息=====");
           System.out.println(message.getMessageProperties());
           System.out.println(new String(message.getBody()));
       });
       return container;
   }

public void start() {
    //如果已启动，则什么也不执行，直接退出
 if (isRunning()) {
  return;
 }
    //initialized是否执行初始化，没有则执行afterPropertiesSet()方法进行初始化，执行完成后initialized设置成true
 if (!this.initialized) {
  synchronized (this.lifecycleMonitor) {
   if (!this.initialized) {
    afterPropertiesSet();
   }
  }
 }
 try {
  logger.debug("Starting Rabbit listener container.");
        //验证RabbitAdmin，mismatchedQueuesFatal=true时，spring context中RabbitAdmin数量不能大于1
  configureAdminIfNeeded();
        //执行RabbitAdmin#initialize方法，spring context中注入的exchanges, queues and bindings执行声明式创建
        /*
        总结一下，我们发现，要想自动创建队列，SimpleMessageListenerContainer需要满足这么两点:
         mismatchedQueuesFatal属性设置为true
         autoDeclare属性也设置为true
        */
  checkMismatchedQueues();
        //启动核心
  doStart();
 }
 catch (Exception ex) {
  throw convertRabbitAccessException(ex);
 }
 finally {
  this.lazyLoad = false;
 }
}

protected void doStart() {
 Assert.state(!this.consumerBatchEnabled || getMessageListener() instanceof BatchMessageListener
   || getMessageListener() instanceof ChannelAwareBatchMessageListener,
   "When setting 'consumerBatchEnabled' to true, the listener must support batching");
 //如果MessageListener是ListenerContainerAware，则进行expectedQueueNames校验
    checkListenerContainerAware();
    //调用父类doStart()方法，主要是active和running都设置成true
 super.doStart();
 synchronized (this.consumersMonitor) {
  if (this.consumers != null) {
   throw new IllegalStateException("A stopped container should not have consumers");
  }
        /*
        创建BlockingQueueConsumer类型consumer，每个concurrentConsumers并发对应创建一个对象，并存储到Set<BlockingQueueConsumer> consumers集合中，
        返回值就是创建consumer对象个数，具体创建逻辑见：SimpleMessageListenerContainer#createBlockingQueueConsumer，主要注意下prefetchCount计算：
        int actualPrefetchCount = getPrefetchCount() > this.batchSize ? getPrefetchCount() : this.batchSize;即如果prefetchCount大于batchSize，则其就是实际值，否则prefetchCount等于batchSize值
        */
  int newConsumers = initializeConsumers();
  if (this.consumers == null) {
   logger.info("Consumers were initialized and then cleared " +
     "(presumably the container was stopped concurrently)");
   return;
  }
  if (newConsumers <= 0) {
   if (logger.isInfoEnabled()) {
    logger.info("Consumers are already running");
   }
   return;
  }
        /*
        每个并发对应一个BlockingQueueConsumer对象，这里将每个BlockingQueueConsumer对象封装成AsyncMessageProcessingConsumer，这样可以丢到线程池中异步执行
        */
  Set<AsyncMessageProcessingConsumer> processors = new HashSet<AsyncMessageProcessingConsumer>();
  for (BlockingQueueConsumer consumer : this.consumers) {
            //将BlockingQueueConsumer对象封装成AsyncMessageProcessingConsumer进行异步执行
   AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer);
            //存储到processors集合中
   processors.add(processor);
            //将AsyncMessageProcessingConsumer丢到线程池中执行
   getTaskExecutor().execute(processor);
   if (getApplicationEventPublisher() != null) {
                //事件发送
    getApplicationEventPublisher().publishEvent(new AsyncConsumerStartedEvent(this, consumer));
   }
  }
        //判断启动过程中是否存在异常
  waitForConsumersToStart(processors);
 }
}

//BlockingQueueConsumer.getQueueCount() < 1，表示当前consumer没有设置任何监听队列，则没必要启动
if (this.consumer.getQueueCount() < 1) {
  if (logger.isDebugEnabled()) {
   logger.debug("Consumer stopping; no queues for " + this.consumer);
  }
  SimpleMessageListenerContainer.this.cancellationLock.release(this.consumer);
  if (getApplicationEventPublisher() != null) {
   getApplicationEventPublisher().publishEvent(
     new AsyncConsumerStoppedEvent(SimpleMessageListenerContainer.this, this.consumer));
  }
  this.start.countDown();
  return;
}

try {
 initialize();
 while (isActive(this.consumer) || this.consumer.hasDelivery() || !this.consumer.cancelled()) {
  mainLoop();
 }
}
catch (InterruptedException e) {
    省略。。。
}
catch (QueuesNotAvailableException ex) {
 省略。。。
}
省略。。。

protected void processDelivery(Command command, Basic.Deliver method) {
    Basic.Deliver m = method;

    //根据Deliver的consumerTag获取到InternalConsumer对象，因为一个Channel上可能存在多个consumer，需要找到Broker是针对哪个consumer进行的响应
    Consumer callback = _consumers.get(m.getConsumerTag());
    if (callback == null) {
        if (defaultConsumer == null) {
            // No handler set. We should blow up as this message
            // needs acking, just dropping it is not enough. See bug
            // 22587 for discussion.
            throw new IllegalStateException("Unsolicited delivery -" +
                    " see Channel.setDefaultConsumer to handle this" +
                    " case.");
        }
        else {
            callback = defaultConsumer;
        }
    }

    Envelope envelope = new Envelope(m.getDeliveryTag(),
                                     m.getRedelivered(),
                                     m.getExchange(),
                                     m.getRoutingKey());
    try {
        // call metricsCollector before the dispatching (which is async anyway)
        // this way, the message is inside the stats before it is handled
        // in case a manual ack in the callback, the stats will be able to record the ack
        metricsCollector.consumedMessage(this, m.getDeliveryTag(), m.getConsumerTag());
        this.dispatcher.handleDelivery(callback,
                                       m.getConsumerTag(),
                                       envelope,
                                       (BasicProperties) command.getContentHeader(),
                                       command.getContentBody());
    } catch (WorkPoolFullException e) {
        // couldn't enqueue in work pool, propagating
        throw e;
    } catch (Throwable ex) {
        getConnection().getExceptionHandler().handleConsumerException(this,
            ex,
            callback,
            m.getConsumerTag(),
            "handleDelivery");
   }
}

Envelope envelope = new Envelope(m.getDeliveryTag(),
                                         m.getRedelivered(),
                                         m.getExchange(),
                                         m.getRoutingKey());

this.dispatcher.handleDelivery(callback,
                                           m.getConsumerTag(),
                                           envelope,
                                           (BasicProperties) command.getContentHeader(),
                                           command.getContentBody());

BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body, this.queueName));

private void mainLoop() throws Exception { // NOSONAR Exception
 try {
  boolean receivedOk = receiveAndExecute(this.consumer); // At least one message received
  if (SimpleMessageListenerContainer.this.maxConcurrentConsumers != null) {
   checkAdjust(receivedOk);
  }
  long idleEventInterval = getIdleEventInterval();
  if (idleEventInterval > 0) {
   if (receivedOk) {
    updateLastReceive();
   }
   else {
    long now = System.currentTimeMillis();
    long lastAlertAt = SimpleMessageListenerContainer.this.lastNoMessageAlert.get();
    long lastReceive = getLastReceive();
    if (now > lastReceive + idleEventInterval
      && now > lastAlertAt + idleEventInterval
      && SimpleMessageListenerContainer.this.lastNoMessageAlert
      .compareAndSet(lastAlertAt, now)) {
     publishIdleContainerEvent(now - lastReceive);
    }
   }
  }
 }
 catch (ListenerExecutionFailedException ex) {
  // Continue to process, otherwise re-throw
  if (ex.getCause() instanceof NoSuchMethodException) {
   throw new FatalListenerExecutionException("Invalid listener", ex);
  }
 }
 catch (AmqpRejectAndDontRequeueException rejectEx) {
  /*
   *  These will normally be wrapped by an LEFE if thrown by the
   *  listener, but we will also honor it if thrown by an
  *  error handler.
  */
 }
}

private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Exception { //NOSONAR

 Channel channel = consumer.getChannel();

 List<Message> messages = null;
 long deliveryTag = 0;

    //batchSize默认是1，用于指定一次从queue中提取消息数量
 for (int i = 0; i < this.batchSize; i++) {
  logger.trace("Waiting for message from consumer.");
  Message message = consumer.nextMessage(this.receiveTimeout);
  if (message == null) {
   break;
  }
        //是否批处理
  if (this.consumerBatchEnabled) {
   Collection<MessagePostProcessor> afterReceivePostProcessors = getAfterReceivePostProcessors();
   if (afterReceivePostProcessors != null) {
    Message original = message;
    deliveryTag = message.getMessageProperties().getDeliveryTag();
    for (MessagePostProcessor processor : getAfterReceivePostProcessors()) {
     message = processor.postProcessMessage(message);
     if (message == null) {
      channel.basicAck(deliveryTag, false);
      if (this.logger.isDebugEnabled()) {
       this.logger.debug(
         "Message Post Processor returned 'null', discarding message " + original);
      }
      break;
     }
    }
   }
   if (message != null) {
    if (messages == null) {
     messages = new ArrayList<>(this.batchSize);
    }
    if (isDeBatchingEnabled() && getBatchingStrategy().canDebatch(message.getMessageProperties())) {
     final List<Message> messageList = messages;
     getBatchingStrategy().deBatch(message, fragment -> messageList.add(fragment));
    }
    else {
     messages.add(message);
    }
   }
  }
  else {
   messages = debatch(message);
   if (messages != null) {
    break;
   }
   try {
                //执行MessageListener
    executeListener(channel, message);
   }
   catch (ImmediateAcknowledgeAmqpException e) {
    if (this.logger.isDebugEnabled()) {
     this.logger.debug("User requested ack for failed delivery '"
       + e.getMessage() + "': "
       + message.getMessageProperties().getDeliveryTag());
    }
    break;
   }
   catch (Exception ex) {
    if (causeChainHasImmediateAcknowledgeAmqpException(ex)) {
     if (this.logger.isDebugEnabled()) {
      this.logger.debug("User requested ack for failed delivery: "
        + message.getMessageProperties().getDeliveryTag());
     }
     break;
    }
    if (getTransactionManager() != null) {
     if (getTransactionAttribute().rollbackOn(ex)) {
      RabbitResourceHolder resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager
        .getResource(getConnectionFactory());
      if (resourceHolder != null) {
       consumer.clearDeliveryTags();
      }
      else {
       /*
       * If we don't actually have a transaction, we have to roll back
       * manually. See prepareHolderForRollback().
       */
       consumer.rollbackOnExceptionIfNecessary(ex);
      }
       throw ex; // encompassing transaction will handle the rollback.
      }
      else {
       if (this.logger.isDebugEnabled()) {
        this.logger.debug("No rollback for " + ex);
       }
       break;
      }
     }
     else {
      consumer.rollbackOnExceptionIfNecessary(ex);
      throw ex;
     }
    }
   }
  }
  if (messages != null) {
   executeWithList(channel, messages, deliveryTag, consumer);
  }

  return consumer.commitIfNecessary(isChannelLocallyTransacted());

 }

container.setQueueNames("test01", "test02");
container.setConcurrentConsumers(3);
container.setConsumerTagStrategy(queue -> "consumer idx:"+consumerIdx.getAndIncrement());