org.springframework.beans.factory.parsing.BeanComponentDefinition#org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler源码实例Demo

@Before
public void setUp() throws Exception {

	logger.debug("Setting up before '" + this.testName.getMethodName() + "'");

	int port = SocketUtils.findAvailableTcpPort(61613);

	this.activeMQBroker = new BrokerService();
	this.activeMQBroker.addConnector("stomp://127.0.0.1:" + port);
	this.activeMQBroker.setStartAsync(false);
	this.activeMQBroker.setPersistent(false);
	this.activeMQBroker.setUseJmx(false);
	this.activeMQBroker.getSystemUsage().getMemoryUsage().setLimit(1024 * 1024 * 5);
	this.activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5);
	this.activeMQBroker.start();

	ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
	taskScheduler.afterPropertiesSet();

	this.client = new ReactorNettyTcpStompClient("127.0.0.1", port);
	this.client.setMessageConverter(new StringMessageConverter());
	this.client.setTaskScheduler(taskScheduler);
}
 

@Test
public void standaloneEventRegistryWithCustomTaskScheduler() {
    contextRunner
        .withBean(ThreadPoolTaskScheduler.class)
        .withPropertyValues("flowable.eventregistry.enable-change-detection:true")
        .run(context -> {

        EventRegistryEngine eventRegistryEngine = context.getBean(EventRegistryEngine.class);
        assertThat(eventRegistryEngine).as("Event registry engine").isNotNull();
        assertAllServicesPresent(context, eventRegistryEngine);

        assertThat(eventRegistryEngine.getEventRegistryEngineConfiguration().getEventRegistryChangeDetectionManager())
            .isInstanceOf(DefaultEventRegistryChangeDetectionManager.class);
            assertThat(eventRegistryEngine.getEventRegistryEngineConfiguration().getEventRegistryChangeDetectionExecutor())
                .isInstanceOf(DefaultSpringEventRegistryChangeDetectionExecutor.class);

        DefaultSpringEventRegistryChangeDetectionExecutor executor = (DefaultSpringEventRegistryChangeDetectionExecutor)
            eventRegistryEngine.getEventRegistryEngineConfiguration().getEventRegistryChangeDetectionExecutor();
        assertThat(executor.getTaskScheduler()).isEqualTo(context.getBean(TaskScheduler.class));
    });
}
 

@Test
public void postProcessAfterInitialization() throws Exception {
    assertThat(processor.postProcessAfterInitialization(mock(Executor.class), toBeExcluded).getClass(),
            not(equalTo(ContextAwareExecutor.class)));
    //concurrent
    assertThat(processor.postProcessAfterInitialization(mock(Executor.class), beanName).getClass(),
            equalTo(ContextAwareExecutor.class));
    assertThat(processor.postProcessAfterInitialization(mock(ExecutorService.class), beanName).getClass(),
            equalTo(ContextAwareExecutorService.class));
    assertThat(processor.postProcessAfterInitialization(mock(ScheduledExecutorService.class), beanName).getClass(),
            equalTo(ContextAwareScheduledExecutorService.class));

    //spring
    assertThat(processor.postProcessAfterInitialization(mock(TaskScheduler.class), beanName).getClass(),
            equalTo(ContextAwareTaskScheduler.class));
    assertThat(processor.postProcessAfterInitialization(new ThreadPoolTaskExecutor(), beanName).getClass(),
            equalTo(ContextAwareThreadPoolTaskExecutor.class));
    assertThat(processor.postProcessAfterInitialization(new ThreadPoolTaskScheduler(), beanName).getClass(),
            equalTo(ContextAwareThreadPoolTaskScheduler.class));
    assertThat(processor.postProcessAfterInitialization(mock(AsyncListenableTaskExecutor.class), beanName).getClass(),
            equalTo(ContextAwareAsyncListenableTaskExecutor.class));
    assertThat(processor.postProcessAfterInitialization(mock(AsyncTaskExecutor.class), beanName).getClass(),
            equalTo(ContextAwareAsyncTaskExecutor.class));
    assertThat(processor.postProcessAfterInitialization(mock(SchedulingTaskExecutor.class), beanName).getClass(),
            equalTo(ContextAwareSchedulingTaskExecutor.class));
}
 

