devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spring-RabbitMQ源码解读_org.springframework.amqp.rabbit.rabbitlistenerendp

作者：IT小白 | 2024-04-08 00:05:02

赞

踩

org.springframework.amqp.rabbit.rabbitlistenerendpointcontainer

前言

1.仓库地址

https://gitee.com/JiuLongBingShi/spring-rabbit-king.git

2.maven依赖

  1. <dependency>
  2.     <groupId>com.king.springboot</groupId>
  3.     <artifactId>king-rabbit</artifactId>
  4.     <version>2.1.0</version>
  5. </dependency>

一、知识点  
1、RabbitListenerEndpoint ：
Endpoint为终端，像电脑、手机都是终端，他们都可以接受外部信息并响应，如手机来短信了就有提示。这里也用了终端的概念，例如：被@RabbitListener注解修饰方法也有终端的特点 可以接受外部信息并响应。

  1. /*
  2.  * Copyright 2002-2018 the original author or authors.
  3.  *
  4.  * Licensed under the Apache License, Version 2.0 (the "License");
  5.  * you may not use this file except in compliance with the License.
  6.  * You may obtain a copy of the License at
  7.  *
  8.  *      https://www.apache.org/licenses/LICENSE-2.0
  9.  *
  10.  * Unless required by applicable law or agreed to in writing, software
  11.  * distributed under the License is distributed on an "AS IS" BASIS,
  12.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13.  * See the License for the specific language governing permissions and
  14.  * limitations under the License.
  15.  */
  17. package org.springframework.amqp.rabbit.listener;
  18.  
  19. import org.springframework.amqp.support.converter.MessageConverter;
  20. import org.springframework.lang.Nullable;
  21.  
  22. /**
  23.  * Model for a Rabbit listener endpoint. Can be used against a
  24.  * {@link org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer
  25.  * RabbitListenerConfigurer} to register endpoints programmatically.
  26.  *
  27.  * @author Stephane Nicoll
  28.  * @author Gary Russell
  29.  * @since 1.4
  30.  */
  31. public interface RabbitListenerEndpoint {
  32.  
  33. 	/**
  34. 	 * @return the id of this endpoint. The id can be further qualified
  35. 	 * when the endpoint is resolved against its actual listener
  36. 	 * container.
  37. 	 * @see RabbitListenerContainerFactory#createListenerContainer
  38. 	 */
  39. 	String getId();
  40.  
  41. 	/**
  42. 	 * @return the group of this endpoint or null if not in a group.
  43. 	 * @since 1.5
  44. 	 */
  45. 	String getGroup();
  46.  
  47. 	/**
  48. 	 * @return the concurrency of this endpoint.
  49. 	 * @since 2.0
  50. 	 */
  51. 	String getConcurrency();
  52.  
  53. 	/**
  54. 	 * Override of the default autoStartup property.
  55. 	 * @return the autoStartup.
  56. 	 * @since 2.0
  57. 	 */
  58. 	Boolean getAutoStartup();
  59.  
  60. 	/**
  61. 	 * Setup the specified message listener container with the model
  62. 	 * defined by this endpoint.
  63. 	 * <p>This endpoint must provide the requested missing option(s) of
  64. 	 * the specified container to make it usable. Usually, this is about
  65. 	 * setting the {@code queues} and the {@code messageListener} to
  66. 	 * use but an implementation may override any default setting that
  67. 	 * was already set.
  68. 	 * @param listenerContainer the listener container to configure
  69. 	 */
  70. 	void setupListenerContainer(MessageListenerContainer listenerContainer);
  71.  
  72. 	/**
  73. 	 * The preferred way for a container factory to pass a message converter
  74. 	 * to the endpoint's adapter.
  75. 	 * @param converter the converter.
  76. 	 * @since 2.0.8
  77. 	 */
  78. 	default void setMessageConverter(MessageConverter converter) {
  79. 		// NOSONAR
  80. 	}
  82. 	/**
  83. 	 * Used by the container factory to check if this endpoint supports the
  84. 	 * preferred way for a container factory to pass a message converter
  85. 	 * to the endpoint's adapter. If null is returned, the factory will
  86. 	 * fall back to the legacy method of passing the converter via the
  87. 	 * container.
  88. 	 * @return the converter.
  89. 	 * @since 2.0.8
  90. 	 */
  91. 	@Nullable
  92. 	default MessageConverter getMessageConverter() {
  93. 		return null;
  94. 	}
  95.  
  96. }


这里的终端有很多种，刚才说到 『被@RabbitListener注解修饰方法也是终端』 就是 MethodRabbitListenerEnpoint 方法终端。

接口方法中 void setupListenerContainer(MessageListenerContainer listenerContainer) 方法，中的MessageListenerContainer就是用来接收并处理消息的。

在抽象类中AbstractRabbitListenerEndpoint对setupListenerContainer方法进了实现：

  1. @Override
  2. 	public void setupListenerContainer(MessageListenerContainer listenerContainer) {
  3. 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer;
  4.  
  5. 		boolean queuesEmpty = getQueues().isEmpty();
  6. 		boolean queueNamesEmpty = getQueueNames().isEmpty();
  7. 		if (!queuesEmpty && !queueNamesEmpty) {
  8. 			throw new IllegalStateException("Queues or queue names must be provided but not both for " + this);
  9. 		}
  10. 		if (queuesEmpty) {
  11. 			Collection<String> names = getQueueNames();
  12. 			container.setQueueNames(names.toArray(new String[names.size()]));
  13. 		}
  14. 		else {
  15. 			Collection<Queue> instances = getQueues();
  16. 			container.setQueues(instances.toArray(new Queue[instances.size()]));
  17. 		}
  18.  
  19. 		container.setExclusive(isExclusive());
  20. 		if (getPriority() != null) {
  21. 			Map<String, Object> args = new HashMap<String, Object>();
  22. 			args.put("x-priority", getPriority());
  23. 			container.setConsumerArguments(args);
  24. 		}
  25.  
  26. 		if (getAdmin() != null) {
  27. 			container.setAmqpAdmin(getAdmin());
  28. 		}
  29.         //入口
  30. 		setupMessageListener(listenerContainer);
  31. 	}
  1. private void setupMessageListener(MessageListenerContainer container) {
  2. 		//
  3. 		//MessageListenerContainer为容器，容器是要装东西的，这里显然是装MessageListener。
  4. 		//createMessageListener(container)为抽象方法，用于子类覆盖。
  5. 		MessageListener messageListener = createMessageListener(container);
  6. 		Assert.state(messageListener != null, () -> "Endpoint [" + this + "] must provide a non null message listener");
  7. 		container.setupMessageListener(messageListener);
  8. 	}


    


    
   
