Java多线程 - 利用Callable或CompletableFuture实现多线程异步任务执行

news/2024/11/16 7:33:33/

文章目录

      • 1. Callable接口源码
      • 2. Future接口的源码
      • 3. RunnableFuture接口和FutureTask实现类
      • 4. 利用线程池和Callable接口实现异步执行任务
      • 5. 利用CompleteFutable实现多线程异步任务执行

1. Callable接口源码

@FunctionalInterface
public interface Callable<V> {// 这个call()方法有返回值，且声明了受检异常，可以直接抛出Exception异常V call() throws Exception;
}

2. Future接口的源码

public interface Future<V> {// 取消异步执行中的任务boolean cancel(boolean mayInterruptIfRunning);boolean isCancelled();// 判断异步任务是否执行成功boolean isDone();// 获取异步任务完成后的结果V get() throws InterruptedException, ExecutionException;V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
}

Future接口中的方法：

get()：获取异步任务执行的结果。注意，这个方法的调用是阻塞性的。如果异步任务没有执行完成，异步结果获取线程（调用线程）会一直被阻塞，一直阻塞到异步任务执行完成，其异步结果返回给调用线程。

get(Long timeout,TimeUnit unit)：设置时限，（调用线程）阻塞性地获取异步任务执行的结果。该方法的调用也是阻塞性的，但是结果获取线程（调用线程）会有一个阻塞时长限制，不会无限制地阻塞和等待，如果其阻塞时间超过设定的timeout时间，该方法将抛出异常，调用线程可捕获此异常。

boolean isDone()：获取异步任务的执行状态。如果任务执行结束，就返回true。

boolean isCancelled()：获取异步任务的取消状态。如果任务完成前被取消，就返回true。

boolean cancel(boolean mayInterruptRunning)：取消异步任务的执行。

3. RunnableFuture接口和FutureTask实现类

如何使用Callable接口创建线程呢？

public interface RunnableFuture<V> extends Runnable, Future<V> {void run();
}

RunnableFuture只是一个接口，无法直接创建对象，如果需要创建对象，就需用到它的实现类——FutureTask。

public class FutureTask<V> implements RunnableFuture<V> {private Callable<V> callable;// 构造方法public FutureTask(Callable<V> callable) {if (callable == null) throw new NullPointerException();// callable实例属性需要在FutureTask实例构造时进行初始化this.callable = callable;this.state = NEW;       }
}

RunnableFuture接口实现了2个目标：

(1) RunnableFuture通过继承Runnable接口，从而保证了其实例可以作为Thread线程实例的target目标；
(2) RunnableFuture通过继承Future接口，从而保证了可以获取未来的异步执行结果。

首先，通过实现Runnable接口的方式创建一个异步执行任务：

public class CallableTaskDemo implements Callable {// call()方法有返回值，并且可以抛出Exception异常@Overridepublic String call() throws Exception {System.out.println("实现Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回线程执行结果";}
}

方式1：通过Thread类创建线程执行异步任务

public class CallableDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {// 创建异步执任务实例Callable callable = new CallableTaskDemo();// 初始化callable实例属性FutureTask futureTask = new FutureTask(callable);// 创建线程执行异步任务Thread thread = new Thread(futureTask);thread.start();System.out.println("获取异步执行任务结果："+futureTask.get());// 获取异步执行任务结果：返回线程执行结果}
}

方式2：通过线程池创建线程，并提交异步执行任务：

public class CallableDemo2 {// 通过线程池创建3个线程private static ExecutorService executorService = Executors.newFixedThreadPool(3);public static void main(String[] args) throws ExecutionException, InterruptedException {// 通过线程池的submit()方法提交异步执行任务，有返回结果Future submit = executorService.submit(new CallableTaskDemo());// 获取返回结果System.out.println(submit.get());}
}

对于Calleble来说，Future和FutureTask均可以用来获取任务执行结果，不过Future是个接口，FutureTask是Future的具体实现，而且FutureTask还间接实现了Runnable接口，也就是说FutureTask可以作为Runnable任务提交给线程池。

4. 利用线程池和Callable接口实现异步执行任务

1、定义3个线程执行任务：

