天津做网站推广的网站,博物馆网站做的最好的,免费开源的网站系统,win主机wordpress自定义线程池-初步了解
创建一个固定大小的线程池
在Java中#xff0c;你可以通过自定义线程池并指定线程的名称来实现你的需求。下面是一个简单的示例#xff0c;展示了如何创建一个固定大小的线程池#xff0c;并给每个线程指定一个名称#xff1a;
import java.util.…自定义线程池-初步了解
创建一个固定大小的线程池
在Java中你可以通过自定义线程池并指定线程的名称来实现你的需求。下面是一个简单的示例展示了如何创建一个固定大小的线程池并给每个线程指定一个名称
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class CustomThreadPool {public static void main(String[] args) {// 创建一个固定大小的线程池int poolSize 10;ExecutorService executor Executors.newFixedThreadPool(poolSize);// 为线程池中的每个线程指定名称for (int i 0; i poolSize; i) {executor.execute(() - {Thread.currentThread().setName(Thread- i);// 这里可以执行你的任务代码});}// 关闭线程池这不会立即关闭而是等待所有任务都完成executor.shutdown();}
}在上述代码中我们使用了Executors.newFixedThreadPool来创建一个固定大小的线程池。然后我们使用一个循环来为线程池中的每个线程执行任务并在任务中为每个线程指定一个名称。最后我们调用executor.shutdown()来关闭线程池。 请注意线程的名称只是用于标识和调试目的并不会影响线程的行为。另外如果你想在任务执行期间获取线程的名称你需要在适当的地方例如在任务代码中使用Thread.currentThread().getName()来获取它。
自定义ThreadPoolExecutor
ThreadPoolExecutor是Java中的一个线程池实现它提供了创建和管理线程池的功能。你可以通过继承ThreadPoolExecutor来自定义一个线程池。
以下是一个自定义的ThreadPoolExecutor的示例代码它设置了线程池的基本属性和参数
import java.util.concurrent.*;public class CustomThreadPoolExecutor extends ThreadPoolExecutor {public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueueRunnable workQueue) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);}// 重写方法自定义线程池的行为Overrideprotected void beforeExecute(Thread t, Runnable r) {// 在任务执行前执行的代码System.out.println(Before executing task: r.toString());}Overrideprotected void afterExecute(Runnable r, Throwable t) {// 在任务执行后执行的代码System.out.println(After executing task: r.toString());}Overrideprotected void terminated() {// 线程池终止时执行的代码System.out.println(ThreadPoolExecutor terminated.);}
}让我们详细解释一下每个方法和参数的作用
CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue)这是构造函数用于初始化线程池的基本属性。corePoolSize线程池的核心线程数。即使线程处于空闲状态也会保留的线程数。maximumPoolSize线程池允许的最大线程数。当队列满了并且已有线程数小于核心线程数则创建新的线程执行任务。keepAliveTime线程池中超过额定大小的线程在关闭之前等待新任务的最长时间。这里使用TimeUnit来指定时间单位。unitkeepAliveTime的时间单位。workQueue用于存放等待执行的任务的队列。当线程池中的线程数超过核心线程数时新任务会被放入此队列等待执行。beforeExecute(Thread t, Runnable r)在任务r开始执行之前调用的方法。可以在此添加自定义的预处理逻辑。afterExecute(Runnable r, Throwable t)在任务r执行结束之后调用的方法。可以在此添加自定义的清理逻辑。terminated()当ThreadPoolExecutor被终止时调用的方法。可以在此添加自定义的终止逻辑。 通过重写这些方法你可以在任务执行的生命周期的不同阶段添加自定义的行为以满足你的需求。 示例
上面自定义的CustomThreadPoolExecutor是一个线程池的实现你可以通过创建一个CustomThreadPoolExecutor对象来创建一个线程池并使用该线程池来执行任务。
以下是一个使用CustomThreadPoolExecutor的示例代码
import java.util.concurrent.*;public class CustomThreadPoolExecutorExample {public static void main(String[] args) {// 创建一个CustomThreadPoolExecutor对象CustomThreadPoolExecutor customThreadPoolExecutor new CustomThreadPoolExecutor(2, // 核心线程数4, // 最大线程数60L, // 保持活跃时间TimeUnit.SECONDS, // 时间单位new LinkedBlockingQueueRunnable() // 等待队列);// 提交任务给线程池执行for (int i 0; i 6; i) {customThreadPoolExecutor.execute(new Task(i));}// 关闭线程池customThreadPoolExecutor.shutdown();}
}class Task implements Runnable {private int taskId;public Task(int taskId) {this.taskId taskId;}Overridepublic void run() {System.out.println(Task taskId is running.);}
}在这个示例中我们首先创建了一个CustomThreadPoolExecutor对象并设置了它的核心线程数、最大线程数、保持活跃时间和等待队列。然后我们通过调用execute()方法向线程池提交了6个任务。每个任务是一个实现了Runnable接口的Task对象它的run()方法只是简单地输出一个消息。最后我们调用shutdown()方法关闭线程池。
在任务执行过程中CustomThreadPoolExecutor会自动管理线程的创建和销毁并保证任务按照提交的顺序执行。同时我们还重写了beforeExecute()、afterExecute()和terminated()方法分别在任务执行前、执行后和线程池终止时执行自定义的逻辑。例如在beforeExecute()方法中我们可以输出任务的信息以便观察任务的执行顺序。
ThreadPoolExecutor中有以下几个重要的指标1
核心线程数(corePoolSize)线程池中的常驻核心线程数即使没有任务需要执行核心线程也不会被回收。当有新任务提交时如果核心线程都在忙碌则会创建新的线程来处理任务。最大线程数(maximumPoolSize)线程池能够容纳同时执行的最大线程数。当工作队列满了并且活动线程数达到最大线程数时如果还有新任务提交线程池将创建新的线程来处理任务。但是超过最大线程数的线程可能会导致资源消耗过大。空闲线程存活时间(keepAliveTime)空闲线程存活时间指的是非核心线程在没有任务执行时的最长存活时间。当线程池中的线程数超过核心线程数且空闲时间达到设定值时多余的线程将被终止直到线程池中的线程数不超过核心线程数。时间单位(unit)时间单位是用于表示核心线程数和空闲线程存活时间的单位。常见的时间单位包括秒、毫秒、分钟等。工作队列(workQueue)用于存储待执行的任务。当线程池中的线程都在忙碌时新提交的任务将被添加到工作队列中等待执行。线程工厂(threadFactory)用于创建新线程。线程工厂提供了创建线程的方法可以自定义线程的名称、优先级等属性。拒绝策略(rejectedExecutionHandler)定义了当线程池无法接受新任务时的处理策略。当工作队列已满且线程池中的线程数已达到最大线程数时新任务将被拒绝执行。
在运行过程中要监控线程池中线程的各个指标需要监控以下指标
核心线程数监控常驻核心线程数。最大线程数监控能够容纳同时执行的最大线程数。当前线程数监控当前正在执行任务的线程数。队列中任务数监控工作队列中待执行的任务数。已完成任务数监控已经执行完成的任务数。总任务数监控总共提交的任务数。线程空闲时间监控线程在没有任务执行时的空闲时间。线程异常数监控线程执行任务时抛出的异常数。
要获取线程池中线程的各个指标可以通过以下方法
使用线程池自带的监控工具例如 ThreadPoolExecutor 中的 ThreadPoolExecutor.getPoolSize(), ThreadPoolExecutor.getActiveCount(), ThreadPoolExecutor.getTaskCount(), ThreadPoolExecutor.getCompletedTaskCount() 等方法。使用线程池中的线程的 API例如 Thread.getState(), Thread.getName(), Thread.getStackTrace() 等方法获取线程的各个状态信息。使用第三方监控工具例如 JMX、VisualVM 等工具来监控线程池的状态信息。自定义监控程序通过编写程序来获取线程池的状态信息例如通过网络连接、文件读取等方式获取线程池的状态信息。
要编写一套详细的自定义监控程序需要完成以下步骤
确定监控指标根据需求确定需要监控的指标例如线程池中的核心线程数、最大线程数、当前线程数、队列中任务数、已完成任务数、总任务数、线程空闲时间、线程异常数等。设计数据采集方式根据监控指标设计数据采集方式例如编写程序通过调用线程池的 API 来获取状态信息或者通过网络连接、文件读取等方式获取线程池的状态信息。编写数据采集程序根据设计的数据采集方式编写程序实现数据采集功能。存储数据将采集到的数据存储到数据库或文件中以便后续分析。分析数据对采集到的数据进行统计分析例如计算平均响应时间、成功率等指标并根据分析结果提供相应的建议或警告。展示数据将分析后的数据通过图表或报告等方式展示出来以便用户直观地了解线程池的状态信息。
下面是一个简单的示例程序用于监控线程池的状态信息
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;public class ThreadPoolMonitor implements Runnable {private ThreadPoolExecutor threadPool;private long interval;private TimeUnit unit;public ThreadPoolMonitor(ThreadPoolExecutor threadPool, long interval, TimeUnit unit) {this.threadPool threadPool;this.interval interval;this.unit unit;}Overridepublic void run() {while (true) {// 获取线程池状态信息int corePoolSize threadPool.getCorePoolSize();int maximumPoolSize threadPool.getMaximumPoolSize();int activeCount threadPool.getActiveCount();int completedTaskCount threadPool.getCompletedTaskCount();int taskCount threadPool.getTaskCount();long keepAliveTime threadPool.getKeepAliveTime(TimeUnit.MILLISECONDS);TimeUnit unit threadPool.getKeepAliveTimeUnit();String workQueue threadPool.getQueue().getClass().getName();ThreadFactory threadFactory threadPool.getThreadFactory();RejectedExecutionHandler rejectedExecutionHandler threadPool.getRejectedExecutionHandler();// 输出状态信息System.out.println(new Date() CorePoolSize: corePoolSize , MaximumPoolSize: maximumPoolSize , ActiveCount: activeCount , CompletedTaskCount: completedTaskCount , TaskCount: taskCount , KeepAliveTime: keepAliveTime unit , WorkQueue: workQueue , ThreadFactory: threadFactory , RejectedExecutionHandler: rejectedExecutionHandler);try {// 休眠一段时间等待下一次采集Thread.sleep(interval);} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) {ThreadPoolExecutor threadPool (ThreadPoolExecutor) Executors.newFixedThreadPool(10);ThreadPoolMonitor monitor new ThreadPoolMonitor(threadPool, 1000, TimeUnit.MILLISECONDS);new Thread(monitor).start();}
}要获取达到最大线程数后被拒绝的线程数或异常数可以通过线程池的 getRejectedExecutionHandler() 方法获取 RejectedExecutionHandler 对象然后根据具体情况实现自定义的拒绝策略。
Java 中提供了几种默认的拒绝策略
ThreadPoolExecutor.AbortPolicy: 默认的拒绝策略直接抛出 RejectedExecutionException 异常。ThreadPoolExecutor.CallerRunsPolicy: 调用者线程运行被拒绝的任务不抛出异常。ThreadPoolExecutor.DiscardPolicy: 忽略被拒绝的任务不保存也不抛出异常。ThreadPoolExecutor.DiscardOldestPolicy: 丢弃队列中最老的任务然后尝试重新提交。ThreadPoolExecutor.WaitForTasksPolicy: 等待当前任务队列的任务都完成后再执行被拒绝的任务。
ThreadPoolExecutor.CallerRunsPolicy 如果以上策略都不满足需求可以自定义实现 RejectedExecutionHandler 接口根据具体需求进行处理。例如可以统计被拒绝的线程数或异常数或者将任务保存到数据库或文件等地方以便后续处理。 ThreadPoolExecutor.CallerRunsPolicy是Java线程池中的一个拒绝策略当线程池拒绝一个新任务时它会调用该策略来处理被拒绝的任务。 当拒绝策略为CallerRunsPolicy时线程池会调用执行者所在的线程来执行被拒绝的任务也就是说直接在调用execute方法的线程中运行被拒绝的任务。如果执行程序已关闭则会丢弃该任务。这种策略会降低对于新任务提交速度影响程序的整体性能。如果您的应用程序可以承受此延迟并且要求任何一个任务请求都要被执行可以选择该策略。
以下是一个简单的示例代码演示如何获取达到最大线程数后被拒绝的线程数
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.RejectedExecutionHandler;public class ThreadPoolMonitor implements Runnable {private ThreadPoolExecutor threadPool;private long interval;private TimeUnit unit;public ThreadPoolMonitor(ThreadPoolExecutor threadPool, long interval, TimeUnit unit) {this.threadPool threadPool;this.interval interval;this.unit unit;}Overridepublic void run() {while (true) {// 获取线程池状态信息int corePoolSize threadPool.getCorePoolSize();int maximumPoolSize threadPool.getMaximumPoolSize();int activeCount threadPool.getActiveCount();int completedTaskCount threadPool.getCompletedTaskCount();int taskCount threadPool.getTaskCount();long keepAliveTime threadPool.getKeepAliveTime(TimeUnit.MILLISECONDS);TimeUnit unit threadPool.getKeepAliveTimeUnit();String workQueue threadPool.getQueue().getClass().getName();ThreadFactory threadFactory threadPool.getThreadFactory();RejectedExecutionHandler rejectedExecutionHandler threadPool.getRejectedExecutionHandler();int rejectedCount 0; // 被拒绝的线程数if (rejectedExecutionHandler instanceof ThreadPoolExecutor.AbortPolicy) {rejectedCount ((ThreadPoolExecutor.AbortPolicy) rejectedExecutionHandler).getRejectedExecutionCount();} else if (rejectedExecutionHandler instanceof ThreadPoolExecutor.CallerRunsPolicy) {rejectedCount ((ThreadPoolExecutor.CallerRunsPolicy) rejectedExecutionHandler).getRunCount();} else if (rejectedExecutionHandler instanceof ThreadPoolExecutor.DiscardPolicy) {rejectedCount ((ThreadPoolExecutor.DiscardPolicy) rejectedExecutionHandler).getDiscardCount();} else if (rejectedExecutionHandler instanceof ThreadPoolExecutor.DiscardOldestPolicy) {rejectedCount ((ThreadPoolExecutor.DiscardOldestPolicy) rejectedExecutionHandler).getDiscardCount();} else if (rejectedExecutionHandler instanceof ThreadPoolExecutor.WaitForTasksPolicy) {rejectedCount ((ThreadPoolExecutor.WaitForTasksPolicy) rejectedExecutionHandler).getTaskCount();}System.out.println(new Date() CorePoolSize: corePoolSize , MaximumPoolSize: maximumPoolSize , ActiveCount: activeCount , CompletedTaskCount: completedTaskCount , TaskCount: taskCount , KeepAliveTime: keepAliveTime unit , WorkQueue: workQueue , ThreadFactory: threadFactory , RejectedExecutionHandler: rejectedExecutionHandler , RejectedCount: rejectedCount);try {// 休眠一段时间等待下一次采集Thread.sleep(interval);} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) {ThreadPoolExecutor threadPool (ThreadPoolExecutor) Executors.newFixedThreadPool(10);ThreadPoolMonitor monitor new ThreadPoolMonitor(threadPool, 1000, TimeUnit.MILLISECONDS}下面是一个关于线程池中线程各个指标的含义的表格
指标含义maximumPoolSize最大线程数当线程池繁忙时最多可以拥有的线程数corePoolSize核心线程数线程池中始终保持的线程数minimumPoolSize最小线程数线程池中线程数的下限值poolSize当前线程数线程池中正在运行的线程数activeCount活动线程数当前正在执行任务的线程数taskCount任务数线程池中已提交的任务总数completedTaskCount完成的任务数线程池中已经完成的任务总数largestPoolSize最大的线程数线程池曾拥有的最大线程数keepAliveTime空闲线程存活时间当线程超过该时间没有任务时就会被回收TimeUnit时间单位用于描述参数3的单位如秒、毫秒等BlockingQueue任务队列用于保存待执行任务的容器ThreadFactory线程工厂用于创建线程池中线程的工厂方法可以通过它来设置线程的命名规则、优先级和线程类型RejectedExecutionHandler拒绝策略当任务量超过线程池可以保存的最大任务数时执行的策略
CompletableFutureThreadPoolThread实现并发异步处理
CompletableFuture 是 Java 8 引入的一个强大的并发工具它允许你以异步的方式处理任务并且可以方便地组合多个异步任务。ThreadPoolExecutor 则是 Java 提供的一个线程池工具它可以根据需要创建和管理线程以支持并发执行任务。
下面是一个使用 CompletableFuture 和 ThreadPoolExecutor 实现并发异步处理的示例
import java.util.concurrent.*;public class CompletableFutureWithThreadPool {public static void main(String[] args) {// 创建一个固定大小的线程池ThreadPoolExecutor executor new ThreadPoolExecutor(5, // 核心线程数10, // 最大线程数60, // 线程空闲超过60秒则销毁TimeUnit.SECONDS, // 时间单位new LinkedBlockingQueueRunnable() // 任务队列);// 使用 CompletableFuture 来执行异步任务CompletableFutureString future1 CompletableFuture.supplyAsync(() - {// 模拟一个耗时操作try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}return 任务1完成;}, executor);CompletableFutureString future2 CompletableFuture.supplyAsync(() - {// 模拟一个耗时操作try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}return 任务2完成;}, executor);CompletableFutureVoid all CompletableFuture.allOf(future1, future2);all.thenRun(() - {System.out.println(所有任务都已完成);executor.shutdown(); // 关闭线程池});}
}在上面的示例中我们首先创建了一个 ThreadPoolExecutor并指定了其核心线程数、最大线程数、线程空闲超时时间以及任务队列。然后我们使用 CompletableFuture.supplyAsync 方法来执行两个异步任务这两个任务会使用我们创建的线程池来执行。每个异步任务都是一个耗时操作模拟了某些需要长时间运行的任务。最后我们使用 CompletableFuture.allOf 方法来等待两个异步任务都完成当所有任务都完成后我们输出一条消息并关闭线程池。 通过这个示例你可以看到如何使用 CompletableFuture 和 ThreadPoolExecutor 来实现并发异步处理。你可以根据实际需求调整线程池的参数以及异步任务的逻辑。
注意 是的对于一直在运行的线上业务流程通常不需要显式地关闭线程池。线程池会自动管理线程的生命周期根据需要创建和销毁线程。当提交任务到线程池时线程池会根据其配置的参数决定是否创建新的线程来执行任务。如果当前线程数已经达到池中线程数的上限则任务会等待直到有空闲线程可用。 一般情况下线程池会一直运行直到没有任务提交或者显式地关闭线程池。即使没有任务提交线程池中的线程也不会立即被销毁而是会保持空闲状态等待新的任务到来。 然而需要注意的是长时间运行的线程池可能会占用系统资源并且如果存在长时间运行的任务可能会阻塞其他任务的执行。因此在设计和使用线程池时应该根据业务需求和系统资源情况进行合理的配置和监控以确保系统的稳定性和性能。 如果你的程序一直在运行并且线程池已经调用了shutdown方法那么线程池中的线程将不再接受新的任务。此时如果你尝试提交新的任务给线程池将会抛出RejectedExecutionException异常。 如果你希望在程序运行时能够暂停线程池的执行而不是完全关闭线程池你可以使用线程池的shutdown或shutdownNow方法。这两个方法都会停止接受新的任务但shutdownNow方法会尝试停止所有正在执行的任务而shutdown方法则会等待已提交的任务执行完毕后再停止。 如果你希望在程序运行时能够动态地调整线程池的大小或改变线程池的行为可以考虑使用可扩展的线程池或配置更灵活的线程池实现例如ForkJoinPool或ExecutorService的定制实现。这些线程池可以更好地适应不同的应用程序需求。
果你的程序一直在运行并且线程池已经调用了shutdown方法那么线程池中的线程将不再接受新的任务。此时如果你尝试提交新的任务给线程池将会抛出RejectedExecutionException异常。 如果你希望在程序运行时能够暂停线程池的执行而不是完全关闭线程池你可以使用线程池的shutdown或shutdownNow方法。这两个方法都会停止接受新的任务但shutdownNow方法会尝试停止所有正在执行的任务而shutdown方法则会等待已提交的任务执行完毕后再停止。 如果你希望在程序运行时能够动态地调整线程池的大小或改变线程池的行为可以考虑使用可扩展的线程池或配置更灵活的线程池实现例如ForkJoinPool或ExecutorService的定制实现。这些线程池可以更好地适应不同的应用程序需求。
