当前位置： 首页 > article >正文

java-07 多线程-并发编程(线程池，线程状态)

article 2024/10/11 14:16:05

  并发编程是指在一个程序中同时执行多个任务或线程。这通常涉及到多线程编程、线程同步、并发容器等技术。这些技术可以用来解决多线程环境中的问题，如线程安全、资源竞争、死锁等问题。在实际的Java并发编程中，还需要考虑到线程池、Future、Callable、ExecutorService等概念。

在另一篇文章介绍了多线程、进程、并发、并行等基本概念，并分析了线程安全问题产生的原因，同时也整理了线程实现的4种方式，并做了对比，请参考 java-06 多线程-4种实现方式

如果你觉得我分享的内容或者我的努力对你有帮助，或者你只是想表达对我的支持和鼓励，请考虑给我点赞、评论、收藏。您的鼓励是我前进的动力，让我感到非常感激。

文章目录

  • 2 线程池
    • 2.1 ThreadPoolExecutor
    • 2.2 Executors
  • 3 线程状态

2 线程池

2.1 ThreadPoolExecutor

Java 中的线程池接口为 ExecutorService，一个常用的实现类为 ThreadPoolExecutor，其构造函数为：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit,
                   BlockingQueue<Runnable> workQueue,
                   ThreadFactory threadFactory,
                   RejectedExecutionHandler handler)

参数说明：

  • corePoolSize：线程池的核心线程数，即任务队列未达到队列容量时，最大可以同时运行的线程数量。即使线程是空闲的，它们也不会被销毁，除非线程池被关闭。
  • maximumPoolSize：线程池的最大线程数。在没有核心线程空闲的情况下，如果任务数量增加，线程池可以扩展到最大线程数。如果任务数量继续增加，超过线程池最大大小的任务将会被拒绝执行。
  • keepAliveTime：非核心线程的最大空闲时间。当线程池中的线程数量超过 corePoolSize，多余的非核心线程会在空闲时间超过 - keepAliveTime 后被销毁，以减少资源占用。
  • unit：时间单位，用于指定 keepAliveTime 的时间单位。
  • workQueue：用于存储等待执行的任务的阻塞队列。当所有核心线程都忙碌时，新任务将被放入队列等待执行。常用的队列类型包括 - LinkedBlockingQueue（最大长度为 Integer.MAX_VALE，即无界队列）、ArrayBlockingQueue（有界队列）、PriorityBlockingQueue（基于堆的优先级队列）等。
  • threadFactory：用于创建线程的工厂。可以通过提供自己实现的 ThreadFactory 自定义线程的创建过程。
  • handler：拒绝策略，用于处理无法提交给线程池执行的任务。当任务数量超过线程池最大大小且队列已满时，将使用拒绝策略处理任务。

新任务拒绝策略：

ThreadPoolExecutor.AbortPolicy丢弃任务并抛出 RejectedExecutionException 异常，是默认的策略
ThreadPoolExecutor.DiscardPolicy丢弃任务，但是不抛出异常，这是不推荐的做法
ThreadPoolExecutor.DiscardOldestPolicy抛弃队列中等待最久的任务，然后把当前任务加入队列中
ThreadPoolExecutor.CallerRunsPolicy由主线程负责调用任务的 run() 方法从而绕过线程池直接执行

注意事项：

  • 新任务提交时发现核心线程都在忙，任务队列也满了，并且还可以创建临时线程，此时才会创建临时线程。
  • 核心线程和临时线程都在忙，任务队列也满了，新的任务过来的时候才会开始拒绝任务。

常用方法：

方法名称说明
void execute(Runnable command)执行 Runnable 任务
Future submit(Callable task)执行 Callable 任务，返回一个 Future 对象，用于获取线程返回的结果
void shutdown()等全部任务执行完毕后，再关闭线程池
List shutdownNow()立刻关闭线程池，停止正在执行的任务，并返回队列中未执行的任务