/**
 * Runs deadlock guard task
 *
 * @param deadlockGuardTask
 *            deadlock guard task
 */
@SuppressWarnings("unchecked")
public void runDeadlockFuture(Runnable deadlockGuardTask) {
    if (deadlockFuture == null) {
        ThreadPoolTaskScheduler deadlockGuard = conn.getDeadlockGuardScheduler();
        if (deadlockGuard != null) {
            try {
                deadlockFuture = (ScheduledFuture<Runnable>) deadlockGuard.schedule(deadlockGuardTask, new Date(packet.getExpirationTime()));
            } catch (TaskRejectedException e) {
                log.warn("DeadlockGuard task is rejected for {}", sessionId, e);
            }
        } else {
            log.debug("Deadlock guard is null for {}", sessionId);
        }
    } else {
        log.warn("Deadlock future is already create for {}", sessionId);
    }
}
 

private List<ThreadPoolTaskScheduler> getTaskSchedulers() {
    List<ThreadPoolTaskScheduler> taskSchedulers = new ArrayList<>();
    Object[] keys = consulIndexes.keySet().toArray();
    for (int i = 0; i < consulIndexes.size(); i++) {
        ThreadPoolTaskScheduler threadPoolTaskScheduler = getTaskScheduler();
        if (!StringUtils.isEmpty(keys[i])) {
            String[] keyArray = keys[i].toString().split("/");
            if (keyArray.length >= 1) {
                keys[i] = keyArray[keyArray.length - 1];
                threadPoolTaskScheduler.setThreadNamePrefix(keys[i] + "-");
            }
        }
        taskSchedulers.add(threadPoolTaskScheduler);
    }
    return taskSchedulers;
}
 

/**
 * thread pool for scheduler parallel task (not async):
 * pull block, trans monitor, statistic trans, delete info, reset groupList
 * @return ThreadPoolTaskScheduler
 */
@Bean(destroyMethod = "shutdown")
public ThreadPoolTaskScheduler mgrTaskScheduler() {
    log.info("start mgrTaskScheduler init..");
    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setPoolSize(schedulerProperties.getPoolSize());
    scheduler.setThreadNamePrefix(schedulerProperties.getThreadNamePrefix());
    scheduler.setAwaitTerminationSeconds(schedulerProperties.getAwaitTerminationSeconds());
    scheduler.setWaitForTasksToCompleteOnShutdown(
            schedulerProperties.getWaitForTasksToCompleteOnShutdown());
    return scheduler;
}
 

@Bean
public ThreadPoolTaskScheduler proxyTaskScheduler(){
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.setPoolSize(10);
    threadPoolTaskScheduler.setThreadNamePrefix("ops-task-");
    return threadPoolTaskScheduler;
}
 

@Bean(name = {"messageBrokerTaskScheduler", "messageBrokerSockJsTaskScheduler"})
public ThreadPoolTaskScheduler messageBrokerTaskScheduler() {
	ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
	scheduler.setThreadNamePrefix("MessageBroker-");
	scheduler.setPoolSize(Runtime.getRuntime().availableProcessors());
	scheduler.setRemoveOnCancelPolicy(true);
	return scheduler;
}
 

