[源码解析]之-spring cloud task启动执行流程详解

亦凉 2022-10-22 02:54 225阅读 0赞

*日子过得真快，又到了2021年的第一个假期*

*今天成都的天气真不错，很适合晒被子*

*趁着这个假期的第一天，感觉将上周每天下班回来查看的spring cloud task源码整理一下吧！*

# 前言 #

在采用spring cloud task实现定时任务默认是允许并发执行的，也就是同一时刻运行同时运行多个同一task实例

但是task的一般业务场景都是不允许并发执行的，为此spring cloud task官方的文档中也有对此的说明并提供了配置

详情点此链接进行查看  
[Restricting Spring Cloud Task Instances][]

其主要内容如下：

Spring Cloud Task lets you establish that only one task with a given task name can be run at a time. To do so, you need to establish the task name and set spring.cloud.task.single-instance-enabled=true for each task execution. While the first task execution is running, any other time you try to run a task with the same task name and`spring.cloud.task.single-instance-enabled=true`, the task fails with the following error message: Task with name “application” is already running. The default value for spring.cloud.task.single-instance-enabled is false. The following example shows how to set spring.cloud.task.single-instance-enabled to true:

> spring.cloud.task.single-instance-enabled=true or false

To use this feature, you must add the following Spring Integration dependencies to your application:

<dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-core</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-jdbc</artifactId>
    </dependency>

上面的主要内容是说可以通过配置**spring.cloud.task.single-instance-enabled**来确保同一名称的task name同一时间只运行一个，如果已经运行了与当前实例名称相同的task,第二个实例运行时会报错，错误内容为：*Task with name “application” is already running.*

但让上面的配置有效必须有一个前提条件，那就是得要引入**Spring Integration**的依赖

--------------------

结合公司业务需要，我们组开发的task同一时间也是不允许相同的task并发运行的，为此我也照着官方给出的说明进行了配置，通过本地的测试也确实实现了同一task的单一运行

![千军万马过独木桥图片一张][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center]

但是某天意外的情况就发生了，通过spring cloud data flow观察到某个task持续了好几个小时都一直是运行失败，错误代码全为1

![what the fuck图片一张][what the fuck]

根据官方的说明错误代码为1代表已经有实例在运行了

> The exit code for the application will be 1 if the task fails because this feature is enabled and another task is running with the same task name.

然后我观察了下当前所有的task在k8s中pod的状态，发现出错的task的状态全为error

为了先能让task运行起来，我便先将所有出错的task进行了删除，以确保当前时刻没有task是在运行的

等到下一轮job开始调度时发现task仍然报错了，显示出了同样的错误代码1

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 1]

当时我便猜想应该是由于锁没有释放掉造成的，通过spring-integration和spring-integration-jdbc的依赖应该会在数据库中创建一个**锁记录**，之后便让公司的运维大哥查询了下线上的数据

> select \* from TASK\_LOCK

查询一看确实发现了有一条数据存在

为了快速解决问题当时让运维大哥手动执行了删除操作，之后一切便终于平静了

![删库跑路图片][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 2]

也正因为task运行报错激发了我想要看一下spring cloud task实现分布式锁源码的冲动，趁着又是一个不加班的夜晚，一起来学习一下吧！

# spring cloud task源码解析 #

为了显得专业，先将spring cloud task源码拉下来,源码地址如下

https://github.com/spring-cloud/spring-cloud-task.git

之后再导入到idea中,这货的源码主要就长下面这样：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 3]

一个spring boot项目想变成spring cloud task项目只用引入**spring-cloud-task-core**依赖，再加上 **@EnableTask** 注解就可以了。所以就直接看下spring-cloud-task-core的源码

<dependency>
    			<groupId>org.springframework.cloud</groupId>
    			<artifactId>spring-cloud-task-core</artifactId>
    		</dependency>

## @EnableTask注解作用 ##

先来了解下 **@EnableTask**注解的作用

package org.springframework.cloud.task.configuration;
    
    import java.lang.annotation.Documented;
    import java.lang.annotation.ElementType;
    import java.lang.annotation.Inherited;
    import java.lang.annotation.Retention;
    import java.lang.annotation.RetentionPolicy;
    import java.lang.annotation.Target;
    
    import org.springframework.context.annotation.Import;
    
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @Import(TaskLifecycleConfiguration.class)
    public @interface EnableTask { 
    }

其中包含了 **@Import(TaskLifecycleConfiguration.class)**

