SpringBoot CommandLineRunner的异步任务机制使用

2024-08-21 10:11 开发作者：架构随笔录

1.概要分析

1.1 手动创建线程

在 CommandLineRunner 的 run 方法中，你可以直接创建并启动一个新的线程来执行异步任务。

这种方法简单直接，但需要注意线程管理和异常处理。

@Component  
public class MyCommandLineRunner implements CommandLineRunner {  
  
    @Override  
    public void run(String... args) throws Exception {  
        new Thread(() -> {  
            // 异步执行的代码  
            System.out.println("异步任务执行中...");  
            // 模拟耗时操作  
            try {  
                Thread.sleep(1000);  
            } catch (InterruptedException e) {  
                Thread.currentThread().interrupt();  
            }  
            System.out.println("异步任务完成");  
        }).start();  
  
        // 主线程继续执行，不会等待异步任务完成  
        System.out.println("CommandLineRunner 执行完毕，主线程继续");  
    }  
}

1.2 使用 Spring 的异步执行功能

如果你希望利用 Spring 的异步支持来执行异步任务，你可以在 CommandLineRunner 中注入一个使用 @Async 注解的服务。但是，需要注意的是，由于 CommandLineRunner 的 run 方法本身是在 Spring 容器完全初始化之后同步执行的，因此即使你调用了一个异步服务方法，run 方法本身仍然会立即返回，不会等待异步任务完成。

首先，你需要在 Spring Boot 应用中启用异步支持，通过在启动类上添加 @EnableAsync 注解。

然后，你可以创建一个异步服务：

@Service  
public class AsyncService {  
  
    @Async  
    public void executeAsyncTask() {  
        // 异步执行的代码  
        System.out.println("异步任务执行中...");  
        // 模拟耗时操作  
        try {  
            Thread.sleep(1000);  
        } catch (InterruptedException e) {  
            Thread.currentThread().interrupt();  
        }  
        System.out.println("异步任务完成");  
    }  
}

在 CommandLineRunner 中注入并使用这个服务：

@Component  
public class MyCommandLineRunner implements CommandLineRunner {  
  
    @Autowired  
    private AsyncService asyncService;  
  
    @Override  
    public void run(String... args) throws Exception {  
        asyncService.executeAsyncTask(); // 调用异步方法，但 CommandLineRunner 的 run 方法会立即返回  
  
        System.out.println("CommandLineRunner 执行完毕，主线程继续，不会等待异步任务完成");  
    }  
}

虽然 CommandLineRunner 本身不支持异步执行，但你可以通过手动创建线程或使用 Spring 的异步支持来在 CommandLineRunner 中执行异步任务。然而，需要注意的是，CommandLineRunner 的 run 方法本身仍然是同步执行的，它不会等待任何异步任务完成。

如果你的应用程序依赖于异步任务的结果，你可能需要采用其他机制（如 Future、CompletableFuture 或消息队列）来管理异步任务的执行和结果。

2.核心原理分析

org.springframework.boot.CommandLineRunner 是 Spring Boot 框架中的一个核心接口，其原理分析可以从以下几个方面进行。

2.1 接口定义与功能

CommandLineRunner 是一个函数式接口（Functional Interface），它只定义了一个抽象方法 run(String... args)。

这个方法在 Spring Boot 应用程序启动完成后被调用，允许开发者执行一些初始化操作或启动后的任务。这些任务可能包括数据初始化、缓存预热、日志记录等。

2.2 执行时机

当 Spring Boot 应用程序启动时，Spring 容器会完成一系列的初始化操作，包括 Bean 的加载和依赖注入等。

在所有 Spring Bean 都初始化完成后，Spring Boot 会查找所有实现了 CommandLineRunner 接口的 Bean，并依次调用它们的 run 方法。

这意味着 CommandLineRunner 的执行时机是在 Spring 上下文准备好之后，但在应用程序对外提供服务之前。

2.3 命令行参数传递

CommandLineRunner 的 run 方法接收一个 String... args 参数，这个参数包含了启动应用程序时传递给它的命令行参数。

这使得开发者可以根据命令行参数的不同来执行不同的初始化逻辑。

2.4 实现与注册

要使用 CommandLineRunner，开发者需要创建一个类并实现这个接口，然后重写 run 方法以定义自己的初始化逻辑。

为了让 Spring 容器能够扫描到这个实现类并将其注册为一个 Bean，通常会在类上添加 @Component 或其他类似的注解（如 @Service、@Repository 等）。此外，也可以通过编程方式在配置编程客栈类中显式地注册这个 Bean。

2.5 执行顺序

如果应用程序中有多个实现了 CommandLineRunner 接口的类，那么它们的 run 方法将按照一定的顺序执行。

