引入 rabbitmq adapter 后，使用 @RabbitListener 等注解预先定义好监听后，在项目启动后，再使用 RabbitListenerEndpointRegistry 注册监听时，出现 [java.util.concurrent.RejectedExecutionException]

版本信息

Jdk版本：1.8.0_331
SpringBoot版本：2.1.9.RELEASE
DynamicTp版本：1.1.6.1
配置中心版本：Nacos 2.1.2.RELEASE

问题描述

引入 rabbitmq adapter 后，使用 @RabbitListener 等注解预先定义好监听后，在项目启动后，再使用 RabbitListenerEndpointRegistry 注册监听时，出现 [java.util.concurrent.RejectedExecutionException]。

配置文件：

spring:
cloud:
nacos:
  discovery:
    server-addr: ${NACOS_HOST:127.0.0.1}:${NACOS_PORT:8848}
    heart-beat-timeout: 60
    watch-delay: 30000
    namespace: ${NACOS_NAMESPACE:rs10}
    group: business
    metadata:
      name: ${spring.application.name}
      version: @project.version@
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://192.168.3.111:3306/db_1
username: root
password: 1
redis:
host: ${REDIS_HOST:192.168.3.111}
port: ${REDIS_PORT:6379}
password: ${REDIS_PASSWORD:123456}
rabbitmq:
host: ${RABBIT_MQ_HOST:192.168.3.111}
port: ${RABBIT_MQ_PORT:5672}
username: ${RABBIT_MQ_USERNAME:server_user}
password: ${RABBIT_MQ_PASSWORD:123456}
virtual-host: "/"
publisher-confirms: true
publisher-returns: true
cache:
  channel:
    size: 20
template:
  retry:
    enabled: true
    initial-interval: 2000
  mandatory: true
listener:
  type: direct
  direct:
    acknowledge-mode: manual
    retry:
      enabled: true
mvc:
throw-exception-if-no-handler-found: true
dynamic:
tp:
  enabled: true
  enabledBanner: false
  enabledCollect: true
  collectorTypes: micrometer
  monitorInterval: 5
  rabbitmqTp:
    - threadPoolName: 'rabbitMqTp#rabbitConnectionFactory'
      threadPoolAliasName: rabbitmq-sub-pool
      corePoolSize: 30
      maximumPoolSize: 60
      keepAliveTime: 60
      unit: seconds
      runTimeout: 10000
      queueTimeout: 10000
      notifyEnabled: false
    - threadPoolName: 'rabbitMqTp#rabbitConnectionFactory#publish'
      threadPoolAliasName: rabbitmq-pub-pool
      corePoolSize: 30
      maximumPoolSize: 60
      keepAliveTime: 60
      unit: seconds
      runTimeout: 10000
      queueTimeout: 10000
      notifyEnabled: false
  executors:
    - threadPoolName: 'dtpExecutor'
      threadPoolAliasName: dtp-executor
      executorType: common
      corePoolSize: 10
      maximumPoolSize: 20
      queueCapacity: 1024
      queueType: VariableLinkedBlockingQueue
      rejectedHandlerType: CallerRunsPolicy
      keepAliveTime: 60
      threadNamePrefix: dtp-exec
      allowCoreThreadTimeOut: false
      waitForTasksToCompleteOnShutdown: true
      awaitTerminationSeconds: 10
      preStartAllCoreThreads: false
      runTimeout: 10000
      queueTimeout: 10000
      taskWrapperNames:
        - "ttl"
        - "mdc"
      notifyEnabled: false
logging:
level:
com.rs.cloud: trace
com.alibaba.nacos.client.config.impl.ClientWorker: warn
org.springframework.amqp: trace
org.springframework.retry: trace

代码使用步骤：

设置 AbstractConnectionFactory 的 Executor


import org.dromara.dynamictp.adapter.rabbitmq.RabbitMqDtpAdapter;
import org.dromara.dynamictp.common.entity.TpExecutorProps;
import org.dromara.dynamictp.common.properties.DtpProperties;
import org.dromara.dynamictp.core.executor.DtpExecutor;
import org.dromara.dynamictp.core.support.DtpLifecycleSupport;
import org.dromara.dynamictp.core.support.ExecutorWrapper;
import org.dromara.dynamictp.core.support.ThreadPoolExecutorProxy;
import org.dromara.dynamictp.starter.adapter.rabbitmq.autoconfigure.RabbitMqTpAutoConfiguration;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SimplePropertyValueConnectionNameStrategy;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import javax.annotation.PostConstruct; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger;