子类MethodRabbitListenerEndpoint中覆盖了createMessageListener。

  1. @Override
  2. 	protected MessagingMessageListenerAdapter createMessageListener(MessageListenerContainer container) {
  3. 		Assert.state(this.messageHandlerMethodFactory != null,
  4. 				"Could not create message listener - MessageHandlerMethodFactory not set");
  5. 		MessagingMessageListenerAdapter messageListener = createMessageListenerInstance();
  6. 		messageListener.setHandlerAdapter(configureListenerAdapter(messageListener));
  7. 		String replyToAddress = getDefaultReplyToAddress();
  8. 		if (replyToAddress != null) {
  9. 			messageListener.setResponseAddress(replyToAddress);
  10. 		}
  11. 		MessageConverter messageConverter = getMessageConverter();
  12. 		if (messageConverter == null) {
  13. 			// fall back to the legacy converter holder in the container
  14. 			messageConverter = container.getMessageConverter();
  15. 		}
  16. 		if (messageConverter != null) {
  17. 			messageListener.setMessageConverter(messageConverter);
  18. 		}
  19. 		if (getBeanResolver() != null) {
  20. 			messageListener.setBeanResolver(getBeanResolver());
  21. 		}
  22. 		return messageListener;
  23. 	}


    
    

2、RabbitListenerContainerFactory
RabbitListenerContainerFactory为：rabbit 监听器容器工厂。既然为 监听器容器工厂 那一定是生产监听器容器。

监听器容器工厂：

  1. /*
  2.  * Copyright 2002-2018 the original author or authors.
  3.  *
  4.  * Licensed under the Apache License, Version 2.0 (the "License");
  5.  * you may not use this file except in compliance with the License.
  6.  * You may obtain a copy of the License at
  7.  *
  8.  *      https://www.apache.org/licenses/LICENSE-2.0
  9.  *
  10.  * Unless required by applicable law or agreed to in writing, software
  11.  * distributed under the License is distributed on an "AS IS" BASIS,
  12.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13.  * See the License for the specific language governing permissions and
  14.  * limitations under the License.
  15.  */
  17. package org.springframework.amqp.rabbit.listener;
  18.  
  19. import org.springframework.lang.Nullable;
  20.  
  21. /**
  22.  * Factory of {@link MessageListenerContainer}s.
  23.  * @param <C> the container type.
  24.  * @author Stephane Nicoll
  25.  * @author Gary Russell
  26.  * @since 1.4
  27.  * @see RabbitListenerEndpoint
  28.  */
  29. @FunctionalInterface
  30. public interface RabbitListenerContainerFactory<C extends MessageListenerContainer> {
  31.  
  32. 	/**
  33. 	 * Create a {@link MessageListenerContainer} for the given
  34. 	 * {@link RabbitListenerEndpoint}.
  35. 	 * @param endpoint the endpoint to configure.
  36. 	 * @return the created container.
  37. 	 */
  38. 	C createListenerContainer(@Nullable RabbitListenerEndpoint endpoint);
  39.  
  40. 	/**
  41. 	 * Create a {@link MessageListenerContainer} with no
  42. 	 * {@link org.springframework.amqp.core.MessageListener} or queues; the listener must
  43. 	 * be added later before the container is started.
  44. 	 * @return the created container.
  45. 	 * @since 2.1.
  46. 	 */
  47. 	default C createListenerContainer() {
  48. 		return createListenerContainer(null);
  49. 	}
  50.  
  51. }


监听器容器： 

  1. /*
  2.  * Copyright 2014-2019 the original author or authors.
  3.  *
  4.  * Licensed under the Apache License, Version 2.0 (the "License");
  5.  * you may not use this file except in compliance with the License.
  6.  * You may obtain a copy of the License at
  7.  *
  8.  *      https://www.apache.org/licenses/LICENSE-2.0
  9.  *
  10.  * Unless required by applicable law or agreed to in writing, software
  11.  * distributed under the License is distributed on an "AS IS" BASIS,
  12.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13.  * See the License for the specific language governing permissions and
  14.  * limitations under the License.
  15.  */
  17. package org.springframework.amqp.rabbit.listener;
  18.  
  19. import org.springframework.amqp.core.MessageListener;
  20. import org.springframework.amqp.support.converter.MessageConverter;
  21. import org.springframework.context.SmartLifecycle;
  22.  
  23. /**
  24.  * Internal abstraction used by the framework representing a message
  25.  * listener container. Not meant to be implemented externally.
  26.  *
  27.  * @author Stephane Nicoll
  28.  * @author Gary Russell
  29.  * @since 1.4
  30.  */
  31. public interface MessageListenerContainer extends SmartLifecycle {
  32.  
  33. 	/**
  34. 	 * Setup the message listener to use. Throws an {@link IllegalArgumentException}
  35. 	 * if that message listener type is not supported.
  36. 	 * @param messageListener the {@code object} to wrapped to the {@code MessageListener}.
  37. 	 */
  38. 	void setupMessageListener(MessageListener messageListener);
  39.  
  40. 	/**
  41. 	 * @return the {@link MessageConverter} that can be used to
  42. 	 * convert {@link org.springframework.amqp.core.Message}, if any.
  43. 	 * @deprecated - this converter is not used by the container; it was only
  44. 	 * used to configure the converter for a {@code @RabbitListener} adapter.
  45. 	 * That is now handled differently. If you are manually creating a listener
  46. 	 * container, the converter must be configured in a listener adapter (if
  47. 	 * present).
  48. 	 */
  49. 	@Deprecated
  50. 	MessageConverter getMessageConverter();
  51.  
  52. 	/**
  53. 	 * Do not check for missing or mismatched queues during startup. Used for lazily
  54. 	 * loaded message listener containers to avoid a deadlock when starting such
  55. 	 * containers. Applications lazily loading containers should verify the queue
  56. 	 * configuration before loading the container bean.
  57. 	 * @since 2.1.5
  58. 	 */
  59. 	default void lazyLoad() {
  60. 		// no-op
  61. 	}
  62.  
  63. }


 监听器容器工厂：

 在其抽象子类AbstractRabbitListenerContainerFactory中对 createListenerContainer 方法已做了实现。