默认情况下，执行顺序取决于 Spring 容器注册这些 Bean 的顺序。但是，开发者可以通过 @Order 注解或实现 Ordered 接口来指定执行顺序。

@Order 注解的值越小，优先级越高，相应的 run 方法就会越早执行。

2.6 与 ApplicationRunner 的区别

值得注意的是，Spring Boot 还提供了另一个类似的接口 ApplicationRunner，它也用于在应用程序启动后执行初始化任务。与 CommandLineRunner 不同的是，ApplicationRunner 的 run 方法接收一个 ApplicationArguments 参数而不是 String... args。

ApplicationArguments 提供了对命令行参数的更高级别访问，包括选项和非选项参数等。此外，如果同时存在 CommandLineRunner 和 ApplicationRunner 的实现，那么 CommandLineRunner 的实现会先于 ApplicationRunner 的实现被调用。

2.7 应用场景

CommandLineRunner 适用于需要在应用程序启动后立即执行的任务场景，如数据初始化、配置加载、缓存预热等。通过使用 CommandLineRunner，开发者可以确保这些任务在应用程序对外提供服务之前完成，从而提高应用程序的性能和用户体验。

综上所述，org.sphttp://www.devze.comringframework.boot.CommandLineRunner 是 Spring Boot 框架中用于执行启动后任务的强大机制，它通过简单的接口定义和灵活的注册方式，为开发者提供了方便、高效的初始化操作手段。

3.部分源码分析

3.1 启动Spring Boot应用程序

/**
 * 启动应用程序。
 * 
 * @param args 命令行参数
 * @return ConfigurableApplicationContext 应用程序上下文
 */
public ConfigurableApplicationContext run(String... args) {
    // 记录应用程序启动时间
    long startTime = System.nanoTime();
    
    // 创建引导上下文
    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);
        
        // 配置忽略BeanInfo的设置
        configureIgnoreBeanInfo(environment);
        
        // 打印启动横幅
        Banner printedBanner = printBanner(environment);
        
        // 创建应用程序上下文
js        context = createApplicationContext();
        
        // 设置应用启动器
        context.setApplicationStartup(this.applicationStartup);
        
        // 准备应用程序上下文
        prepareContext(bootstrapContext, context, environment, listeners, applicationArguments, printedBanner);
        
        // 刷新上下文，使配置生效
        refreshContext(context);
        
        // 启动后配置
        afterRefresh(context, applicationArguments);
        
        // 计算应用程序启动时间
        Duration timeTakenToStartup = Duration.ofNanos(System.nanoTime() - startTime);
        
        // 如果启用了启动信息日志，则记录启动信息
        if (this.logStartupInfo) {
            new StartupInfoLogger(this.mainApplicationClass).logStarted(getApplicationLog(), timeTakenToStartup);
        }
        
        // 通知监听器应用程序已启动
        listeners.started(context, timeTakenToStartup);
        
        // 调用应用程序运行者
        callRunners(context, applicationArguments);
    } catch (Throwable ex) {
        // 处理启动失败
        handleRunFailure(context, ex, listeners);
        throw new IllegalStateException(ex);
    }
    
    try {
        // 计算应用程序就绪时间
        Duration timeTakenToReady = Duration.ofNanos(System.nanoTime() - startTime);
        
        // 通知监听器应用程序已就绪
        listeners.ready(context, timeTakenToReady);
    } catch (Throwable ex) {
        // 处理就绪失败
        handleRunFailure(context, ex, null);
        throw new IllegalStateException(ex);
    }
    
    // 返回应用程序上下文
    return context;
}

3.2 调用所有的Runner实现类

/**
 * 调用所有的Runner实现类。
 * 
 * 本方法的目的是遍历ApplicationContext中所有的Runner实例，并根据它们的类型分别调用相应的方法。
 * Runner和CommandLineRunner是Spring Boot提供的一组接口，用于在应用程序启动后执行一些自定义的初始化逻辑。
 * 这里通过判断Runner的类型，来决定是调用ApplicationRunner还是CommandLineRunner的方法，从而实现对不同类型Runner的兼容处理。
 * 
 * @param context Spring应用上下文，用于获取BeanProvider以获取Runner实例。
 * @param args 命令行参数，传递给每个Runner的调用方法。
 */
private void callRunners(ApplicationContext context, ApplicationArguments args) {
    // 通过BeanProvider获取所有Runner类型的bean，并以有序的方式遍历它们
    context.getBeanProvider(Runner.class).orderedStream().forEach((runner) -> {
        // 如果runner是ApplicationRunner类型，则调用callRunner方法，并传入ApplicationRunner和命令行参数
        if (runner instanceof ApplicationRunner) {
            callRunner((ApplicationRunner) runner, args);
        }
        // 如果runner是CommandLineRunner类型，则同样调用callRunner方法，并传入CommandLineRunner和命令行参数
        if (runner instanceof CommandLineRunner) {
            callRunner((CommandLineRunner) runner, args);
        }
    });
}

