java线程池ExecutorService超时处理小结

2024-09-27 10:09 开发作者：小百菜

方式一、使用 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 = nwww.devze.comew 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) {
                    php    // 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.LojavascriptcalDateTime;
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() + "】" + LocalDa编程客栈teTime.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 fina编程客栈l 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超时内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！

继续阅读：java ExecutorService超时 java线程池超时

java线程池ExecutorService超时处理小结

目录

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

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

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

更多精彩内容

精彩评论

最新开发

如何实现Python编写的图形界面可以自由拖动

mysql字符集引起的java.sql.SQLException:Incorrect string value:问题

解决Feign异步调用丢失上下文问题

python模拟重载初始化函数的方法详解

SQL注入攻击及其在SpringBoot中使用MyBatisPlus的防范策略的方法

开发排行榜

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

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

return base64.b64encode(b).decode(

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

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