我常用的是其子类SimpleRabbitListenerContainerFactory

  1. public class SimpleRabbitListenerContainerFactory
  2. 		extends AbstractRabbitListenerContainerFactory<SimpleMessageListenerContainer> {
  3.         //...
  4. }


SimpleRabbitListenerContainerFactory工厂生产的是SimpleMessageListenerContainer

  1. public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractMessageListenerContainer>
  2. 		implements RabbitListenerContainerFactory<C>, ApplicationContextAware, ApplicationEventPublisherAware {
  3. //...
  4.  
  5.  
  6. //在使用SimpleRabbitListenerContainerFactory时，下面的 C 就是 SimpleMessageListenerContainer
  7.  
  8. 	@SuppressWarnings("deprecation")
  9. 	@Override
  10. 	public C createListenerContainer(RabbitListenerEndpoint endpoint) {
  11. //创建SimpleMessageListenerContainer
  12.         
  13. 		C instance = createContainerInstance();
  14.  
  15. 		JavaUtils javaUtils = JavaUtils.INSTANCE.acceptIfNotNull(this.connectionFactory, instance::setConnectionFactory)
  16. 			.acceptIfNotNull(this.errorHandler, instance::setErrorHandler);
  17. 		if (this.messageConverter != null) {
  18. 			if (endpoint != null) {
  19. 				endpoint.setMessageConverter(this.messageConverter);
  20. 				if (endpoint.getMessageConverter() == null) {
  21. 					instance.setMessageConverter(this.messageConverter);
  22. 				}
  23. 			}
  24. 			else {
  25. 				instance.setMessageConverter(this.messageConverter);
  26. 			}
  27. 		}
  28. 		javaUtils
  29. 			.acceptIfNotNull(this.acknowledgeMode, instance::setAcknowledgeMode)
  30. 			.acceptIfNotNull(this.channelTransacted, instance::setChannelTransacted)
  31. 			.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
  32. 			.acceptIfNotNull(this.taskExecutor, instance::setTaskExecutor)
  33. 			.acceptIfNotNull(this.transactionManager, instance::setTransactionManager)
  34. 			.acceptIfNotNull(this.prefetchCount, instance::setPrefetchCount)
  35. 			.acceptIfNotNull(this.defaultRequeueRejected, instance::setDefaultRequeueRejected)
  36. 			.acceptIfNotNull(this.adviceChain, instance::setAdviceChain)
  37. 			.acceptIfNotNull(this.recoveryBackOff, instance::setRecoveryBackOff)
  38. 			.acceptIfNotNull(this.mismatchedQueuesFatal, instance::setMismatchedQueuesFatal)
  39. 			.acceptIfNotNull(this.missingQueuesFatal, instance::setMissingQueuesFatal)
  40. 			.acceptIfNotNull(this.consumerTagStrategy, instance::setConsumerTagStrategy)
  41. 			.acceptIfNotNull(this.idleEventInterval, instance::setIdleEventInterval)
  42. 			.acceptIfNotNull(this.failedDeclarationRetryInterval, instance::setFailedDeclarationRetryInterval)
  43. 			.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
  44. 			.acceptIfNotNull(this.autoStartup, instance::setAutoStartup)
  45. 			.acceptIfNotNull(this.phase, instance::setPhase)
  46. 			.acceptIfNotNull(this.afterReceivePostProcessors, instance::setAfterReceivePostProcessors);
  47. 		if (endpoint != null) {
  48. 			if (endpoint.getAutoStartup() != null) {
  49. 				instance.setAutoStartup(endpoint.getAutoStartup());
  50. 			}
  51. 			instance.setListenerId(endpoint.getId());
  52.             
  53.             //setupListenerContainer方法参数为 MessageListenerContainer类型，
  54.             //SimpleMessageListenerContainer是MessageListenerContainer的子类。
  55. 			endpoint.setupListenerContainer(instance);
  56. 		}
  57. 		if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {
  58. 			AbstractAdaptableMessageListener messageListener = (AbstractAdaptableMessageListener) instance
  59. 					.getMessageListener();
  60. 			javaUtils
  61. 				.acceptIfNotNull(this.beforeSendReplyPostProcessors, messageListener::setBeforeSendReplyPostProcessors)
  62. 				.acceptIfNotNull(this.retryTemplate, messageListener::setRetryTemplate)
  63. 				.acceptIfCondition(this.retryTemplate != null && this.recoveryCallback != null, this.recoveryCallback,
  64. 					messageListener::setRecoveryCallback)
  65. 				.acceptIfNotNull(this.defaultRequeueRejected, messageListener::setDefaultRequeueRejected);
  66. 		}
  67. 		initializeContainer(instance, endpoint);
  68.  
  69. 		if (this.containerConfigurer != null) {
  70. 			this.containerConfigurer.accept(instance);
  71. 		}
  72.  
  73. 		return instance;
  74. 	}
  75. }


 

二、开始
 