在TaskLifecycleConfiguration中有如下代码：

@Autowired
    	public TaskLifecycleConfiguration(TaskProperties taskProperties,
    			ConfigurableApplicationContext context, TaskRepository taskRepository,
    			TaskExplorer taskExplorer, TaskNameResolver taskNameResolver,
    			ObjectProvider<ApplicationArguments> applicationArguments) {
    
    		this.taskProperties = taskProperties;
    		this.context = context;
    		this.taskRepository = taskRepository;
    		this.taskExplorer = taskExplorer;
    		this.taskNameResolver = taskNameResolver;
    		this.applicationArguments = applicationArguments.getIfAvailable();
    	}
    
    	@Bean
    	public TaskLifecycleListener taskLifecycleListener() {
    		return this.taskLifecycleListener;
    	}
    
    	/**
    	 * Initializes the {@link TaskLifecycleListener} for the task app.
    	 */
    	@PostConstruct
    	protected void initialize() {
    		if (!this.initialized) {
    			this.taskLifecycleListener = new TaskLifecycleListener(this.taskRepository,
    					this.taskNameResolver, this.applicationArguments, this.taskExplorer,
    					this.taskProperties,
    					new TaskListenerExecutorObjectFactory(this.context));
    
    			this.initialized = true;
    		}
    	}

通过PostContruct初始化方法创建一个**TaskLifecycleListener**的bean，bean中将task关联的配置文件也传了进去，初始化了创建了一个很重要的**TaskListenerExecutorObjectFactory**的类。这两个类很重要,下面我们一一品读一下

# TaskLifecycleListener #

看来一下TaskLifecycleListener的类描述

public class TaskLifecycleListener implements ApplicationListener<ApplicationEvent>,
    		SmartLifecycle, DisposableBean, Ordered

从TaskLifecycleListener的实现方法我们需要关注两个地方,一个是**ApplicationListener** 另一个是**SmartLifecycle**

由于在spring中只要实现了**ApplicationListener**，就可以监听到spring中的**ApplicationEvent**，通过**onApplicationEvent**进行处理

TaskLifecycleListener中处理ApplicationEvent事件的代码如下：

/**
    	 * Utilizes {@link ApplicationEvent}s to determine the end and failure of a task.
    	 * Specifically:
    	 * <ul>
    	 * <li>{@link ApplicationReadyEvent} - Successful end of a task</li>
    	 * <li>{@link ApplicationFailedEvent} - Failure of a task</li>
    	 * </ul>
    	 * @param applicationEvent The application being listened for.
    	 */
    	@Override
    	public void onApplicationEvent(ApplicationEvent applicationEvent) {
    		if (applicationEvent instanceof ApplicationFailedEvent) {
    			this.applicationFailedException = ((ApplicationFailedEvent) applicationEvent)
    					.getException();
    			doTaskEnd();
    		}
    		else if (applicationEvent instanceof ExitCodeEvent) {
    			this.exitCodeEvent = (ExitCodeEvent) applicationEvent;
    		}
    		else if (applicationEvent instanceof ApplicationReadyEvent) {
    			doTaskEnd();
    		}
    	}

从代码可以看出TaskLifecycleListener的onApplicationEvent方法主要做了善后的操作，执行了doTaskEnd方法

再看下TaskLifecycleListener实现了SmartLifecycle接口做的事情

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 4]

它重写了SmartLifecycle中的start方法和stop方法,分别调用了doTaskStart和doTaskEnd方法,这两个是比较重要的方法，下面分别重点读一下

### doTaskStart方法 ###

org.springframework.cloud.task.listener.TaskLifecycleListener\#doTaskStart