public class FirstCallableTask implements Callable<String> {// call()方法有返回值，并且可以抛出Exception异常@Overridepublic String call() throws Exception {System.out.println("实现 First Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 First Callable 接口线程执行结果";}
}public class SecondCallableTask implements Callable<String> {@Overridepublic String call() throws Exception {System.out.println("实现 Second Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Second Callable 接口线程执行结果";}
}public class ThirdCallableTask implements Callable<String> {@Overridepublic String call() throws Exception {System.out.println("实现 Third Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Third Callable 接口线程执行结果";}
}

2、定义线程池提交线程任务：

@Service
@Slf4j
public class BeanLoadService implements ApplicationContextAware, ApplicationListener<ContextStoppedEvent> {private ApplicationContext applicationContext;// 定义一个线程池private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,7,4,TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadFactoryBuilder().setNamePrefix(BeanLoadService.class.getSimpleName() + "-pool-%d").setDaemon(true).build(),new ThreadPoolExecutor.DiscardOldestPolicy());public Map<String,Integer> richInfo() {Callable<String> firstCallable = new FirstCallableTask();Callable<String> secondCallable = new SecondCallableTask();Callable<String> thirdCallable = new ThirdCallableTask();List<Callable<String>> callableList = new ArrayList<>();callableList.add(firstCallable);callableList.add(secondCallable);callableList.add(thirdCallable);try {List<Future<String>> futures = THREAD_POOL_EXECUTOR.invokeAll(callableList, 1, TimeUnit.MINUTES);for (Future<String> future : futures) {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}}} catch (Exception e) {log.warn("incident delay data warn ", e);}}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextStoppedEvent contextStoppedEvent) {try {THREAD_POOL_EXECUTOR.shutdown();} catch (Exception e) {log.error("停止线程池失败", e);}}
}

3、测试：

@SpringBootTest
@RunWith(SpringRunner.class)
public class BeanLoadServiceTest {@Autowiredprivate BeanLoadService beanLoadService;@Testpublic void test(){beanLoadService.richInfo();}
}

实现 First Callable接口来编写异步执行任务
实现 Second Callable接口来编写异步执行任务
实现 Third Callable接口来编写异步执行任务
返回 First Callable 接口线程执行结果
返回 Second Callable 接口线程执行结果
返回 Third Callable 接口线程执行结果

4、定义线程池提交线程任务（内部类）：

@Service
@Slf4j
public class BeanLoadService implements ApplicationContextAware, ApplicationListener<ContextStoppedEvent> {private ApplicationContext applicationContext;// 定义一个线程池private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,7,4,TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadFactoryBuilder().setNamePrefix(BeanLoadService.class.getSimpleName() + "-pool-%d").setDaemon(true).build(),new ThreadPoolExecutor.DiscardOldestPolicy());public Map<String,Integer> richInfo() {Callable<String> firstCallable = getFirstCallable();Callable<String> secondCallable = getSecondCallable();Callable<String> thirdCallable = getThirdCallable();List<Callable<String>> callableList = new ArrayList<>();callableList.add(firstCallable);callableList.add(secondCallable);callableList.add(thirdCallable);try {List<Future<String>> futures = THREAD_POOL_EXECUTOR.invokeAll(callableList, 1, TimeUnit.MINUTES);for (Future<String> future : futures) {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}}} catch (Exception e) {log.warn("incident delay data warn ", e);}}private Callable<String> getThirdCallable() {Callable<String> callable = new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("实现 Third Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Third Callable 接口线程执行结果";}};}private Callable<String> getSecondCallable() {Callable<String> callable = new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("实现 Second Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Second Callable 接口线程执行结果";};};return callable;}private Callable<String> getFirstCallable() {Callable<String> callable = new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("实现 First Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 First Callable 接口线程执行结果";}};return callable;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextStoppedEvent contextStoppedEvent) {try {THREAD_POOL_EXECUTOR.shutdown();} catch (Exception e) {log.error("停止线程池失败", e);}}
}

5、定义线程池提交任务：（Java 8）

@Service
@Slf4j
public class BeanLoadService implements ApplicationContextAware, ApplicationListener<ContextStoppedEvent> {private ApplicationContext applicationContext;// 定义一个线程池private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,7,4,TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadFactoryBuilder().setNamePrefix(BeanLoadService.class.getSimpleName() + "-pool-%d").setDaemon(true).build(),new ThreadPoolExecutor.DiscardOldestPolicy());public void richInfo() {Callable<String> firstCallable = getFirstCallable();Callable<String> secondCallable = getSecondCallable();Callable<String> thirdCallable = getThirdCallable();List<Callable<String>> callableList = new ArrayList<>();callableList.add(firstCallable);callableList.add(secondCallable);callableList.add(thirdCallable);try {List<Future<String>> futures = THREAD_POOL_EXECUTOR.invokeAll(callableList, 1, TimeUnit.MINUTES);for (Future<String> future : futures) {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}}} catch (Exception e) {log.warn("incident delay data warn ", e);}}private Callable<String> getThirdCallable() {Callable<String> callable = ()->{System.out.println("实现 Third Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Third Callable 接口线程执行结果";};return callable;}private Callable<String> getSecondCallable() {Callable<String> callable = ()-> {System.out.println("实现 Second Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Second Callable 接口线程执行结果";};return callable;}private Callable<String> getFirstCallable() {Callable<String> callable  = () -> {System.out.println("实现 First Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 First Callable 接口线程执行结果";};return callable;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextStoppedEvent contextStoppedEvent) {try {THREAD_POOL_EXECUTOR.shutdown();} catch (Exception e) {log.error("停止线程池失败", e);}}
}

5. 利用CompleteFutable实现多线程异步任务执行

在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture， 提供了非常强大的Future 的扩展功能， 可以帮助我们简化异步编程的复杂性， 提供了函数式编程的能力， 可以通过回调的方式处理计算结果， 并且提供了转换和组合 CompletableFuture 的方法。CompletableFuture 类实现了 Future 接口， 所以你还是可以像以前一样通过get方法阻塞或者轮询的方式获得结果， 但是这种方式不推荐使用。

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类， 都可以获取线程的执行结果。

CompletableFuture 提供了四个静态方法来创建一个异步操作 ：

runXxxx 都是没有返回结果的， supplyXxx 都是可以获取返回结果的，可以传入自定义的线程池， 否则就用默认的线程池 。

//子任务包装一个Runnable实例，并调用ForkJoinPool.commonPool()线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable)//子任务包装一个Runnable实例，并调用指定的executor线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)//子任务包装一个Supplier实例，并调用ForkJoinPool.commonPool()线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)//子任务包装一个Supplier实例，并使用指定的executor线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

@Service
@Slf4j
public class BeanLoadService {public void getInfo() {List<CompletableFuture<String>> futures = new ArrayList<>();CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "返回 First Callable 接口线程执行结果";});futures.add(firstFuture);CompletableFuture<String> secondFuture = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "返回 Second Callable 接口线程执行结果";});futures.add(secondFuture);CompletableFuture<String> thirdFuture = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "返回 Third Callable 接口线程执行结果";});futures.add(thirdFuture);try {futures.forEach(future -> {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}});} catch (Exception e) {log.warn("incident delay data warn ", e);}}
}