1、HandlerMethod
  在被@RabbitListener标注的方法。该方法会被封装成HanderMethod。

  1. package org.springframework.messaging.handler;
  2.  
  3. import java.lang.annotation.Annotation;
  4. import java.lang.reflect.Method;
  5. import org.apache.commons.logging.Log;
  6. import org.apache.commons.logging.LogFactory;
  7. import org.springframework.beans.factory.BeanFactory;
  8. import org.springframework.core.BridgeMethodResolver;
  9. import org.springframework.core.GenericTypeResolver;
  10. import org.springframework.core.MethodParameter;
  11. import org.springframework.core.annotation.AnnotatedElementUtils;
  12. import org.springframework.core.annotation.SynthesizingMethodParameter;
  13. import org.springframework.lang.Nullable;
  14. import org.springframework.util.Assert;
  15. import org.springframework.util.ClassUtils;
  16.  
  17. public class HandlerMethod {
  18.     public static final Log defaultLogger = LogFactory.getLog(HandlerMethod.class);
  19.     protected Log logger;
  20.     private final Object bean;
  21.     @Nullable
  22.     private final BeanFactory beanFactory;
  23.     private final Class<?> beanType;
  24.     private final Method method;
  25.     private final Method bridgedMethod;
  26.     private final MethodParameter[] parameters;
  27.     @Nullable
  28.     private HandlerMethod resolvedFromHandlerMethod;
  29.  
  30.     public HandlerMethod(Object bean, Method method) {
  31.         this.logger = defaultLogger;
  32.         Assert.notNull(bean, "Bean is required");
  33.         Assert.notNull(method, "Method is required");
  34.         this.bean = bean;
  35.         this.beanFactory = null;
  36.         this.beanType = ClassUtils.getUserClass(bean);
  37.         this.method = method;
  38.         this.bridgedMethod = BridgeMethodResolver.findBridgedMethod(method);
  39.         this.parameters = this.initMethodParameters();
  40.     }
  41. }


我们常用其子类InvocableHandlerMethod。该类提供一个Object invoke(Message<?> message, Object... providedArgs) 方法，
 可以通过该invoke方法调用handlerMethod 中的method：method.invoke(bean,args)

HandlerAdapter称为hander适配器，该适配器中有InvocableHandlerMethod、DelegatingInvocableHandler这两个hander方法。注：InvocableHandlerMethod上文刚说过。

  1.  
  2. public class HandlerAdapter {
  3.  
  4. 	private final InvocableHandlerMethod invokerHandlerMethod;
  5.  
  6. 	private final DelegatingInvocableHandler delegatingHandler;
  7.  
  8. 	public HandlerAdapter(InvocableHandlerMethod invokerHandlerMethod) {
  9. 		this.invokerHandlerMethod = invokerHandlerMethod;
  10. 		this.delegatingHandler = null;
  11. 	}
  12.  
  13. 	public HandlerAdapter(DelegatingInvocableHandler delegatingHandler) {
  14. 		this.invokerHandlerMethod = null;
  15. 		this.delegatingHandler = delegatingHandler;
  16. 	}
  17.  
  18. 	public InvocationResult invoke(Message<?> message, Object... providedArgs) throws Exception { // NOSONAR
  19. 		if (this.invokerHandlerMethod != null) {
  20. 			// //InvocableHandlerMethod不为null,就调用invokerHandlerMethod.invoke方法。
  21. 			return new InvocationResult(this.invokerHandlerMethod.invoke(message, providedArgs),
  22. 					null, this.invokerHandlerMethod.getMethod().getGenericReturnType());
  23. 		}
  24. 		else if (this.delegatingHandler.hasDefaultHandler()) {
  25. 			// Needed to avoid returning raw Message which matches Object
  26. 			Object[] args = new Object[providedArgs.length + 1];
  27. 			args[0] = message.getPayload();
  28. 			System.arraycopy(providedArgs, 0, args, 1, providedArgs.length);
  29. 			return this.delegatingHandler.invoke(message, args);
  30. 		}
  31. 		else {
  32. 			return this.delegatingHandler.invoke(message, providedArgs);
  33. 		}
  34. 	}
  35.     //...
  36. }


这里可以简单的理解调用HandlerAdapter.invoke方法可以间接调用@RabbitListener修饰的方法。

2、ChannelAwareMessageListener 
消费端接收到mq的消息会调用ChannelAwareMessageListener 接口中的onMessage方法。

  1. /*
  2.  * Copyright 2002-2018 the original author or authors.
  3.  *
  4.  * Licensed under the Apache License, Version 2.0 (the "License");
  5.  * you may not use this file except in compliance with the License.
  6.  * You may obtain a copy of the License at
  7.  *
  8.  *      https://www.apache.org/licenses/LICENSE-2.0
  9.  *
  10.  * Unless required by applicable law or agreed to in writing, software
  11.  * distributed under the License is distributed on an "AS IS" BASIS,
  12.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13.  * See the License for the specific language governing permissions and
  14.  * limitations under the License.
  15.  */
  17. package org.springframework.amqp.rabbit.listener.api;
  18.  
  19. import org.springframework.amqp.core.Message;
  20. import org.springframework.amqp.core.MessageListener;
  21.  
  22. import com.rabbitmq.client.Channel;
  23.  
  24. /**
  25.  * A message listener that is aware of the Channel on which the message was received.
  26.  *
  27.  * @author Mark Pollack
  28.  * @author Gary Russell
  29.  */
  30. @FunctionalInterface
  31. public interface ChannelAwareMessageListener extends MessageListener {
  32.  
  33. 	/**
  34. 	 * 回调函数处理接受的消息
  35. 	 * 
  36. 	 * Callback for processing a received Rabbit message.
  37. 	 * <p>Implementors are supposed to process the given Message,
  38. 	 * typically sending reply messages through the given Session.
  39. 	 * @param message the received AMQP message (never <code>null</code>)
  40. 	 * @param channel the underlying Rabbit Channel (never <code>null</code>)
  41. 	 * @throws Exception Any.
  42. 	 */
  43. 	void onMessage(Message message, Channel channel) throws Exception; // NOSONAR
  44.  
  45. 	@Override
  46. 	default void onMessage(Message message) {
  47. 		throw new IllegalStateException("Should never be called for a ChannelAwareMessageListener");
  48. 	}
  49.  
  50. }