private void doTaskStart() { 
    		try { 
    			if (!this.started) { 
    			    //初始化taskExecutionListeners
    				this.taskExecutionListeners = new ArrayList<>();
                    //实不相瞒，这一行看不懂，为啥获取到了一个object但是又没有接收对象
    				this.taskListenerExecutorObjectFactory.getObject();
    				if (!CollectionUtils.isEmpty(this.taskExecutionListenersFromContext)) { 
    					this.taskExecutionListeners
    							.addAll(this.taskExecutionListenersFromContext);
    				}
    				//向taskExecutionListeners中添加taskListenerExecutorObjectFactory获取到了taskListener对象，这一行很重要，下面有单独的分析
    				this.taskExecutionListeners
    						.add(this.taskListenerExecutorObjectFactory.getObject());
    
    				List<String> args = new ArrayList<>(0);
    
    				if (this.applicationArguments != null) { 
    					args = Arrays.asList(this.applicationArguments.getSourceArgs());
    				}
    				//判断是否指定了executionId
    				if (this.taskProperties.getExecutionid() != null) { 
    				    //指定了executionId,则从taskExplorer中获取到此id的TaskExecution信息
    					TaskExecution taskExecution = this.taskExplorer
    							.getTaskExecution(this.taskProperties.getExecutionid());
    					//taskId获取不到信息就知道抛出异常
    					Assert.notNull(taskExecution,
    							String.format("Invalid TaskExecution, ID %s not found",
    									this.taskProperties.getExecutionid()));
    					//如果taskId对应的task早已执行完毕了,同样抛出异常
    					Assert.isNull(taskExecution.getEndTime(), String.format(
    							"Invalid TaskExecution, ID %s task is already complete",
    							this.taskProperties.getExecutionid()));
    					//创建一个taskExecution记录
    					this.taskExecution = this.taskRepository.startTaskExecution(
    							this.taskProperties.getExecutionid(),
    							this.taskNameResolver.getTaskName(), new Date(), args,
    							this.taskProperties.getExternalExecutionId(),
    							this.taskProperties.getParentExecutionId());
    				}
    				else { 
    				    //定义一个TaskExecution，并指定task的相关信息，包括task名称和task启动时间以及task的参数信息
    					TaskExecution taskExecution = new TaskExecution();
    					taskExecution.setTaskName(this.taskNameResolver.getTaskName());
    					taskExecution.setStartTime(new Date());
    					taskExecution.setArguments(args);
    					taskExecution.setExternalExecutionId(
    							this.taskProperties.getExternalExecutionId());
    					taskExecution.setParentExecutionId(
    							this.taskProperties.getParentExecutionId());
    					//在数据库中进行记录task的执行信息
    					this.taskExecution = this.taskRepository
    							.createTaskExecution(taskExecution);
    				}
    			}
    			else { 
    			    //早已初始化过了，打个错误日志出来
    				logger.error(
    						"Multiple start events have been received. The first one was "
    								+ "recorded.");
    			}
                //执行invokeOnTaskStartup方法,进行启动前的监听方法
    			setExitMessage(invokeOnTaskStartup(this.taskExecution));
    		}
    		catch (Throwable t) { 
    			// This scenario will result in a context that was not startup.
    			this.applicationFailedException = t;
    			this.doTaskEnd();
    			throw t;
    		}
    	}
    	
    	private TaskExecution invokeOnTaskStartup(TaskExecution taskExecution) { 
    		//执行task启动前的数据采集
    		this.taskMetrics.onTaskStartup(taskExecution);
    		//手动深拷贝出一个同样数据的TaskExecution出来
    		TaskExecution listenerTaskExecution = getTaskExecutionCopy(taskExecution);
    		List<TaskExecutionListener> startupListenerList = new ArrayList<>(
    				this.taskExecutionListeners);
    		if (!CollectionUtils.isEmpty(startupListenerList)) { 
    			try { 
    			    //进行集合反转倒置一下
    				Collections.reverse(startupListenerList);
    				for (TaskExecutionListener taskExecutionListener : startupListenerList) { 
    				    //遍历执行一下它的onTaskStartup方法详细看下面的TaskListenerExecutor方法,其分布式锁的获取部分就在其中
    					taskExecutionListener.onTaskStartup(listenerTaskExecution);
    				}
    			}
    			catch (Throwable currentListenerException) { 
    				logger.error(currentListenerException);
    				this.listenerFailed = true;
    				//执行过程中报错了，则设置错误信息
    				this.taskExecution.setErrorMessage(currentListenerException.getMessage());
    				this.listenerException = currentListenerException;
    				throw currentListenerException;
    			}
    		}
    		//返回listenerTaskExecution
    		return listenerTaskExecution;
    	}

通过上面的代码我们知道 **doTaskStart()** 方法通过 **taskListenerExecutorObjectFactory.getObject()** 方法获取到了注册在spring容器中关于task的一些独有的监听器（taskListenerExecutorObjectFactory是在TaskLifecycleConfiguration初始化时指定了spring context的）,获取到了监听器后又执行了invokeOnTaskStartup方法，并将监听器开始执行各自的onTaskStartup方法。

当然在执行上面的流程中还对执行信息生成了对象并保存到了数据库中

### TaskListenerExecutorObjectFactory ###

