猫头鹰的深夜翻译：Spring线程 TaskExecutor

zhangyayun0

2019-06-29

关注关注

前言

在多线程中web应用很常见，尤其当你需要开发长期任务。

在Spring中，我们可以额外注意并使用框架已经提供的工具，而不是创造我们自己的线程。

Spring提供了TaskExecutor作为Executors的抽象。这个接口类似于java.util.concurrent.Executor接口。在spring中有许多预先开发好的该接口的实现，可以在官方文档中详细查看。

通过在Spring上下文中配置一个TaskExecutor的实现，你可以将你的TaskExecutor的实现注入到bean中，并可以在bean中访问到该线程池。

在bean中使用线程池的方式如下：

package com.gkatzioura.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;
import java.util.List;
/**
 * Created by gkatzioura on 4/26/17.
 */
@Service
public class AsynchronousService {
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private TaskExecutor taskExecutor;
    public void executeAsynchronously() {
        taskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //TODO add long running task
            }
        });
    }
}

配置

首先需要在Spring上下文中注册一个线程池。

package com.gkatzioura.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
/**
 * Created by gkatzioura on 4/26/17.
 */
@Configuration
public class ThreadConfig {
    @Bean
    public TaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(4);
        executor.setThreadNamePrefix("default_task_executor_thread");
        executor.initialize();
        return executor;
    }
}

这里配置了线程池的核心线程数，最大线程数和线程池中线程名称的前缀。

线程池配置完毕后，接下来的步骤就很简单了，将线程池注入到spring的component中，然后将Runnable任务提交给线程池来完成。

由于我们的异步代码可能需要与应用程序的其他组件交互并注入它们，因此一种不错的方法是创建Prototype范围的Runnable实例。(Prototype在每一次调用时会新建一个实例)

package com.gkatzioura;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
/**
 * Created by gkatzioura on 10/18/17.
 */
@Component
@Scope("prototype")
public class MyThread implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(MyThread.class);
    @Override
    public void run() {
        LOGGER.info("Called from thread");
    }
}

然后，我们将executors注入我们的服务并使用它来执行可运行的实例。

package com.gkatzioura.service;
import com.gkatzioura.MyThread;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;
import java.util.List;
/**
 * Created by gkatzioura on 4/26/17.
 */
@Service
public class AsynchronousService {
    @Autowired
    private TaskExecutor taskExecutor;
    @Autowired
    private ApplicationContext applicationContext;
    public void executeAsynchronously() {
        MyThread myThread = applicationContext.getBean(MyThread.class);
        taskExecutor.execute(myThread);
    }
}

Async关键字

通过spring提供的Async关键字，我们甚至无需将异步任务封装到Runnable类中，直接使用注解即可。

@Async
@Transactional
public void printEmployees() {
 
    List<Employee> employees = entityManager.createQuery("SELECT e FROM Employee e").getResultList();
    employees.stream().forEach(e->System.out.println(e.getEmail()));
}

该类会通过代理的方式提交给默认的线程池。

@Async
@Transactional
public CompletableFuture<List<Employee>> fetchEmployess() {
    List<Employee> employees = entityManager.createQuery("SELECT e FROM Employee e").getResultList();
    return CompletableFuture.completedFuture(employees);
}

该方法会返回异步执行的Future结果。

需要注意，如果要开启Async关键字，则需要在配置中添加EnableAsync信息。

package com.gkatzioura.config;
 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
import java.util.concurrent.Executor;
 
/**
 * Created by gkatzioura on 4/26/17.
 */
@Configuration
@EnableAsync
public class ThreadConfig {
 
    @Bean
    public TaskExecutor threadPoolTaskExecutor() {
 
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(4);
        executor.setThreadNamePrefix("sgfgd");
        executor.initialize();
 
        return executor;
    }
 
}