下面要说的是其子类：MessagingMessageListenerAdapter

  1. public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageListener {
  2.  
  3. 	private HandlerAdapter handlerAdapter;
  4.  
  5. 	private final MessagingMessageConverterAdapter messagingMessageConverter;
  6.  
  7. 	private final boolean returnExceptions;
  8.  
  9. 	private final RabbitListenerErrorHandler errorHandler;
  10.  
  11. 	public MessagingMessageListenerAdapter() {
  12. 		this(null, null);
  13. 	}
  14.  
  15. 	public MessagingMessageListenerAdapter(Object bean, Method method) {
  16. 		this(bean, method, false, null);
  17. 	}
  18.  
  19. 	public MessagingMessageListenerAdapter(Object bean, Method method, boolean returnExceptions,
  20. 			RabbitListenerErrorHandler errorHandler) {
  21. 		this.messagingMessageConverter = new MessagingMessageConverterAdapter(bean, method);
  22. 		this.returnExceptions = returnExceptions;
  23. 		this.errorHandler = errorHandler;
  24. 	}
  25.  
  26. @Override
  27. 	public void onMessage(org.springframework.amqp.core.Message amqpMessage, Channel channel) throws Exception { // NOSONAR
  28.  
  29. 		//ChannelAwareMessageListener.onMessage()方法中的message是org.springframework.amqp.core.Message类型。
  30. 		//这里转换成org.springframework.messaging.Message类型。
  31. 		Message<?> message = toMessagingMessage(amqpMessage);
  32. 		if (logger.isDebugEnabled()) {
  33. 			logger.debug("Processing [" + message + "]");
  34. 		}
  35. 		InvocationResult result = null;
  36. 		try {
  37. 			//入口
  38. 			result = invokeHandler(amqpMessage, channel, message);
  39. 			if (result.getReturnValue() != null) {
  40. 				handleResult(result, amqpMessage, channel, message);
  41. 			}
  42. 			else {
  43. 				logger.trace("No result object given - no result to handle");
  44. 			}
  45. 		}
  46. 		catch (ListenerExecutionFailedException e) {
  47. 			if (this.errorHandler != null) {
  48. 				try {
  49. 					message = MessageBuilder.fromMessage(message)
  50. 							.setHeader(AmqpHeaders.CHANNEL, channel)
  51. 							.build();
  52. 					Object errorResult = this.errorHandler.handleError(amqpMessage, message, e);
  53. 					if (errorResult != null) {
  54. 						handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()),
  55. 								amqpMessage, channel, message);
  56. 					}
  57. 					else {
  58. 						logger.trace("Error handler returned no result");
  59. 					}
  60. 				}
  61. 				catch (Exception ex) {
  62. 					returnOrThrow(amqpMessage, channel, message, ex, ex);
  63. 				}
  64. 			}
  65. 			else {
  66. 				returnOrThrow(amqpMessage, channel, message, e.getCause(), e);
  67. 			}
  68. 		}
  69. 	}
  70.  
  71. /**
  72. 	 * Invoke the handler, wrapping any exception to a {@link ListenerExecutionFailedException}
  73. 	 * with a dedicated error message.
  74. 	 * @param amqpMessage the raw message.
  75. 	 * @param channel the channel.
  76. 	 * @param message the messaging message.
  77. 	 * @return the result of invoking the handler.
  78. 	 */
  79. 	private InvocationResult invokeHandler(org.springframework.amqp.core.Message amqpMessage, Channel channel,
  80. 			Message<?> message) {
  81. 		try {
  82. 			return this.handlerAdapter.invoke(message, amqpMessage, channel);
  83. 		}
  84. 		catch (MessagingException ex) {
  85. 			throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
  86. 					"be invoked with the incoming message", message.getPayload()), ex, amqpMessage);
  87. 		}
  88. 		catch (Exception ex) {
  89. 			throw new ListenerExecutionFailedException("Listener method '" +
  90. 					this.handlerAdapter.getMethodAsString(message.getPayload()) + "' threw exception", ex, amqpMessage);
  91. 		}
  92. 	}
  93. //...
  94.  
  95. }


onMessage->invokehandler->this.handlerMethod.invoke 。this.handlerMethod为 HandlerAdapter类型。上面刚说完：调用HandlerAdapter.invoke方法可以间接调用@RabbitListener修饰的方法。

现在我们就已经梳洗清楚从onMessage到调用被@RabbitListener修饰的方法的整个流程

3、SimpleMessageListenerContainer
 

SimpleMessageListenerContainer 中有一个内部类AsyncMessageProcessingConsumer ：

  1. //当终端接受到信息时，一个线程会执行该run方法。
  2.     private final class AsyncMessageProcessingConsumer implements Runnable {
  3.  
  4.  
  5. 		private static final int ABORT_EVENT_WAIT_SECONDS = 5;
  6.  
  7. 		private final BlockingQueueConsumer consumer;
  8.  
  9. 		private final CountDownLatch start;
  10.  
  11. 		private volatile FatalListenerStartupException startupException;
  12.  
  13. 		private int consecutiveIdles;
  14.  
  15. 		private int consecutiveMessages;
  16.  
  17.  
  18. 		AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
  19. 			this.consumer = consumer;
  20. 			this.start = new CountDownLatch(1);
  21. 		}
  22.  
  23.  
  24.         @Override
  25.         public void run() { // NOSONAR - line count
  26. 			if (!isActive()) {
  27. 				return;
  28. 			}
  29.  
  30. 			boolean aborted = false;
  31.  
  32. 			this.consumer.setLocallyTransacted(isChannelLocallyTransacted());
  33.  
  34. 			String routingLookupKey = getRoutingLookupKey();
  35. 			if (routingLookupKey != null) {
  36. 				SimpleResourceHolder.bind(getRoutingConnectionFactory(), routingLookupKey); // NOSONAR both never null
  37. 			}
  38.  
  39. 			if (this.consumer.getQueueCount() < 1) {
  40. 				if (logger.isDebugEnabled()) {
  41. 					logger.debug("Consumer stopping; no queues for " + this.consumer);
  42. 				}
  43. 				SimpleMessageListenerContainer.this.cancellationLock.release(this.consumer);
  44. 				if (getApplicationEventPublisher() != null) {
  45. 					getApplicationEventPublisher().publishEvent(
  46. 							new AsyncConsumerStoppedEvent(SimpleMessageListenerContainer.this, this.consumer));
  47. 				}
  48. 				this.start.countDown();
  49. 				return;
  50. 			}
  51.  
  52. 			try {
  53. 				initialize();
  54. 				while (isActive(this.consumer) || this.consumer.hasDelivery() || !this.consumer.cancelled()) {
  55.                     //这个 while 可以无限的循环，
  56. 					mainLoop();
  57. 				}
  58. 			}catch (InterruptedException e) {
  59.             }
  60.         //...
  61.         }
  62.  
  63.         private void mainLoop() throws Exception { // NOSONAR Exception
  64. 			try {
  65. 				boolean receivedOk = receiveAndExecute(this.consumer); // At least one message received
  66. 				if (SimpleMessageListenerContainer.this.maxConcurrentConsumers != null) {
  67. 					checkAdjust(receivedOk);
  68. 				}
  69. 				long idleEventInterval = getIdleEventInterval();
  70. 				if (idleEventInterval > 0) {
  71. 					if (receivedOk) {
  72. 						updateLastReceive();
  73. 					}
  74. 					else {
  75. 						long now = System.currentTimeMillis();
  76. 						long lastAlertAt = SimpleMessageListenerContainer.this.lastNoMessageAlert.get();
  77. 						long lastReceive = getLastReceive();
  78. 						if (now > lastReceive + idleEventInterval
  79. 								&& now > lastAlertAt + idleEventInterval
  80. 								&& SimpleMessageListenerContainer.this.lastNoMessageAlert
  81. 								.compareAndSet(lastAlertAt, now)) {
  82. 							publishIdleContainerEvent(now - lastReceive);
  83. 						}
  84. 					}
  85. 				}
  86. 			}
  87. 			catch (ListenerExecutionFailedException ex) {
  88. 				// Continue to process, otherwise re-throw
  89. 				if (ex.getCause() instanceof NoSuchMethodException) {
  90. 					throw new FatalListenerExecutionException("Invalid listener", ex);
  91. 				}
  92. 			}
  93. 			catch (AmqpRejectAndDontRequeueException rejectEx) {
  94. 				/*
  95. 				 *  These will normally be wrapped by an LEFE if thrown by the
  96. 				 *  listener, but we will also honor it if thrown by an
  97. 				 *  error handler.
  98. 				 */
  99. 			}
  100. 		}
  101.     }
  102. }