上面描述的TaskListenerExecutorObjectFactory这个工厂类的实现也很值得了解一下

public class TaskListenerExecutorObjectFactory
    		implements ObjectFactory<TaskExecutionListener>

从类的定义中可以看出**TaskListenerExecutorObjectFactory**它是一个用于创建TaskExecutionListener的工厂类,调用getObject()方法就会返回一个**TaskExecutionListener**的对象

@Override
    	public TaskListenerExecutor getObject() { 
    		this.beforeTaskInstances = new HashMap<>();
    		this.afterTaskInstances = new HashMap<>();
    		this.failedTaskInstances = new HashMap<>();
    		initializeExecutor();
    		return new TaskListenerExecutor(this.beforeTaskInstances, this.afterTaskInstances,
    				this.failedTaskInstances);
    	}
    	
    	private void initializeExecutor() { 
    		ConfigurableListableBeanFactory factory = this.context.getBeanFactory();
    		//获取出spring中的所有已经注册好的bean
    		for (String beanName : this.context.getBeanDefinitionNames()) { 
                //依次遍历beanName,判断是否是scoped的代码bean
    			if (!ScopedProxyUtils.isScopedTarget(beanName)) { 
    				Class<?> type = null;
    				try { 
    					type = AutoProxyUtils.determineTargetClass(factory, beanName);
    				}
    				catch (RuntimeException ex) { 
    					// An unresolvable bean type, probably from a lazy bean - let's ignore
    					// it.
    					if (logger.isDebugEnabled()) { 
    						logger.debug("Could not resolve target class for bean with name '"
    								+ beanName + "'", ex);
    					}
    				}
    				if (type != null) { 
    					if (ScopedObject.class.isAssignableFrom(type)) { 
    						try { 
    							type = AutoProxyUtils.determineTargetClass(factory,
    									ScopedProxyUtils.getTargetBeanName(beanName));
    						}
    						catch (RuntimeException ex) { 
    							// An invalid scoped proxy arrangement - let's ignore it.
    							if (logger.isDebugEnabled()) { 
    								logger.debug(
    										"Could not resolve target bean for scoped proxy '"
    												+ beanName + "'",
    										ex);
    							}
    						}
    					}
    					try { 
    					   //对bean进行处理
    						processBean(beanName, type);
    					}
    					catch (RuntimeException ex) { 
    						throw new BeanInitializationException(
    								"Failed to process @BeforeTask "
    										+ "annotation on bean with name '" + beanName
    										+ "'",
    								ex);
    					}
    				}
    			}
    		}
    
    	}
    	
    	private void processBean(String beanName, final Class<?> type) { 
    	    //nonAnnotatedClasses缓存,相当于黑名单,存放没有task相关注解的类
    		if (!this.nonAnnotatedClasses.contains(type)) { 
    		    //通过MethodGetter方法获取出有BeforeTask注解的方法
    			Map<Method, BeforeTask> beforeTaskMethods = (new MethodGetter<BeforeTask>())
    					.getMethods(type, BeforeTask.class);
    			//通过MethodGetter方法获取出有AfterTask注解的方法
    			Map<Method, AfterTask> afterTaskMethods = (new MethodGetter<AfterTask>())
    					.getMethods(type, AfterTask.class);
    			//通过MethodGetter方法获取出有FailedTask注解的方法
    			Map<Method, FailedTask> failedTaskMethods = (new MethodGetter<FailedTask>())
    					.getMethods(type, FailedTask.class);
                //没有BeforeTask和AfterTask注解的方法,则将type这个类加到黑明单中去
    			if (beforeTaskMethods.isEmpty() && afterTaskMethods.isEmpty()) { 
    				this.nonAnnotatedClasses.add(type);
    				return;
    			}
    			//type这个类中有BeforeTask注解
    			if (!beforeTaskMethods.isEmpty()) { 
    				for (Method beforeTaskMethod : beforeTaskMethods.keySet()) { 
    					//将这个BeforeTask注解修饰的方法和它的类放到beforeTaskInstances中
    					this.beforeTaskInstances.put(beforeTaskMethod,
    							this.context.getBean(beanName));
    				}
    			}
    			//type这个类中有AfterTask注解
    			if (!afterTaskMethods.isEmpty()) { 
    				for (Method afterTaskMethod : afterTaskMethods.keySet()) { 
    				    //将这个AfterTask注解修饰的方法和它的类放到afterTaskInstances中
    					this.afterTaskInstances.put(afterTaskMethod,
    							this.context.getBean(beanName));
    				}
    			}
    			//type这个类中有FailedTask注解
    			if (!failedTaskMethods.isEmpty()) { 
    				for (Method failedTaskMethod : failedTaskMethods.keySet()) { 
    					//将这个FailedTask注解修饰的方法和它的类放到failedTaskInstances中
    					this.failedTaskInstances.put(failedTaskMethod,
    							this.context.getBean(beanName));
    				}
    			}
    		}
    	}
    	
    	private static class MethodGetter<T extends Annotation> { 
    
    		public Map<Method, T> getMethods(final Class<?> type,
    				final Class<T> annotationClass) { 
    			return MethodIntrospector.selectMethods(type,
    					(MethodIntrospector.MetadataLookup<T>) method -> AnnotationUtils
    							.findAnnotation(method, annotationClass));
    		}
    
    	}

