java线程池ExecutorService超时处理小结

更新时间：2024年09月26日 10:23:04 作者：小百菜

使用ExecutorService时,设置子线程执行超时是一个常见需求,本文就来详细的介绍一下ExecutorService超时的三种方法,感兴趣的可以了解一下

方式一、使用 future.get() 来监听超时取消

这种办法看似能解决问题，但是当任务累积处理不过来时，会漏执行。
比如下面的例子，就实际只会执行一个子线程。

package com.study;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.*;

public class Test {
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final ExecutorService threadPool = Executors.newFixedThreadPool(1);

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 10; i++) {
            Future<?> future = threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(LocalDateTime.now().format(formatter));
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        // e.printStackTrace();
                    }
                }
            });
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        future.get(3, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    } catch (TimeoutException e) {//超时异常
                        future.cancel(true); // 超时后取消任务
                    }
                }
            }).start();
        }
    }
}

方式二、在子线程内部，超时后去发送中断信号

package com.study;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.*;

public class Test {
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final ExecutorService threadPool = Executors.newFixedThreadPool(1);
    private static final ScheduledExecutorService timeoutExecutor = new ScheduledThreadPoolExecutor(1);//监听超时，这个数量要和线程池数量相同

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 10; i++) {
            // int delay = 3;
            int delay = i + 1;
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    ScheduledFuture<?> schedule = null;
                    try {
                        Thread thread = Thread.currentThread();
                        // 启动一个定时器，如果任务执行超过3秒则中断当前线程
                        schedule = timeoutExecutor.schedule(() -> {
                            thread.interrupt(); // 中断当前正在执行的任务
                        }, delay, TimeUnit.SECONDS);
                        System.out.println(LocalDateTime.now().format(formatter));
                        Thread.sleep(5000);
                        // FileOutputStream fos = new FileOutputStream("d:/test.txt" + k);
                        // for (int j = 0; j < 1000000; j++) {
                        //     fos.write("123".getBytes());
                        // }
                        // fos.close();
                    } catch (InterruptedException e) {
                        // e.printStackTrace();
                    } finally {
                        if (schedule != null) {
                            //取消任务
                            schedule.cancel(true);
                        }
                    }
                }
            });
        }
    }
}

这里其实还是有问题，把 Thread.sleep(5000);改成注释的io阻塞，还是要等线程执行结束后才会取消线程执行。

所以单纯使用 future 是实现不了这个场景的逻辑的。

timeoutExecutor 数量和线程池数量要一致的原因如下示例。

package com.study;

import java.time.LocalDateTime;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ScheduledExecutorServiceExample {

    private static final ScheduledExecutorService timeoutExecutor = new ScheduledThreadPoolExecutor(2);

    public static void main(String[] args) throws InterruptedException {
        // 调用schedule方法两次
        scheduleTask("Task 1");
        scheduleTask("Task 2");
        scheduleTask("Task 3");
    }

    private static void scheduleTask(String taskName) {
        timeoutExecutor.schedule(() -> {
            System.out.println(taskName + " started at: " + LocalDateTime.now());
            try {
                // 模拟任务执行
                Thread.sleep(2000); // 假设每个任务执行2秒
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, 3, TimeUnit.SECONDS);
    }
}

方式三、自己定义锁来实现

package com.study;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.*;

public class Test {
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final ExecutorService threadPool = Executors.newFixedThreadPool(16);//这里不能设置为1了，这里已经不是用来控制并发数量了，只是为了重复利用线程
    private static final ScheduledExecutorService timeoutExecutor = new ScheduledThreadPoolExecutor(1);//监听超时，这个数量要和线程池数量相同

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 10; i++) {
            Thread.sleep(50);
            // int delay = 3;
            int delay = i + 1;
            int k = i;
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    ScheduledFuture<?> schedule = null;
                    try {
                        ThreadPool.awaitThread();
                        // 启动一个定时器，如果任务执行超过3秒则中断当前线程
                        // timeoutExecutor如果只有一个线程池，这里面的代码片段会阻塞，上一个线程在这里的代码片段执行完后，当前线程才会执行这里的代码片段，
                        // 但是影响不大，因为这里的代码片段只是释放动作，一瞬间就会执行完，所以影响不大，
                        // 如果其他场景这里阻塞时间比较久，那么timeoutExecutor线程大小要和threadPool线程大小一致。
                        schedule = timeoutExecutor.schedule(() -> {
                            System.out.println("释放1");
                            ThreadPool.releaseThread();
                        }, delay, TimeUnit.SECONDS);
                        System.out.println("【" + Thread.currentThread().getName() + "】" + LocalDateTime.now().format(formatter));
                        Thread.sleep(5000);
                        // FileOutputStream fos = new FileOutputStream("d:/test.txt" + k);
                        // for (int j = 0; j < 1000000; j++) {
                        //     fos.write("123".getBytes());
                        // }
                        // fos.close();
                    } catch (Exception e) {
                        System.out.println("异常");
                    } finally {
                        if (schedule != null) {
                            //cancel返回true任务还未执行，需要取消任务
                            if (schedule.cancel(true)) {
                                System.out.println("释放2");
                                ThreadPool.releaseThread();
                            }
                        }
                    }
                }
            });
        }
    }
}