run方法中调用了receiveAndExecute方法，方法名直译：接受并执行。

  1.  
  2. package org.springframework.amqp.rabbit.listener;
  3.  
  4. import java.io.IOException;
  5. import java.util.ArrayList;
  6. import java.util.Arrays;
  7. import java.util.Collection;
  8. import java.util.HashSet;
  9. import java.util.Iterator;
  10. import java.util.List;
  11. import java.util.Set;
  12. import java.util.concurrent.BlockingQueue;
  13. import java.util.concurrent.CountDownLatch;
  14. import java.util.concurrent.LinkedBlockingQueue;
  15. import java.util.concurrent.TimeUnit;
  16. import java.util.concurrent.TimeoutException;
  17. import java.util.concurrent.atomic.AtomicLong;
  18. import java.util.concurrent.atomic.AtomicReference;
  19.  
  20. import org.springframework.amqp.AmqpAuthenticationException;
  21. import org.springframework.amqp.AmqpConnectException;
  22. import org.springframework.amqp.AmqpException;
  23. import org.springframework.amqp.AmqpIOException;
  24. import org.springframework.amqp.AmqpIllegalStateException;
  25. import org.springframework.amqp.AmqpRejectAndDontRequeueException;
  26. import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
  27. import org.springframework.amqp.core.Message;
  28. import org.springframework.amqp.core.Queue;
  29. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  30. import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
  31. import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry;
  32. import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
  33. import org.springframework.amqp.rabbit.connection.RabbitUtils;
  34. import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
  35. import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
  36. import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
  37. import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
  38. import org.springframework.amqp.rabbit.support.ConsumerCancelledException;
  39. import org.springframework.amqp.rabbit.support.ListenerContainerAware;
  40. import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
  41. import org.springframework.jmx.export.annotation.ManagedMetric;
  42. import org.springframework.jmx.support.MetricType;
  43. import org.springframework.transaction.PlatformTransactionManager;
  44. import org.springframework.transaction.support.TransactionSynchronizationManager;
  45. import org.springframework.transaction.support.TransactionTemplate;
  46. import org.springframework.util.Assert;
  47. import org.springframework.util.backoff.BackOffExecution;
  48.  
  49. import com.rabbitmq.client.Channel;
  50. import com.rabbitmq.client.PossibleAuthenticationFailureException;
  51. import com.rabbitmq.client.ShutdownSignalException;
  52.  
  53. /**
  54.  * @author Mark Pollack
  55.  * @author Mark Fisher
  56.  * @author Dave Syer
  57.  * @author Gary Russell
  58.  * @author Artem Bilan
  59.  * @author Alex Panchenko
  60.  *
  61.  * @since 1.0
  62.  */
  63. public class SimpleMessageListenerContainer extends AbstractMessageListenerContainer {
  64.  
  65.  
  66.      //接受并执行
  67.     private boolean receiveAndExecute(final BlockingQueueConsumer consumer) throws Throwable {
  68.  
  69.        //do接受并执行
  70.         return doReceiveAndExecute(consumer);
  71.     }
  72.      //do接受并执行
  73.     private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Throwable {
  74.  
  75.         Channel channel = consumer.getChannel();
  76.         for (int i = 0; i < this.txSize; i++) {//txSize为一次事务接受的消息个数
  77.             //读取消息，这里阻塞的，但是有一个超时时间。
  78.             Message message = consumer.nextMessage(this.receiveTimeout);
  79.             if (message == null) {//阻塞超时
  80.                 break;
  81.             }
  82.             try {
  83.                 executeListener(channel, message);//消息接收已完成，现在开始处理消息。
  84.             }
  85.             catch (Exception e) {}
  86.         }
  87.         return consumer.commitIfNecessary(isChannelLocallyTransacted());
  88.     }
  89.         
  90.     
  91.     //处理消息开始。该方法在其父类中
  92.     protected void executeListener(Channel channel, Message messageIn) throws Exception {
  93.         try {
  94.             Message message = messageIn;
  95.             if (……) {
  96.                 //批处理信息，这个不研究
  97.             }else {
  98.                 invokeListener(channel, message);
  99.             }
  100.         }catch (Exception ex) {}
  101.     }
  102.     //在其父类中
  103.     protected void invokeListener(Channel channel, Message message) throws Exception {
  104.         //这里this.proxy.invokeListener最终会调用actualInvokeListener方法。
  105.         this.proxy.invokeListener(channel, message);
  106.     }
  107.     //在其父类中
  108.     protected void actualInvokeListener(Channel channel, Message message) throws Exception {
  109.         Object listener = getMessageListener();
  110.         if (listener instanceof ChannelAwareMessageListener) {
  111.             doInvokeListener((ChannelAwareMessageListener) listener, channel, message);
  112.         }
  113.         else if (listener instanceof MessageListener) {
  114.             //……
  115.             doInvokeListener((MessageListener) listener, message)
  116.         }else{
  117.             //……
  118.         }
  119.         
  120.     }    
  121.  
  122.     protected void doInvokeListener(ChannelAwareMessageListener listener, Channel channel, Message message)
  123.             throws Exception {
  124.             Channel channelToUse = channel;
  125.             try {
  126.                 listener.onMessage(message, channelToUse);
  127.             }
  128.             catch (Exception e) {
  129.                 throw wrapToListenerExecutionFailedExceptionIfNeeded(e, message);
  130.             }
  131.  
  132.     }
  133. }