@Test
public void downloadAndMergeRemoteIndexInEmptyIndex()
    throws Exception
{
    Path repoDirectory = Paths.get( FileUtils.getBasedir( ), "target/repo-" + Long.toString( System.currentTimeMillis( ) ) );

    RemoteRepository remoteRepository = getRemoteRepository(repoDirectory);

    repositoryRegistry.putRepository( remoteRepository);
    repositoryRegistry.reload();

    downloadRemoteIndexScheduler.startup();

    downloadRemoteIndexScheduler.scheduleDownloadRemote( "test-repo-re", true, true );

    ( (ThreadPoolTaskScheduler) downloadRemoteIndexScheduler.getTaskScheduler() ).getScheduledExecutor().awaitTermination(
        10, TimeUnit.SECONDS );

    repositoryRegistry.removeRepository( "test-repo-re" );

    // search
    BooleanQuery.Builder iQuery = new BooleanQuery.Builder();
    iQuery.add( indexer.constructQuery( MAVEN.GROUP_ID, new StringSearchExpression( "commons-logging" ) ),
                BooleanClause.Occur.SHOULD );

    remoteRepository = getRemoteRepository( repoDirectory );
    FlatSearchRequest rq = new FlatSearchRequest( iQuery.build() );
    rq.setContexts(
        Arrays.asList( remoteRepository.getIndexingContext().getBaseContext(IndexingContext.class) ) );

    FlatSearchResponse response = indexer.searchFlat(rq);

    log.info( "returned hit count:{}", response.getReturnedHitsCount() );
    Assertions.assertThat( response.getReturnedHitsCount() ).isEqualTo( 8 );
}
 

@Test
public void should_schedule_register_task() {
	ApplicationRegistrator registrator = mock(ApplicationRegistrator.class);
	ThreadPoolTaskScheduler scheduler = mock(ThreadPoolTaskScheduler.class);
	RegistrationApplicationListener listener = new RegistrationApplicationListener(registrator, scheduler);

	listener.onApplicationReady(new ApplicationReadyEvent(mock(SpringApplication.class), null,
			mock(ConfigurableWebApplicationContext.class)));

	verify(scheduler).scheduleAtFixedRate(isA(Runnable.class), eq(Duration.ofSeconds(10)));
}
 

Object createAsyncTaskExecutorProxy(Object bean, boolean cglibProxy,
		AsyncTaskExecutor executor) {
	return getProxiedObject(bean, cglibProxy, executor, () -> {
		if (bean instanceof ThreadPoolTaskScheduler) {
			return new LazyTraceThreadPoolTaskScheduler(this.beanFactory,
					(ThreadPoolTaskScheduler) executor);
		}
		return new LazyTraceAsyncTaskExecutor(this.beanFactory, executor);
	});
}
 

private ThreadPoolTaskScheduler registerThreadPoolTaskScheduler(final String jobConfigurationBeanName) {
    final String beanName = jobConfigurationBeanName + EXECUTOR_SUFFIX;
    this.beanRegistrar.registerBean(
            ThreadPoolTaskScheduler.class,
            beanName,
            null,
            null,
            null,
            null,
            null
    );
    return this.applicationContext.getBean(beanName, ThreadPoolTaskScheduler.class);
}
 

/**
 * Configures the scheduler to allow multiple pools.
 *
 * @param taskRegistrar The task registrar.
 */
@Override
public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();

    threadPoolTaskScheduler.setPoolSize(POOL_SIZE);
    threadPoolTaskScheduler.setThreadNamePrefix("scheduled-task-pool-");
    threadPoolTaskScheduler.initialize();

    taskRegistrar.setTaskScheduler(threadPoolTaskScheduler);
}
 

public JenkinsStatusUpdateTrigger(Updater<JenkinsStatusUpdater> updater,
								  JenkinsProperties jenkinsProperties,
								  ThreadPoolTaskScheduler taskScheduler) {
	super(taskScheduler);
	this.jenkinsProperties = jenkinsProperties;
	this.updater = updater;
}
 