4.典型的应用场景分析

Seata中的TM（Transaction Manager，事务管理器）和RM（Resource Manager，资源管理器）是分布式事务框架中的关键角色，它们各自承担着特定的职责，以确保分布式事务的一致性和完整性。

4.1 TM（Transaction Manager，事务管理器）

4.1.1 定义与职责

（1）TM负责定义全局事务的范围，即开始全局事务、提交或回滚全局事务。它是分布式事务的发起者和终结者，类似于本地事务中的begin…commit…rollback操作，但针对的是全局的分布式事务。
（2）在Seata框架中，TM与业务系统集成在一起，作为客户端存在。当业务操作需要跨多个服务或数据库时，TM会启动一个全局事务，并管理这个事务的生命周期。

4.1.2 工作流程

（1）TM向TC（Transaction Coordinator，事务协调者）请求开启一个全局事务，TC生成一个全局唯一的事务ID（XID）并返回给TM。
（2）TM携带XID进行远程服务调用，XID在微服务调用链中传播，确保所有参与的分支事务都能被正确关联到全局事务中。
（3）当业务操作完成后，TM根据业务逻辑向TC发起全局事务的提交或回滚请求。
（4）TC根据各分支事务的执行结果，决定全局事务的提交或回滚，并通知所有RM进行相应的操作。

4.2 RM（Resource Manager，资源管理器）

4.2.1 定义与职责

（1）RM负责管理分支事务处理的资源，如数据库连接、消息队列等。它是分布式事务中具体执行操作的组件。
（2）RM与TC进行通信，注册分支事务、报告分支事务的状态，并根据TC的指令驱动分支事务的提交或回滚。
（3）在Seata框架中，RM同样与业务系统集成在一起，作为客户端存在。每个参与全局事务的服务或数据库操作，都会有一个对应的RM来管理。

4.2.2 工作流程

（1）在执行具体业务操作之前，RM会向TC注册分支事务，并将其纳入全局事务的管辖范围。
（2）RM执行本地事务操作，如数据库更新、消息发送等，并确保这些操作是可回滚和持久化的。
（3）RM将分支事务的执行结果（提交或回滚）上报给TC。
（4）当TC收到TM的全局事务提交或回滚请求时，它会根据各分支事务的状态决定全局事务的结果，并通知所有RM进行相应的提交或回滚操作。

4.3 Seata RM和TM与Seata Server之间的RPC通信

在Seata中，TM（Transaction Manager，事务管理器）与Seata Server（即TC，Transaction Coordinator，事务协调者）之间的通信是通过RPC（Remote Procedure Call，远程过程调用）实现的。

RPC是一种允许程序在网络上调用远程计算机上程序的技术，就像调用本地计算机上的程序一样。

4.3.1 TM与Seata Server之间的RPC通信

（1）建立连接

TM在启动时会尝试与Seata Server建立长连接。这个连接是通过Netty等网络通信框架实现的，Netty提供了高效、异步的网络通信能力。
在建立连接的过程中，TM会向Seata Server发送注册请求，包括应用ID、事务组名称等信息，以便Seata Server能够识别和管理该TM。

（2）事务管理

一旦连接建立，TM就可以通过RPC调用Seata Server提供的服务来管理全局事务。
例如，当业务操作需要跨多个服务或数据库时，TM会向Swww.devze.comeata Server请求开启一个全局事务，并获取一个全局唯一的事务ID（XID）。
在执行远程服务调用时，TM会将XID携带在调用中，以便参与的RM能够识别并将分支事务注册到全局事务中。
当业务操作完成后，TM会根据业务逻辑向Seata Server发起全局事务的提交或回滚请求。

（3）通信协议

Seata使用自定义的通信协议来进行RPC通信，该协议支持事务的创建、提交、回滚等操作。
通信过程中，Seata还实现了心跳检测、超时重试等机制来确保通信的可靠性和稳定性。

（4）性能优化

为了提高RPC通信的性能，Seata采用了多种优化策略，如使用Netty的主从Reactor多线程模型来处理并发请求、采用批量发送请求来减少网络开销等。

4.3.2 RM与Seata Server之间的RPC通信

在Seata中，RM负责管理分支事务处理的资源，如数据库连接等。

当RM执行分支事务时，它需要与Seata Server进行通信，以注册分支事务、报告分支事务的状态，并根据Seata Server的指令驱动分支事务的提交或回滚。

这种通信是通过RPC机制实现的，它允许RM远程调用Seata Server上的服务。

（1）建立连接

当RM启动时，它会根据配置尝试与Seata Server建立长连接。这个连接是通过Netty等网络通信框架实现的，Netty提供了高效、异步的网络通信能力。
在建立连接的过程中，RM会向Seata Server发送注册请求，包括应用ID、事务组名称等信息，以便Seata Server能够识别和管理该RM。