通过代码可知,TaskListenerExecutorObjectFactory只是从spring context中获取到task关联的Listener，最终并将其封装到一个TaskListenerExecutor中

### spring cloud task执行方法 ###

再看一下springcloud task执行的代码在哪

**org.springframework.boot.SpringApplication.run(String… args)**

public ConfigurableApplicationContext run(String... args) {
    		StopWatch stopWatch = new StopWatch();
    		stopWatch.start();
    		DefaultBootstrapContext bootstrapContext = createBootstrapContext();
    		ConfigurableApplicationContext context = null;
    		configureHeadlessProperty();
    		SpringApplicationRunListeners listeners = getRunListeners(args);
    		listeners.starting(bootstrapContext, this.mainApplicationClass);
    		try {
    			ApplicationArguments applicationArguments = new DefaultApplicationArguments(args);
    			ConfigurableEnvironment environment = prepareEnvironment(listeners, bootstrapContext, applicationArguments);
    			configureIgnoreBeanInfo(environment);
    			Banner printedBanner = printBanner(environment);
    			context = createApplicationContext();
    			context.setApplicationStartup(this.applicationStartup);
    			prepareContext(bootstrapContext, context, environment, listeners, applicationArguments, printedBanner);
    			refreshContext(context);
    			afterRefresh(context, applicationArguments);
    			stopWatch.stop();
    			if (this.logStartupInfo) {
    				new StartupInfoLogger(this.mainApplicationClass).logStarted(getApplicationLog(), stopWatch);
    			}
    			//发送task开始执行消息
    			listeners.started(context);
    			//执行runner
    			callRunners(context, applicationArguments);
    		}
    		catch (Throwable ex) {
    		    //发送task执行失败消息
    			handleRunFailure(context, ex, listeners);
    			throw new IllegalStateException(ex);
    		}
    
    		try {
    		    //发送task执行完毕消息
    			listeners.running(context);
    		}
    		catch (Throwable ex) {
    			handleRunFailure(context, ex, null);
    			throw new IllegalStateException(ex);
    		}
    		return context;
    	}
    
    	private void callRunners(ApplicationContext context, ApplicationArguments args) {
    		List<Object> runners = new ArrayList<>();
    		//从spring容器中找到所有实现了ApplicationRunner的bean
    		runners.addAll(context.getBeansOfType(ApplicationRunner.class).values());
    		//从spring容器中找到所有实现了CommandLineRunner的bean
    		runners.addAll(context.getBeansOfType(CommandLineRunner.class).values());
    		//对runners进行排序
    		AnnotationAwareOrderComparator.sort(runners);
    		//根据排序过程依次执行
    		for (Object runner : new LinkedHashSet<>(runners)) {
    			if (runner instanceof ApplicationRunner) {
    				callRunner((ApplicationRunner) runner, args);
    			}
    			if (runner instanceof CommandLineRunner) {
    				callRunner((CommandLineRunner) runner, args);
    			}
    		}
    	}
    	
    	//调用ApplicationRunner的run方法
    	private void callRunner(ApplicationRunner runner, ApplicationArguments args) {
    		try {
    			(runner).run(args);
    		}
    		catch (Exception ex) {
    			throw new IllegalStateException("Failed to execute ApplicationRunner", ex);
    		}
    	}
    	
    	//调用CommandLineRunner的run方法
    	private void callRunner(CommandLineRunner runner, ApplicationArguments args) {
    		try {
    			(runner).run(args.getSourceArgs());
    		}
    		catch (Exception ex) {
    			throw new IllegalStateException("Failed to execute CommandLineRunner", ex);
    		}
    	}