ThreadPoolExecutor 的基本使用示例：

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        System.out.printf("[%s] %s\n", Thread.currentThread().getName(),
                          LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

public class Main {
    public static void main(String[] args) {
        ExecutorService pool = new ThreadPoolExecutor(2, 3,
                                                      8, TimeUnit.SECONDS,
                                                      new ArrayBlockingQueue<>(2),
                                                      Executors.defaultThreadFactory(),
                                                      new ThreadPoolExecutor.AbortPolicy());

        Runnable target = new MyRunnable();

        pool.execute(target); // 核心线程
        pool.execute(target); // 核心线程
        pool.execute(target); // 任务队列等待
        pool.execute(target); // 任务队列等待
        pool.execute(target); // 任务队列满，启动一个临时线程
        pool.execute(target); // 核心线程和临时线程忙，同时任务队列已满，拒绝任务

        pool.shutdown();
    }
}

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task atreus.ink.MyRunnable@7a0ac6e3 rejected from java.util.concurrent.ThreadPoolExecutor@71be98f5[Running, pool size = 3, active threads = 3, queued tasks = 2, completed tasks = 0]
	at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
	at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
	at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
	at atreus.ink.Main.main(Main.java:20)
[pool-1-thread-2] 2023-08-30 15:57:44
[pool-1-thread-1] 2023-08-30 15:57:44
[pool-1-thread-3] 2023-08-30 15:57:44
[pool-1-thread-1] 2023-08-30 15:57:47
[pool-1-thread-2] 2023-08-30 15:57:47

使用 submit() 以捕获异常：

public class ThreadPoolExceptionHandling {

    @SuppressWarnings("all")
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(1);

        Future<Integer> future = executorService.submit(() -> {
            throw new RuntimeException("Exception in task");
        });

        try {
            // 调用 get() 方法获取任务执行结果，如果任务抛出了异常，这里会抛出 ExecutionException
            Integer result = future.get();
            System.out.println("Task result: " + result);
        } catch (InterruptedException | ExecutionException e) {
            // 处理任务执行中抛出的异常
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }
}

使用案例：

    private static void asyncHandleInputStream(Process proc) {
        ExecutorService asyncHandleInputStream = null;
        try {
            asyncHandleInputStream = new ThreadPoolExecutor(5, 10, 60, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
            asyncHandleInputStream.execute(() -> {
                String inputInfoLine;
                try (BufferedReader inputReader = new BufferedReader(
                    new InputStreamReader(proc.getInputStream(), StandardCharsets.UTF_8))) {
                    while ((inputInfoLine = inputReader.readLine()) != null) {
                    }
                } catch (Exception exp) {
                    LOGGER.error("handle InputStream error! {}", exp);
                }
            });
        } finally {
            if (asyncHandleInputStream != null) {
                asyncHandleInputStream.shutdown();
            }
        }
    }

java.util.concurrent.ExecutionException: java.lang.RuntimeException: Exception in task
	at java.util.concurrent.FutureTask.report(FutureTask.java:122)
	at java.util.concurrent.FutureTask.get(FutureTask.java:192)
	at ThreadPoolExceptionHandling.main(ThreadPoolExceptionHandling.java:16)
Caused by: java.lang.RuntimeException: Exception in task
	at ThreadPoolExceptionHandling.lambda$main$0(ThreadPoolExceptionHandling.java:10)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)

2.2 Executors

Executors 是一个线程池的工具类，提供了很多静态方法用于返回不同特点的线程池对象。

方法名称说明
public static ExecutorService newFixedThreadPool(int nThreads)创建固定线程数量的线程池，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程替代它
public static ExecutorService newSingleThreadExecutor()创建只有一个线程的线程池对象，如果该线程出现异常而结束，那么线程池会补充一个新线程
public static ExecutorService newCachedThreadPool()线程数量随着任务增加而增加，如果线程任务执行完毕且空闲了 60s 则会被回收掉。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)创建一个线程池，可以实现在给定的延迟后运行任务或者定期执行任务

newScheduledThreadPool 的基本使用示例：

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        System.out.printf("[%s] %s\n", Thread.currentThread().getName(),
                          LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
    }
}

package atreus.ink;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
        Runnable target = new MyRunnable();

        // 延迟1秒后执行target任务
        pool.schedule(target, 1, TimeUnit.SECONDS);

        // 延迟2秒后，每隔3秒执行一次target任务
        pool.scheduleAtFixedRate(target, 2, 3, TimeUnit.SECONDS);

        Thread.sleep(10 * 1000);
        pool.shutdown();
    }
}

[pool-1-thread-1] 2023-08-30 16:26:33
[pool-1-thread-2] 2023-08-30 16:26:34
[pool-1-thread-2] 2023-08-30 16:26:37
[pool-1-thread-2] 2023-08-30 16:26:40

3 线程状态

请添加图片描述
java.lang.Thread.State 中定义了六种线程状态，可以通过 getState() 方法获取当前线程的状态。

线程状态说明
NEW通过 new 关键字新建一个线程，但还未调用 start() 方法
RUNNABLE调用 start() 后等待调度（就绪）、正在运行
BLOCKED等待 synchronized 监视器锁时，陷入阻塞状态
WAITING等待其他线程执行特定的操作
TIMED_WAITING具有指定等待时间的等待状态
TERMINATED线程完成执行，变为终止状态

sleep() 和 yield() 的区别：

  • sleep() 会强制线程进入超时等待状态，时间到了之后才会转入就绪状态，是一种相对确定的暂停方式。而 yield() 方法会提示当前线程进入就绪状态，只是一种提示性的暂停，有可能被操作系统忽略。
  • 使用 sleep() 方法需要处理中断异常，而 yield() 不用。

BLOCKED 和 WAITING 均属于线程的阻塞等待状态，区别如下：

  • BLOCKED 是 synchronized 锁竞争失败后被动触发的状态，WAITING 是人为主动触发的状态。
  • BLOCKED 的唤醒时自动触发的，而 WAITING 状态是必须要通过特定的方法来主动唤醒，比如 Object.notify() 方法可以唤醒 Object.wait() 方法阻塞的线程，LockSupport.unpark() 可以唤醒 LockSupport.park() 方法阻塞的线程。

http://www.kler.cn/news/342757.html

相关文章：

  • 程序设计基础I-实验7 函数（编程题）
  • 邮件营销与社交媒体结合：打造全方位营销！
  • ESP8266模块（WIFI STM32）
  • 数据结构之二叉搜索树（key模型与key_value模型）
  • 使用OpenFeing远程调用时为方法添加请求头
  • 掌握 C# 设计模式：从基础到依赖注入
  • Django学习笔记七：用户认证与授权
  • C语言-常见文件操作函数详解(fgetc，fputc，fgets，fputs，fscanf，fprintf，fread，fwrite)
  • C++与Rust那些事之跳过析构函数
  • 杨中科 ASP.NETCORE 异步编程二
  • 【golang】gorm 使用map实现in 条件查询用法
  • 大数据面试-笔试SQL
  • OpenCV视频I/O(19）视频写入类VideoWriter之释放 VideoWriter 对象占用的资源函数release()的使用
  • vue 深拷贝和浅拷贝的区别
  • kafka和zookeeper单机部署
  • 制作一个流水灯，控制发光二极管由上至下再由下至上反复循环点亮显示，每次点亮一个发光二级管(Proteus 与Keil uVision联合仿真）
  • 控制台java原生工具打包jar文件
  • 51c自动驾驶~合集3
  • 模块化沙箱的功能特点
  • 数据结构之——树形结构