在doInvokeListener方中最终调用了onMessage方法。

到这我知道了SimpleMessageListenerContainer内部类AsyncMessageProcessingConsumer被放入线程池中运行，是谁把AsyncMessageProcessingConsumer放入线程池中的呢？


4、RabbitListenerAnnotationBeanPostProcessor
RabbitListenerAnnotationBeanPostProcessor 听名得意：@RabbitListener注解的后置处理器。

  1. public class RabbitListenerAnnotationBeanPostProcessor implements BeanPostProcessor,SmartInitializingSingleton {
  2.  
  3.     //对接口BeanPostProcessor的实现。
  4.     @Override
  5.     public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
  6.         Class<?> targetClass = AopUtils.getTargetClass(bean);
  7.         final TypeMetadata metadata = this.typeCache.computeIfAbsent(targetClass, this::buildMetadata);
  8.         //获取被@RabbitListener修饰的方法。
  9.         for (ListenerMethod lm : metadata.listenerMethods) {
  10.             //获取方法上的@RabbitListener注解。
  11.             for (RabbitListener rabbitListener : lm.annotations) {
  12.                 //处理Amqp监听：
  13.                 //有这三个参数就以实现监听队列并调用方法。参数：rabbitListener为我们的方法上@RabbitListener注解。参数method和bean可以能过反射的方式调用我们的方法。
  14.                 processAmqpListener(rabbitListener, lm.method, bean, beanName);
  15.             }
  16.         }
  17.         return bean;
  18.     }
  19.     
  20.     //处理Amqp监听
  21.     protected void processAmqpListener(RabbitListener rabbitListener, Method method, Object bean, String beanName) {
  22.         Method methodToUse = checkProxy(method, bean);
  23.         
  24.         //Endpoint为终端，像电脑、手机都是终端，他们都可以接受外部信息并响应，如手机来短信了就有提示。
  25.         //这里也用了终端的概念，被@RabbitListener注解修饰方法也有终端的特点可以接受外部信息并响应。
  26.         //MethodRabbitListenerEndpoint名子也很形象，叫方法监听终端。
  27.         MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
  28.         endpoint.setMethod(methodToUse);//终端接收到信息时，会调用 methodToUse
  29.         processListener(endpoint, rabbitListener, bean, methodToUse, beanName);
  30.     }
  31.     //注册员
  32.     private final RabbitListenerEndpointRegistrar registrar = new RabbitListenerEndpointRegistrar();
  33.     
  34.     protected void processListener(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object bean,
  35.             Object adminTarget, String beanName) {
  36.         //下面是为终端注入一些属性
  37.         endpoint.setBean(bean);
  38.         endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
  39.         endpoint.setId(getEndpointId(rabbitListener));
  40.         endpoint.setQueueNames(resolveQueues(rabbitListener));
  41.         endpoint.setConcurrency(resolveExpressionAsStringOrInteger(rabbitListener.concurrency(), "concurrency"));
  42.         endpoint.setBeanFactory(this.beanFactory);
  43.         endpoint.setReturnExceptions(resolveExpressionAsBoolean(rabbitListener.returnExceptions()));
  44.  
  45.         ……
  46.  
  47.         RabbitListenerContainerFactory<?> factory = null;
  48.         String containerFactoryBeanName = resolve(rabbitListener.containerFactory());
  49.         if (StringUtils.hasText(containerFactoryBeanName)) {
  50.             Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");
  51.             try {
  52.                 factory = this.beanFactory.getBean(containerFactoryBeanName, RabbitListenerContainerFactory.class);
  53.             }
  54.             catch (NoSuchBeanDefinitionException ex) {
  55.                 throw new BeanInitializationException("Could not register rabbit listener endpoint on [" +
  56.                         adminTarget + "] for bean " + beanName + ", no " + RabbitListenerContainerFactory.class.getSimpleName() + " with id '" +
  57.                         containerFactoryBeanName + "' was found in the application context", ex);
  58.             }
  59.         }
  60.         //终端准备完成后，还要将终端进行注册。    
  61.         //注册员将终端注册到注册处
  62.         this.registrar.registerEndpoint(endpoint, factory);
  63.     }
  64.     
  65.     
  66.     
  67.  
  68.     
  69.     //对SmartInitializingSingleton的实现，该方法在bean的后置处理BeanPostProcessor之后调用。
  70.     @Override
  71.     public void afterSingletonsInstantiated() {
  72.         this.registrar.setBeanFactory(this.beanFactory);
  73.  
  74.         if (this.containerFactoryBeanName != null) {
  75.             //this.containerFactoryBeanName为“rabbitListenerContainerFactory”
  76.             this.registrar.setContainerFactoryBeanName(this.containerFactoryBeanName);
  77.         }
  78.  
  79.         // Actually register all listeners
  80.         this.registrar.afterPropertiesSet();
  81.  
  82.     }
  83.  
  84. }


 
    