package com.study;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 线程池
 */
public class ThreadPool {
    private static final int MAX_POOL_SIZE = 1; // 最大线程数，控制并发数量
    private static int totalThread = 0; // 总线程数
    private static final Lock lock = new ReentrantLock(true);
    private static final Condition notice = lock.newCondition();

    /**
     * 从线程池获取线程
     */
    public static boolean awaitThread() {
        lock.lock();
        try {
            // 尝试从线程池中获取线程
            if (totalThread < MAX_POOL_SIZE) {
                totalThread++;
                return true;
            }
            // 线程已到达最大线程数，等待归还线程，最长等待1小时，await()会释放当前线程的锁
            if (notice.await(1, TimeUnit.HOURS)) {
                totalThread++;
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return false;
    }


    /**
     * 释放线程到线程池
     */
    public static void releaseThread() {
        lock.lock();
        try {
            totalThread--;
            // 通知有空闲，signal()会唤醒其中一个await()线程
            notice.signal();
        } finally {
            lock.unlock();
        }
    }

}

到此这篇关于java线程池ExecutorService超时处理小结的文章就介绍到这了,更多相关java线程池ExecutorService超时内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

hadoop 切片机制分析与应用
切片这个词对于做过python开发的同学一定不陌生,但是与hadoop中的切片有所区别,hadoop中的切片是为了优化hadoop的job在处理过程中MapTask阶段的性能达到最优而言
2022-02-02
java开启远程debug竟有两种参数(最新推荐)
这篇文章主要介绍了java开启远程debug竟有两种参数,本文结合实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2023-07-07
SpringBoot3通过GraalVM生成exe执行文件问题
文章介绍了如何安装GraalVM和Visual Studio,并通过Spring Boot项目将Java应用程序封装成可执行文件（.exe）
2024-12-12
java实现将汉语转换为拼音功能
这篇文章主要介绍了java实现将汉语转换为拼音功能，非常不错，具有参考借鉴价值 ,需要的朋友可以参考下
2017-05-05
Java Arrays.sort()用法详解
这篇文章主要介绍了Java Arrays.sort()用法详解,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下
2021-08-08
SpringBoot使用Jasypt对YML文件配置内容加密的方法(数据库密码加密)
本文介绍了如何在SpringBoot项目中使用Jasypt对application.yml文件中的敏感信息（如数据库密码）进行加密,通过引入Jasypt依赖、配置加密密钥、加密敏感信息并测试解密功能,可以提高配置文件的安全性,减少因配置文件泄露导致的安全风险,感兴趣的朋友一起看看吧
2025-03-03
Java创建多线程的几种方式实现
这篇文章主要介绍了Java创建多线程的几种方式实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-10-10
Java多线程事务管理的实现
本文主要介绍了Java多线程事务管理的实现，文中通过示例代码介绍的非常详细，需要的朋友们下面随着小编来一起学习学习吧
2021-07-07
10k+点赞的 SpringBoot 后台管理系统教程详解
这篇文章主要介绍了10k+点赞的 SpringBoot 后台管理系统教程详解,本文通过图文并茂的形式给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-01-01
Java HttpClient实现socks代理的示例代码
这篇文章主要介绍了Java HttpClient 实现 socks 代理的示例代码，帮助大家更好的理解和使用Java，感兴趣的朋友可以了解下
2020-11-11