@Configuration @AutoConfigureAfter({RabbitAutoConfiguration.class}) @AutoConfigureBefore({RabbitMqTpAutoConfiguration.class}) @EnableRabbit public class AmqpConfig {

private static final AtomicInteger NUMBER = new AtomicInteger(0);
private static final String DEFAULT_NAME = "rabbitmqPool";
private static final String DEFAULT_ALIAS_NAME_PREFIX = "rabbitmq-pool-";
private final Environment environment;
private final AbstractConnectionFactory connectionFactory;
private final RabbitTemplate rabbitTemplate;
private final RabbitMqDtpAdapter rabbitMqDtpAdapter;
private final DirectRabbitListenerContainerFactory listenerContainerFactory;

public AmqpConfig(Environment environment, AbstractConnectionFactory connectionFactory,
                  RabbitTemplate rabbitTemplate, RabbitMqDtpAdapter rabbitMqDtpAdapter,
                  DirectRabbitListenerContainerFactory listenerContainerFactory) {
    this.environment = environment;
    this.connectionFactory = connectionFactory;
    this.rabbitTemplate = rabbitTemplate;
    this.rabbitMqDtpAdapter = rabbitMqDtpAdapter;
    this.listenerContainerFactory = listenerContainerFactory;
}

@PostConstruct
public void afterPropertiesSet() {
    SimplePropertyValueConnectionNameStrategy cns = new SimplePropertyValueConnectionNameStrategy("spring.application.name");
    cns.setEnvironment(environment);
    connectionFactory.setConnectionNameStrategy(cns);

    ThreadPoolExecutor subscribePool = rabbitmqSubscribePool();
    System.out.println(subscribePool);

    connectionFactory.setExecutor(subscribePool);

    // set dynamic-tp proxy
    String publishTpName = generatePublishThreadPoolName();
    ThreadPoolExecutor rabbitmqPublishPool = rabbitmqPublishPool();
    ThreadPoolExecutorProxy proxy = new ThreadPoolExecutorProxy(rabbitmqPublishPool);
    ConnectionFactory publisherConnectionFactory = connectionFactory.getPublisherConnectionFactory();
    if (publisherConnectionFactory instanceof AbstractConnectionFactory) {
        ((AbstractConnectionFactory) publisherConnectionFactory).setExecutor(proxy);
    }
    Map<String, ExecutorWrapper> executorWrappers = rabbitMqDtpAdapter.getExecutorWrappers();
    executorWrappers.put(publishTpName, new ExecutorWrapper(publishTpName, proxy));
    DtpLifecycleSupport.shutdownGracefulAsync(rabbitmqPublishPool, publishTpName, 5);

    rabbitTemplate.setUsePublisherConnection(true);
    rabbitTemplate.setConfirmCallback(new AmqpConfirmCallback());
    rabbitTemplate.setReturnCallback(new AmqpReturnCallback());

    listenerContainerFactory.setConnectionFactory(connectionFactory);
}

private ThreadPoolExecutor rabbitmqSubscribePool() {
    return assembleThreadPoolExecutor("rabbitMqTp#rabbitConnectionFactory");
}

private ThreadPoolExecutor rabbitmqPublishPool() {
    return assembleThreadPoolExecutor(generatePublishThreadPoolName());
}

private ThreadPoolExecutor assembleThreadPoolExecutor(String name) {
    List<TpExecutorProps> propsList = DtpProperties.getInstance().getRabbitmqTp();

    if (Objects.isNull(propsList)) {
        return getDefaultThreadPoolExecutor();
    }

    TpExecutorProps props = propsList.stream()
            .filter(i -> name.equals(i.getThreadPoolName()))
            .findFirst()
            .orElse(null);

    if (Objects.isNull(props)) {
        return getDefaultThreadPoolExecutor();
    }

    DtpExecutor dtpExecutor = new DtpExecutor(props.getCorePoolSize(), props.getMaximumPoolSize(), props.getKeepAliveTime(), props.getUnit(),
            new LinkedBlockingQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
    dtpExecutor.setThreadPoolName(props.getThreadPoolName());
    dtpExecutor.setThreadPoolAliasName(props.getThreadPoolAliasName());
    dtpExecutor.setRunTimeout(props.getRunTimeout());
    dtpExecutor.setQueueTimeout(props.getQueueTimeout());
    dtpExecutor.setNotifyEnabled(props.isNotifyEnabled());
    return dtpExecutor;
}

private ThreadPoolExecutor getDefaultThreadPoolExecutor() {
    DtpExecutor dtpExecutor = new DtpExecutor(30, 60, 60, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
    int i = NUMBER.getAndIncrement();
    dtpExecutor.setThreadPoolName(DEFAULT_NAME + i);
    dtpExecutor.setThreadPoolAliasName(DEFAULT_ALIAS_NAME_PREFIX + i);
    dtpExecutor.setRunTimeout(2000);
    dtpExecutor.setQueueTimeout(2000);
    dtpExecutor.setNotifyEnabled(false);
    return dtpExecutor;
}

public static String generatePublishThreadPoolName() {
    return "rabbitMqTp#rabbitConnectionFactory#publish";
}

}

2. 定义监听
```java
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
@Order
@Slf4j
public class DefaultDeadLetterListener {

    @RabbitListener(queues = "#{@commonDeadLetterQueue.getName()}", containerFactory = "rabbitListenerContainerFactory")
    public void processDeadLetterPublicMessage(org.springframework.amqp.core.Message message,
                                               @Header(value = "#{T(com.rs.cloud.business.mns.constant.MessageHeaderKey).CONTENT_TYPE_ID}",
                                                       required = false) String contentClassName) {
        log.info("MESSAGE ###: {}", message);
        log.info("contentClassName: {}", contentClassName);
    }
}

使用 RabbitListenerEndpointRegistry 动态注册监听
```
// ...
RabbitListenerEndpointRegistry.registerListenerContainer(endpoint, listenerContainerFactory, true);
// ...
```
- 报错信息：如下图
- 猜测可能原因：启动时，使用 AmqpConfig 中 rabbitmqSubscribePool() 返回的线程池启动了监听容器，后续因为引入了 dynamic-tp-spring-boot-starter-adapter-rabbitmq ，rabbitmqSubscribePool() 返回的线程池被 shutdown，被 ThreadPoolExecutorProxy 替代，但是项目启动时生成的监听使用的 connection 是 rabbitmqSubscribePool() 返回的线程池，而不是 ThreadPoolExecutorProxy
  复现步骤
  
  按照上述编写完代码后，手动调用RabbitListenerEndpointRegistry.registerListenerContainer(endpoint, listenerContainerFactory, true)。

其他信息

无

dromara / dynamic-tp