（2）分支事务注册

当RM执行一个分支事务时，它会向Seata Server注册该分支事务。注册过程中，RM会携带全局事务ID（XID）等信息，以便Seata Server能够将该分支事务关联到相应的全局事务中。
Seata Server在收到注册请求后，会为该分支事务分配一个唯一的分支事务ID，并将其注册到全局事务中。

（3）状态报告与指令执行

在分支事务执行过程中，RM会定期向Seata Server报告分支事务的状态，如正在执行、已提交、已回滚等。
当全局事务需要提交或回滚时，Seata Server会根据各分支事务的状态和结果，向RM发送相应的提交或回滚指令。
RM在收到指令后，会执行相应的操作，如提交本地事务、回滚本地事务等，并将执行结果报告给Seata Server。

（4）心跳检测与异常处理

为了保持连接的活跃状态，RM会定期向Seata Server发送心跳消息。
如果Seata Server在一段时间内没有收到RM的心跳消息，它可能会认为RM已经离线，并采取相应的异常处理措施，如重试连接、记录日志等。

Seata RM与Seata Server之间的RPC通信是Seata分布式事务框架中的重要组成部分。通过高效的RPC通信机制，RM能够远程调用Seata Server提供的服务来管理分支事务，确保分布式事务的一致性和完整性。同时，Seata还通过多种优化策略来提高RPC通信的性能和可靠性。

4.4 Seata Server利用SpringBoot CommandLineRunner启动服务端通信渠道

在Seata Server中，ServerRunner类是一个重要的组件，它继承自Spring Boot的CommandLineRunner接口。

这意味着在Spring Boot应用启动后，ServerRunner的run()方法会被自动执行。这种方法通常用于在应用启动后立即执行一段特定的代码，比如初始化资源、启动服务等。

ServerRunner类的主要职责是初始化Netty通信渠道，即NettyRemotingServer。

Netty是一个高性能的异步事件驱动的网络应用框架，用于快速开发可维护的高性能协议服务器和客户端。

在Seata中，Netty用于节点间的通信，包括事务协调器（TC）、事务管理器（TM）和资源管理器（RM）之间的通信。

以下是ServerRunner类中run()方法可能包含的逻辑的一个简单示例：

@Override  
public void run(String... args) throws Exception {  
    // 初始化Netty通信服务器  
    NettyRemotingServer nettyRemotingServer = new NettyRemotingServer();  
    nettyRemotingServer.setPort(serverPort); // 设置服务器端口  
    nettyRemotingServer.setHost(serverHost); // 设置服务器主机地址  
    nettyRemotingServer.start(); // 启动服务器  
  
    // 其他初始化逻辑，比如注册服务等  
}

在这个例子中，run()方法首先创建了一个NettyRemotingServer实例，并设置了服务器的主机地址和端口号。然后，它调用start()方法来启动Netty服务器，这样Seata Server就可以监听来自其他节点的请求了。

总的来说

ServerRunner类在Seata Server中扮演着重要的角色，它负责初始化Netty通信渠道，为Seata节点间的通信提供基础设施。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持编程客栈(www.devze.com)。

继续阅读：CommandLineRunner异步任务机制 SpringBoot CommandLineRunner SpringBoot CommandLineRunner异步

SpringBoot CommandLineRunner的异步任务机制使用

目录

1.概要分析

1.1 手动创建线程

1.2 使用 Spring 的异步执行功能

2.核心原理分析

2.1 接口定义与功能

2.2 执行时机

2.3 命令行参数传递

2.4 实现与注册

2.5 执行顺序

2.6 与 ApplicationRunner 的区别

2.7 应用场景

3.部分源码分析

3.1 启动Spring Boot应用程序

3.2 调用所有的Runner实现类

4.典型的应用场景分析

4.1 TM（Transaction Manager，事务管理器）

4.2 RM（Resource Manager，资源管理器）

4.3 Seata RM和TM与Seata Server之间的RPC通信

4.4 Seata Server利用SpringBoot CommandLineRunner启动服务端通信渠道

总的来说

更多精彩内容

精彩评论

最新开发

SpringBoot整合Liquibase实现对数据库管理和迁移

java8升级java21方法步骤(完整pom文件升级及代码修改)

javacv ffmpeg使用原生ffmpeg命令方式

golang进行xml文件解析的操作方法

如何使用go实现创建WebSocket服务器

开发排行榜

springboot后端存储富文本内容的思路与步骤(含图片内容)

PyCharm运行python测试,报错“没有发现测试”/“空套件”的解决

return base64.b64encode(b).decode(

基于C语言实现钻石棋游戏的示例代码

Sublime Text 3解决中文乱码问题（实测可用）