@Bean
@ConditionalOnMissingBean
public JenkinsStatusUpdateTrigger statusUpdateTrigger(JenkinsStatusUpdater jenkinsStatusUpdater,
                                                      JenkinsProperties jenkinsProperties,
                                                      ThreadPoolTaskScheduler taskScheduler) {
    return new JenkinsStatusUpdateTrigger(jenkinsStatusUpdater, jenkinsProperties, taskScheduler);
}
 

@Bean
public static ThreadPoolTaskScheduler createThreadPoolTaskScheduler() {
    ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
    taskScheduler.setPoolSize(4);
    taskScheduler.setRemoveOnCancelPolicy(true);
    taskScheduler.setBeanName("status-update-pool");
    taskScheduler.setErrorHandler(e -> log.error("error:", e));
    return taskScheduler;
}
 

@ConditionalOnProperty(name = "second_task_scheduler_bean", havingValue = "true")
@Bean(name = "taskScheduler")
public TaskScheduler secondTaskScheduler() {
    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setPoolSize(2);
    return scheduler;
}
 

@Bean
public TaskScheduler taskScheduler() {
    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setThreadNamePrefix("poolScheduler");
    scheduler.setPoolSize(httpClientProperties.getPoolSizeOfScheduler());
    return scheduler;
}
 

public ConfigWatch(ConsulConfigProperties properties, ConsulClient consul, BmsAuthClient bmsAuthClient,
                   LinkedHashMap<String, Long> initialIndexes, List<ThreadPoolTaskScheduler> taskSchedulers) {
    this.properties = properties;
    this.consul = consul;
    this.consulIndexes = new LinkedHashMap<>(initialIndexes);
    this.watchFutures = new ScheduledFuture<?>[consulIndexes.size()];
    this.bmsAuthClient = bmsAuthClient;
    if (taskSchedulers == null) {
        this.taskSchedulers = getTaskSchedulers();
    } else {
        this.taskSchedulers = taskSchedulers;
    }
}
 

/**
 * Get the task scheduler used by the HeartBeat Service.
 *
 * @return The task scheduler
 */
@Bean
@ConditionalOnMissingBean(name = "heartBeatServiceTaskScheduler")
public TaskScheduler heartBeatServiceTaskScheduler() {
    final ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setPoolSize(SINGLE_THREAD);
    return scheduler;
}
 

@Bean
public TaskScheduler taskScheduler() {
    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setThreadNamePrefix("poolScheduler");
    scheduler.setPoolSize(50);
    return scheduler;
}
 

private TaskScheduler defaultTaskScheduler() {
    ThreadPoolTaskScheduler threadPoolScheduler = new ThreadPoolTaskScheduler();
    threadPoolScheduler.setThreadNamePrefix("GraphQLWSKeepAlive-");
    threadPoolScheduler.setPoolSize(Runtime.getRuntime().availableProcessors());
    threadPoolScheduler.setRemoveOnCancelPolicy(true);
    threadPoolScheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
    threadPoolScheduler.initialize();
    return threadPoolScheduler;
}
 

@Override
public void stop() {
	if (this.running.compareAndSet(true, false) && this.watchFuture != null) {
		// shutdown current user-thread,
		// then the other daemon-threads will terminate automatic.
		((ThreadPoolTaskScheduler) this.taskScheduler).shutdown();

		this.watchFuture.cancel(true);
	}
}
 

Object createAsyncTaskExecutorProxy(
        Object bean, boolean cglibProxy,
        AsyncTaskExecutor executor) {
    return getProxiedObject(bean, cglibProxy, executor, () -> {
        if (bean instanceof ThreadPoolTaskScheduler) {
            return new LazyTraceThreadPoolTaskScheduler((ThreadPoolTaskScheduler) executor);
        }
        return new LazyTraceAsyncTaskExecutor(executor);
    });
}
 

