线程和线程池

Java
271
0
0
2024-02-26
《java并发编程实战》总结
上一篇
JVM笔记
下一篇
标签   Java多线程

线程

线程的几种状态

public enum State {
        //创建后尚未启动的线程处于这种状态。
        NEW,

        //Runable包括了操作系统现线程状态中的Runing和Ready,也就是处于次状态的线程有可能正在执行，也有可能正在等待着CPU为它分配执行时间。
        RUNNABLE,

        //线程被阻塞了，“阻塞状态”与“等待状态”的区别是：“阻塞状态”在等待的时候有一个排它锁，这个事件将在另外一个线程放弃这个锁的时候发生；“等待状态”则是等待一段世间，或者唤醒动作的发生。在程序等待进入同步区域的时候，线程进入这种状态。
        BLOCKED,

        //处于这种状态的线程不会被分配CPU执行时间，它们要等待被其他线程显式的唤醒。
        WAITING,

        //处于这种状态的县城也不会被分配CPU执行时间，不过无需等待被其它线程显式的唤醒，在一定时间之后它们会由系统自动唤醒，在一定时间之后它们会由系统自动唤醒。
        TIMED_WAITING,

        //已终止线程的线程状态，线程已经结束执行。
        TERMINATED;
    }

实现线程的几种方式(4种)

1）继承Thread类，实现run方法

public class MyThread extends Thread {

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            System.out.println(i);
        }
    }
    
}
        MyThread myThread = new MyThread();
        myThread.start();
        System.out.println("------------");

2）实现Runnable接口，实现run方法

public class MyRunable implements Runnable {
    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            System.out.println(Thread.currentThread().getName()+"----"+i);
        }
    }
}
        Thread thread = new Thread(new MyRunable());
        thread.start();
        System.out.println("------------");

3）实现Callable接口，实现call方法

和上面的方式相比，这种有返回结果

public class MyCallable implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {

        int sum = 0;

        for (int i = 0; i < 10; i++) {
            sum += i;
            System.out.println(Thread.currentThread().getName() + "----" + i);
        }
        return sum;
    }
}
        MyCallable myCallable = new MyCallable();
        FutureTask<Integer> futureTask = new FutureTask(myCallable);
        Thread thread = new Thread(futureTask);
        thread.start();
        System.out.println("---------------");
        Integer result = futureTask.get();//get方法是阻塞方法，只有当自定义的线程运行完才会得到结果
        System.out.println("result: " + result);

4）线程池创建线程

//创建线程池
        ExecutorService pool = Executors.newFixedThreadPool(5);
        //为线程池中的线程分配任务
        MyCallable myCallable = new MyCallable();
        Future<Integer> result = pool.submit(myCallable);
        //关闭线程池
        pool.shutdown();
        System.out.println(result.get());

线程池

线程池工作原理

1 先向核心线程 提交任务

2 如果核心线程满了 把任务放在队列中

3 如果队列也满了 ，那就扩招 非核心线程执行提交的任务，此任务不进队列

4 最大线程 和 任务队列都满了，就执行拒绝策略

线程池的核心参数

以下面为例

ExecutorService executorService = Executors.newFixedThreadPool(5);//一个池子有5个线程

跟一下newFixedThreadPool方法

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

继续跟

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

继续跟

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

此时ThreadPoolExecuter的7个参数就出来了

1）int corePoolSize

线程池中常驻的核心线程数

2）int maximumPoolSize

线程池中允许同时容纳执行的最大线程数，此值必须大于等于1

3）long keepAliveTime

多余的空闲线程的存活时间，当前池中线程数量超过corePoolSize时，当空闲时间达到keepAliveTime时，多余线程会被销毁直到只剩下corePoolSize

4）TimeUnit unit

keepAliveTime的单位

5）BlockingQueue<Runnable> workQueue

任务队列，被提交但尚未被执行的任务

6）ThreadFactory threadFactory

表示生成线程池中工作线程的线程工程，用于创建线程，一般默认即可

7）RejectedExecutionHandler handler

