springboot @EnableScheduling 啟用多線程

現象

使用@EnableScheduling注解后，可以發現所有任務都排隊執行，並且調度器線程名稱都是“taskScheduler-1”

原因

默認配置使用單線程調度器

解決方案

配置線程池任務調度器

@Bean
public TaskScheduler taskScheduler() {
    ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
    taskScheduler.setPoolSize(10);
    return taskScheduler;
}

源碼分析

• org.springframework.scheduling.annotation.EnableScheduling

// 導入配置
@Import(SchedulingConfiguration.class)

• org.springframework.scheduling.annotation.SchedulingConfiguration

// 注冊調度注解處理器
@Bean(name = TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public ScheduledAnnotationBeanPostProcessor scheduledAnnotationProcessor() {
	return new ScheduledAnnotationBeanPostProcessor();
}

• org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor

// 為registrar注入調度器實例
private void finishRegistration() {
	if (this.scheduler != null) {
		this.registrar.setScheduler(this.scheduler);
	}

	if (this.beanFactory instanceof ListableBeanFactory) {
		Map<String, SchedulingConfigurer> beans =
				((ListableBeanFactory) this.beanFactory).getBeansOfType(SchedulingConfigurer.class);
		List<SchedulingConfigurer> configurers = new ArrayList<>(beans.values());
		AnnotationAwareOrderComparator.sort(configurers);
		for (SchedulingConfigurer configurer : configurers) {
			configurer.configureTasks(this.registrar);
		}
	}

	if (this.registrar.hasTasks() && this.registrar.getScheduler() == null) {
		// 從beanFactory查找調度器實例注入
		Assert.state(this.beanFactory != null, "BeanFactory must be set to find scheduler by type");
		try {
			// Search for TaskScheduler bean...
			this.registrar.setTaskScheduler(resolveSchedulerBean(this.beanFactory, TaskScheduler.class, false));
		}
		catch (NoUniqueBeanDefinitionException ex) {
			logger.trace("Could not find unique TaskScheduler bean", ex);
			try {
				this.registrar.setTaskScheduler(resolveSchedulerBean(this.beanFactory, TaskScheduler.class, true));
			}
			catch (NoSuchBeanDefinitionException ex2) {
				if (logger.isInfoEnabled()) {
					logger.info("More than one TaskScheduler bean exists within the context, and " +
							"none is named 'taskScheduler'. Mark one of them as primary or name it 'taskScheduler' " +
							"(possibly as an alias); or implement the SchedulingConfigurer interface and call " +
							"ScheduledTaskRegistrar#setScheduler explicitly within the configureTasks() callback: " +
							ex.getBeanNamesFound());
				}
			}
		}
		catch (NoSuchBeanDefinitionException ex) {
			logger.trace("Could not find default TaskScheduler bean", ex);
			// Search for ScheduledExecutorService bean next...
			try {
				this.registrar.setScheduler(resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, false));
			}
			catch (NoUniqueBeanDefinitionException ex2) {
				logger.trace("Could not find unique ScheduledExecutorService bean", ex2);
				try {
					this.registrar.setScheduler(resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, true));
				}
				catch (NoSuchBeanDefinitionException ex3) {
					if (logger.isInfoEnabled()) {
						logger.info("More than one ScheduledExecutorService bean exists within the context, and " +
								"none is named 'taskScheduler'. Mark one of them as primary or name it 'taskScheduler' " +
								"(possibly as an alias); or implement the SchedulingConfigurer interface and call " +
								"ScheduledTaskRegistrar#setScheduler explicitly within the configureTasks() callback: " +
								ex2.getBeanNamesFound());
					}
				}
			}
			catch (NoSuchBeanDefinitionException ex2) {
				logger.trace("Could not find default ScheduledExecutorService bean", ex2);
				// Giving up -> falling back to default scheduler within the registrar...
				logger.info("No TaskScheduler/ScheduledExecutorService bean found for scheduled processing");
			}
		}
	}

	this.registrar.afterPropertiesSet();
}

• org.springframework.scheduling.config.ScheduledTaskRegistrar

// 配置調度任務
protected void scheduleTasks() {
	// 如果沒有任務調度器，則創建單線程調度器
	if (this.taskScheduler == null) {
		this.localExecutor = Executors.newSingleThreadScheduledExecutor();
		this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
	}
	if (this.triggerTasks != null) {
		for (TriggerTask task : this.triggerTasks) {
			addScheduledTask(scheduleTriggerTask(task));
		}
	}
	if (this.cronTasks != null) {
		for (CronTask task : this.cronTasks) {
			addScheduledTask(scheduleCronTask(task));
		}
	}
	if (this.fixedRateTasks != null) {
		for (IntervalTask task : this.fixedRateTasks) {
			addScheduledTask(scheduleFixedRateTask(task));
		}
	}
	if (this.fixedDelayTasks != null) {
		for (IntervalTask task : this.fixedDelayTasks) {
			addScheduledTask(scheduleFixedDelayTask(task));
		}
	}
}

• java.util.concurrent.Executors

// 創建線程池容量為1的調度器
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
 return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}