RabbitListenerEndpointRegistrar ：rabbit 监听器注册员

  1. //终端注册员
  2. public class RabbitListenerEndpointRegistrar   implements  InitializingBean{
  3.  
  4.     private RabbitListenerEndpointRegistry endpointRegistry;//终端注册处
  5.     //终端描述器集合
  6.     private final List<AmqpListenerEndpointDescriptor> endpointDescriptors =
  7.             new ArrayList<AmqpListenerEndpointDescriptor>();
  8.  
  9.     //注册员将终端注册到注册处
  10.     public void registerEndpoint(RabbitListenerEndpoint endpoint, RabbitListenerContainerFactory<?> factory) {
  11.         
  12.         //将 endpoint和 factory 封装成 终端描述器
  13.         AmqpListenerEndpointDescriptor descriptor = new AmqpListenerEndpointDescriptor(endpoint, factory);
  14.         synchronized (this.endpointDescriptors) {
  15.             if (this.startImmediately) { //马下注册
  16.                  //将 终端注册到 终端注册处
  17.                 this.endpointRegistry.registerListenerContainer(descriptor.endpoint,
  18.                         resolveContainerFactory(descriptor), true);
  19.             }
  20.             else {//afterPropertiesSet()方法中统一注册
  21.                                    //放入终端描述器集合
  22.                 this.endpointDescriptors.add(descriptor);
  23.             }
  24.         }
  25.     }
  26.  
  27.      //InitializingBean接口的实现
  28.      @Override
  29.     public void afterPropertiesSet() {
  30.         //统一注册
  31.         registerAllEndpoints();
  32.     }
  33.     //统一注册
  34.     protected void registerAllEndpoints() {
  35.         synchronized (this.endpointDescriptors) {
  36.             for (AmqpListenerEndpointDescriptor descriptor : this.endpointDescriptors) {
  37.                 this.endpointRegistry.registerListenerContainer(
  38.                         descriptor.endpoint, resolveContainerFactory(descriptor));
  39.             }
  40.         }
  41.     }
  42.     
  43.  
  44.  
  45.  
  46.  
  47. }

 RabbitListenerEndpointRegistry：注册处

  1.  
  2.  
  3.  
  4.  
  5. public class RabbitListenerEndpointRegistry  implements SmartLifecycle{
  6.  
  7.     private final Map<String, MessageListenerContainer> listenerContainers =
  8.             new ConcurrentHashMap<String, MessageListenerContainer>();
  9.     //注册终端
  10.     public void registerListenerContainer(RabbitListenerEndpoint endpoint, RabbitListenerContainerFactory<?> factory,
  11.                 boolean startImmediately) {
  12.  
  13.         String id = endpoint.getId();
  14.         synchronized (this.listenerContainers) {
  15.             //创建 listenerContainer
  16.             MessageListenerContainer container = createListenerContainer(endpoint, factory);
  17.             this.listenerContainers.put(id, container);
  18.             ……
  19.  
  20.             if (startImmediately) {
  21.                 startIfNecessary(container);
  22.             }
  23.         }
  24.     }
  25.  
  26.         protected MessageListenerContainer createListenerContainer(RabbitListenerEndpoint endpoint,
  27.             RabbitListenerContainerFactory<?> factory) {
  28.         //调用RabbitListener容器工厂的createListenerContainer方法获取RabbitListener容器
  29.         MessageListenerContainer listenerContainer = factory.createListenerContainer(endpoint);
  30.  
  31.         return listenerContainer;
  32.     }
  33.  
  34.     //start 是对 smartLifecycle 接口的实现，由 spring 容器调用
  35.        //上面的的代码中已经创建了listenerContainer 并放入了listenerContainers集合中，现在要将集合中的 listenerContainer放入线程池中。
  36.         @Override
  37.     public void start() {
  38.         //遍历listenerContainers集合
  39.         for (MessageListenerContainer listenerContainer : getListenerContainers()) {
  40.             startIfNecessary(listenerContainer);
  41.         }
  42.     }
  43.     //调用集合中的listenerContainer的 start 方法。
  44.       private void startIfNecessary(MessageListenerContainer listenerContainer) {
  45.         if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
  46.             listenerContainer.start();
  47.         }
  48.     }
  49.  
  50. }


分析了RabbitListenerAnnotationBeanPostProcessor：@RabbitListener注解的后置处理器 。从读取被@RabbitListener修饰的方法 => 创建 endpoint =>创建MessageListenerContainer并放入集合中=>遍历集合中的MessageListenerContainer并调用 start方法。

MessageListenerContainer中的 start 方法做了什么？

其子类 SimpleMessageListenerContainer是我们学习的重点

  1. public interface MessageListenerContainer extends SmartLifecycle {
  2.  
  3.     /**
  4.      * Setup the message listener to use. Throws an {@link IllegalArgumentException}
  5.      * if that message listener type is not supported.
  6.      */
  7.     void setupMessageListener(Object messageListener);
  8.  
  9.     /**
  10.      * @return the {@link MessageConverter} that can be used to
  11.      * convert {@link org.springframework.amqp.core.Message}, if any.
  12.      */
  13.     MessageConverter getMessageConverter();
  14.  
  15. }


start 方法在其抽象子类 AbstractMessageListenerContainer 中实现了：

    //MessageListenerContainer 同样也继承了 SmartLifecycle接口,SmartLifecycle中的方法是由 Spring 容器调用的，这里我们手动调用了 start 方法，这意味这 start 会被调用两次。
   

  1. @Override
  2.     public void start() {
  3.         if (isRunning()) {//第二调用时会直接 return。
  4.             return;
  5.         }
  6.         
  7.         try {
  8.             doStart();
  9.         }
  10.         catch (Exception ex) {
  11.         }
  12.     }


doStart 方法子类会覆盖：

在子类：SimpleMessageListenerContainer中：

  1. @Override
  2.     protected void doStart() throws Exception {
  3.         ……
  4.         super.doStart();
  5.         synchronized (this.consumersMonitor) {
  6.             
  7.             int newConsumers = initializeConsumers();
  8.             
  9.             Set<AsyncMessageProcessingConsumer> processors = new HashSet<AsyncMessageProcessingConsumer>();
  10.             for (BlockingQueueConsumer consumer : this.consumers) {
  11.                         //AsyncMessageProcessingConsumer是SimpleMessageListenerContainer的内部类
  12.                 AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer);
  13.                 processors.add(processor);
  14.                         //放入线程池
  15.                 getTaskExecutor().execute(processor);
  16.                 
  17.             }
  18.         
  19.         }
  20.     }


RabbitListenerAnnotationBeanPostProcessor调用SimpleMessageListenerContainer是AsyncMessageProcessingConsumer放入线程池中的。


参考链接：https://blog.csdn.net/hong10086/article/details/93672166

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/381570
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号