拒绝策略，表示当队列满了，并且工作线程大于等于线程池的最大线程数（maximumPoolSize ）时如何来拒绝请求执行的runable的策略

自定义线程池的原因

自定义线程池

代码
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 2L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());

其中

corePoolSize=2

maximumPoolSize=5

keepAliveTime=2

TimeUnit=秒

workQueue=new LinkedBlockingQueue<Runnable>(3) ，里面的构造方法传入3，否则默认为Integer.max(源码)

threadFactory=Executors.defaultThreadFactory() 一般就用Executors的默认的线程工厂

handler=new ThreadPoolExecutor.AbortPolicy() 这个是线程池默认的拒绝策略

为什么这么配？

corePoolSize=2 maximumPoolSize=5 这两个数怎来的？

如果是CPU密集型，maximumPoolSize=CPU核数+1

CPU核数怎么看：

System.out.println(Runtime.getRuntime().availableProcessors());

如果是IO密集型，maximumPoolSize=2*CPU (这个不太准备，自行百度)

参考：

什么是CPU密集型、IO密集型？

什么是CPU密集型、IO密集型？_Java技术栈，分享最主流的Java技术-CSDN博客_io密集型和cpu密集型

拒绝策略

  • 1） new ThreadPoolExecutor.AbortPolicy()（默认）

---->这种拒绝策略当达到maximumPoolSize+队列最大值后就会中断报异常

ExecutorService executorService = new ThreadPoolExecutor(2, 5, 2L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());

此定义的线程为最大线程数为5，阻塞队列为3，也就是最大运行同时提交8个线程，如果我同时提交8个线程，那没有问题，运行5个，3个在队列中，没有问题

如果我现在同时运行9个，那就会触发拒绝策略，测试代码如下

public class Start {

    public static void main(String[] args) {

        //自定义线程池，最大线程数为5，等待队列最大为3，最大运行同时提交最大线程数为5+3=8
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 2L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());


        try {
            for (int i = 0; i < 9; i++) {
                int finalI = i;
                executorService.execute(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(3);//睡3秒，才能达到并发目的
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                    }
                    System.out.println(Thread.currentThread().getName() + "\t" + finalI);

                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }
}

  • 2）new ThreadPoolExecutor.CallerRunsPolicy()

----> 该策略既不会抛抛弃任务，也不会抛出异常，而是将某些任务退到调用者，从而降低新任务的流量。

例如下面的测试代码，线程池最大运行8个（5个运行，3个等待），第九个就回退给main方法区运行

    public static void main(String[] args) {

        //自定义线程池，最大线程数为5，等待队列最大为3，最大运行同时提交最大线程数为5+3=8
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 2L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());


        try {
            for (int i = 0; i < 9; i++) {
                int finalI = i;
                executorService.execute(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(3);//睡3秒，才能达到并发目的
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                    }
                    System.out.println(Thread.currentThread().getName() + "\t" + finalI);

                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }
}

  • 3）new ThreadPoolExecutor.DiscardPolicy()

---->该策略默默的丢弃无法处理的任务，不予任何处理也不抛弃异常。如果运行任务丢失，这是最好的一种策略。

例如下面测试代码，丢弃了第9个任务

public class Start {

    public static void main(String[] args) {

        //自定义线程池，最大线程数为5，等待队列最大为3，最大运行同时提交最大线程数为5+3=8
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 2L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardPolicy());


        try {
            for (int i = 0; i < 9; i++) {
                int finalI = i;
                executorService.execute(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(3);//睡3秒，才能达到并发目的
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                    }
                    System.out.println(Thread.currentThread().getName() + "\t" + finalI);

                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }
}

  • 4）new ThreadPoolExecutor.DiscardOldestPolicy()

---->抛弃队列中等待最久的任务，然后把当前任务加入到队列中尝试再次提交当前任务（换句话说：长江后浪推前浪，把前浪拍死在沙滩上）

类似上面new ThreadPoolExecutor.DiscardPolicy()

《java并发编程实战》总结
上一篇
JVM笔记
下一篇