http://www.ppmy.cn/news/26695.html

相关文章

三次握手四次挥手详细解析面试常问

三次握手四次挥手详细解析面试常问

文章目录1.第2次握手传回了ACK&#xff0c;为什么还要传回SYN&#xff1f;2.断开连接-TCP 四次挥手3.为什么要四次挥手&#xff1f;4.为什么不能把服务器发送的 ACK 和 FIN 合并起来&#xff0c;变成三次挥手&#xff1f;5.如果第二次挥手时服务器的 ACK 没有送达客户端&#x…
阅读更多...
PCI设备驱动初探（仅仅是内核部分，不是具体设备驱动）

PCI设备驱动初探（仅仅是内核部分，不是具体设备驱动）

在操作系统中&#xff0c;声卡、网卡之类的设备驱动并不像硬盘、鼠标、键盘等等驱动直接编写就行了。它们是建立在内核PCI驱动基础上的&#xff0c;也就是说这类设备通过PCI总线与系统通信。所以要编写这类的驱动首先要构造一个PCI设备的内核驱动&#xff0c;这样我们才能继续正…
阅读更多...
模拟用户登录-课后程序(JAVA基础案例教程-黑马程序员编著-第五章-课后作业)

模拟用户登录-课后程序(JAVA基础案例教程-黑马程序员编著-第五章-课后作业)

【案例5-3】 模拟用户登录 【案例介绍】 1.任务描述 在使用一些APP时&#xff0c;通常都需要填写用户名和密码。用户名和密码输入都正确才会登录成功&#xff0c;否则会提示用户名或密码错误。 本例要求编写一个程序&#xff0c;模拟用户登录。程序要求如下&#xff1a; 用…
阅读更多...
力扣mysql刷题记录

力扣mysql刷题记录

mysql刷题记录 刷题链接https://leetcode.cn/study-plan/sql/?progressjkih0qc mysql冲&#xff01;mysql刷题记录一. 1699. 两人之间的通话次数题解二、1251. 平均售价题解三. 1571. 仓库经理题解四.1445. 苹果和桔子解五.1193. 每月交易 I题解六.1633. 各赛事的用户注册率题…
阅读更多...
如何使用Hugo Academic Theme构建自己的github主页

如何使用Hugo Academic Theme构建自己的github主页

前期条件 自己已经注册好GitHub 搜索Hugo Academic Theme&#xff08;网址&#xff09; 进入后的网址为&#xff1a;https://academic-demo.netlify.app/ 点击Get Start 出现如下模板&#xff0c;选择一个喜欢的模板&#xff0c;点击“START WITH ACADEMIC RESUME” 点击 …
阅读更多...
calico-kube-controllers 启动失败处理

calico-kube-controllers 启动失败处理

故障描述calico-kube-controllers 异常&#xff0c;不断重启日志信息如下2023-02-21 01:26:47.085 [INFO][1] main.go 92: Loaded configuration from environment config&config.Config{LogLevel:"info", WorkloadEndpointWorkers:1, ProfileWorkers:1, PolicyW…
阅读更多...
CAD正式学习（一）

CAD正式学习（一）

CAD正式学习&#xff08;一&#xff09;&#xff08;23.2.20&#xff09; CAD简介 CAD是Autodesk&#xff08;欧特克&#xff09;公司首次于1982年开发的自动计算机辅助软件&#xff0c;主要用于二维绘图、详细绘制、设计文档和基本三维设计&#xff0c;是广为流行的绘图工具…
阅读更多...
Web前端：什么是Vue Native 框架?有什么特点?

Web前端：什么是Vue Native 框架?有什么特点?

Vue Native是一个使用Vue.Js开发本地移动应用程序的框架。该框架将文档转换为React Native&#xff0c;进而为你提供适用于Android和iOS的本地应用程序。实际上&#xff0c;Vue Native应用程序据说是React API的包装。Vue将Vue.js和React结合在一起&#xff0c;让你的开发团队充…
阅读更多...
最新文章

Copyright @ 2022~2023