@Test
public void should_no_schedule_register_task_when_not_autoRegister() {
	ApplicationRegistrator registrator = mock(ApplicationRegistrator.class);
	ThreadPoolTaskScheduler scheduler = mock(ThreadPoolTaskScheduler.class);
	RegistrationApplicationListener listener = new RegistrationApplicationListener(registrator, scheduler);
	listener.setAutoRegister(false);

	listener.onApplicationReady(new ApplicationReadyEvent(mock(SpringApplication.class), null,
			mock(ConfigurableWebApplicationContext.class)));

	verify(scheduler, never()).scheduleAtFixedRate(isA(Runnable.class), eq(Duration.ofSeconds(10)));
}
 

@Bean
@ConditionalOnMissingBean(name = "pubsubSubscriberThreadPool")
public ThreadPoolTaskScheduler pubsubSubscriberThreadPool() {
	ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
	scheduler.setPoolSize(this.gcpPubSubProperties.getSubscriber().getExecutorThreads());
	scheduler.setThreadNamePrefix("gcp-pubsub-subscriber");
	scheduler.setDaemon(true);
	return scheduler;
}
 

public static RuntimeBeanReference registerScheduler(
		String schedulerName, ParserContext context, @Nullable Object source) {

	if (!context.getRegistry().containsBeanDefinition(schedulerName)) {
		RootBeanDefinition taskSchedulerDef = new RootBeanDefinition(ThreadPoolTaskScheduler.class);
		taskSchedulerDef.setSource(source);
		taskSchedulerDef.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
		taskSchedulerDef.getPropertyValues().add("poolSize", Runtime.getRuntime().availableProcessors());
		taskSchedulerDef.getPropertyValues().add("threadNamePrefix", schedulerName + "-");
		taskSchedulerDef.getPropertyValues().add("removeOnCancelPolicy", true);
		context.getRegistry().registerBeanDefinition(schedulerName, taskSchedulerDef);
		context.registerComponent(new BeanComponentDefinition(taskSchedulerDef, schedulerName));
	}
	return new RuntimeBeanReference(schedulerName);
}
 

@Test
public void findSchedulerByName() {
    ThreadPoolTaskScheduler expected = new ThreadPoolTaskScheduler();
    Throwable exc = new NoUniqueBeanDefinitionException(TaskScheduler.class);
    when(context.getBean(TaskScheduler.class)).thenThrow(exc);
    when(context.getBean("taskScheduler", TaskScheduler.class)).thenReturn(expected);
    assertEquals(expected, beanLocator.resolveTaskScheduler());
}
 

@Test(timeout = 5000)
public void fallbackAfterConnectTimeout() throws Exception {
	TestClientHandler clientHandler = new TestClientHandler();
	this.testFilter.sleepDelayMap.put("/xhr_streaming", 10000L);
	this.testFilter.sendErrorMap.put("/xhr_streaming", 503);
	initSockJsClient(createXhrTransport());
	this.sockJsClient.setConnectTimeoutScheduler(this.wac.getBean(ThreadPoolTaskScheduler.class));
	WebSocketSession clientSession = sockJsClient.doHandshake(clientHandler, this.baseUrl + "/echo").get();
	assertEquals("Fallback didn't occur", XhrClientSockJsSession.class, clientSession.getClass());
	TextMessage message = new TextMessage("message1");
	clientSession.sendMessage(message);
	clientHandler.awaitMessage(message, 5000);
	clientSession.close();
}
 

public DatabaseMySQLTraceLogFlushHandler(DataSource dataSource) {
    Assert.notNull(dataSource, "DatabaseTraceLogFlushHandler must have dataSource.");
    this.dataSource = dataSource;
    doWithConnection(createTableIfNotExist);
    tableReady = true;
    threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.setDaemon(true);
    threadPoolTaskScheduler.setThreadNamePrefix("TraceLogFlush");
    threadPoolTaskScheduler.initialize();
    threadPoolTaskScheduler.schedule(cleanerTask,
        new CronTrigger(System.getProperty("rabbit.trace.flush.trigger", "0 0 1 * * ?")));
}