从上面的代码中可以看到关于runner启动前、运行中、运行异常、运行后的整个执行代码

### doTaskEnd方法 ###

看了task执行前的代码后再看一下task执行完成后的代码逻辑

它是触发方法是通过TaskLifecycleListener的onApplicationEvent监听方法触发的

再仔细看一下doTaskEnd方法

private void doTaskEnd() {
    	    //判断task是否已经执行过结束逻辑了
    		if ((this.listenerFailed || this.started) && !this.finished) {
    			//设置当前时间为结束时间
    			this.taskExecution.setEndTime(new Date());
    
    			if (this.applicationFailedException != null) {
    			    //有异常则记录异常信息
    				this.taskExecution.setErrorMessage(
    						stackTraceToString(this.applicationFailedException));
    			}
    
    			this.taskExecution.setExitCode(calcExitStatus());
    			if (this.applicationFailedException != null) {
    				setExitMessage(invokeOnTaskError(this.taskExecution,
    						this.applicationFailedException));
    			}
                //执行invokeOnTaskEnd方法，进行task监听器的逻辑处理，如锁资源释放等操作
    			setExitMessage(invokeOnTaskEnd(this.taskExecution));
    			//在数据库中记录task完成信息
    			this.taskRepository.completeTaskExecution(this.taskExecution.getExecutionId(),
    					this.taskExecution.getExitCode(), this.taskExecution.getEndTime(),
    					this.taskExecution.getExitMessage(),
    					this.taskExecution.getErrorMessage());
    
    			this.finished = true;
    
    			if (this.taskProperties.getClosecontextEnabled() && this.context.isActive()) {
    				this.context.close();
    			}
    
    		}
    		else if (!this.started) {
    			logger.error("An event to end a task has been received for a task that has "
    					+ "not yet started.");
    		}
    	}
    	
    
    	private TaskExecution invokeOnTaskEnd(TaskExecution taskExecution) {
    		this.taskMetrics.onTaskEnd(taskExecution);
    		TaskExecution listenerTaskExecution = getTaskExecutionCopy(taskExecution);
    		if (this.taskExecutionListeners != null) {
    			try {
    				for (TaskExecutionListener taskExecutionListener : this.taskExecutionListeners) {
    					taskExecutionListener.onTaskEnd(listenerTaskExecution);
    				}
    			}
    			catch (Throwable listenerException) {
    				String errorMessage = stackTraceToString(listenerException);
    				if (StringUtils.hasText(listenerTaskExecution.getErrorMessage())) {
    					errorMessage = String.format("%s :Task also threw this Exception: %s",
    							errorMessage, listenerTaskExecution.getErrorMessage());
    				}
    				logger.error(errorMessage);
    				listenerTaskExecution.setErrorMessage(errorMessage);
    				this.listenerFailed = true;
    			}
    		}
    		return listenerTaskExecution;
    	}
    	
    	//org.springframework.cloud.task.listener.annotation.TaskListenerExecutor.onTaskEnd方法
    	/**
    	 * Executes all the methods that have been annotated with @AfterTask.
    	 * @param taskExecution associated with the event.
    	 */
    	@Override
    	public void onTaskEnd(TaskExecution taskExecution) {
    		executeTaskListener(taskExecution, this.afterTaskInstances.keySet(),
    				this.afterTaskInstances);
    	}

从上面的doTaskEnd方法可以看出，它主要执行了在task运行结束时的一些善后操作；包括记录task的执行完毕信息到数据库中、task执行完成后对于分布式锁的释放

# 总结 #

最后还是简单梳理一下spring cloud task执行的整体流程图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 5]

涉及到的部分知识点

*  spring boot starter原理
 *  spring boot 中enable\*原理
 *  spring event listener事件监听机制
 *  spring中bean的自动装配
 *  spring aware接口

[Restricting Spring Cloud Task Instances]: https://docs.spring.io/spring-cloud-task/docs/current/reference/htmlsingle/#features-single-instance-enabled
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221022/7bbdbfeb954d4d66824c37d7a620f1dc.png
[what the fuck]: /images/20221022/2134457973c94d2e81e21e425619c1d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221022/0bc1d7ba3fa244ea9231b786423d1c24.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221022/6dec29c929344883bc694f63bd0134e2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221022/9fe99a352bf24778953d54652d665173.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221022/befc7e5169444c11aee256abf3034e6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3B1aGFpeWFuZw_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221022/5199b61895b6438087f1c06a4de9910c.png