多线程&&线程池

迷南。 2022-12-29 00:26 181阅读 0赞

# 多线程 #

## 多线程的实现方式主要有三种 ##

*  **extends Thread**
 *  **implements Runnable**
 *  **implements Callable**

## 实现接口 VS 继承 Thread ##

实现接口会更好一些，因为：

*  Java 不支持多重继承，因此继承了 Thread 类就无法继承其它类，但是可以实现多个接口；
 *  类可能只要求可执行就行，继承整个 Thread 类开销过大。

*  ### **extends Thread** ###

继承 Thread 的代码

public class ExtendsThread extends Thread {
        /**
         * 线程名称
         */
        private String threadName;
     
        /**
         * 构造函数
         *
         * @param threadName 线程名称
         */
        public ExtendsThread(String threadName) {
            this.threadName = threadName;
        }
     
        /**
         * 重写run方式
         */
        @Override
        public void run() {
            System.out.println(threadName);
        }
    }

public class ThreadMain {
        public static void main(String[] args) {
            ExtendsThread extendsThread = new ExtendsThread("1");
            extendsThread.start();
            extendsThread = new ExtendsThread("2");
            extendsThread.start();
        }
    }

*  ### **implements Runnable** ###

实现**Runnable**接口

public class ThreadMain {
        public static void main(String[] args) {
            ThreadImplRunnable threadImplRunnable = new ThreadImplRunnable("Thread1");
            new Thread(threadImplRunnable).start();
            threadImplRunnable = new ThreadImplRunnable("Thread2");
            new Thread(threadImplRunnable).start();
            System.out.println("end");
        }
    }

public class ThreadImplRunnable implements Runnable {
        /**
         * 线程名称
         */
        private String threadName;
     
        /**
         * 构造函数
         *
         * @param threadName 线程名称
         */
        public ThreadImplRunnable(String threadName) {
            this.threadName = threadName;
        }
        /**
         * 重写run方法
         */
        @Override
        public void run() {
            System.out.println(threadName);
        }
    }

end
    Thread1
    Thread2

*  ### **implements Callable** ###

实现**Callable**接口主要是可以获取到返回值,futureTask.get() 获取返回的值

public class ThreadImplCallable<T> implements Callable<T> {
        /**
         * 线程名称
         */
        private String threadName;
     
        /**
         * 构造函数
         *
         * @param threadName 线程名称
         */
        public ThreadImplCallable(String threadName) {
            this.threadName = threadName;
        }
        /**
         * 重写call方法
         */
        @Override
        public T call() throws Exception {
            System.out.println(threadName);
            return (T)threadName;
        }
    }

public class ThreadMain {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            Callable<String> threadImplCallable = new ThreadImplCallable<>("Thread1");
            FutureTask<String> futureTask = new FutureTask<>(threadImplCallable);
            Thread thread = new Thread(futureTask);
            thread.start();
            String rValue = futureTask.get();
            System.out.println("Thread1 return value is " + rValue);
            threadImplCallable = new ThreadImplCallable<>("Thread2");
            futureTask = new FutureTask<>(threadImplCallable);
            thread = new Thread(futureTask);
            thread.start();
            rValue = futureTask.get();
            System.out.println("Thread2 return value is " + rValue);
     
            System.out.println("end");
        }
    }

Thread1
    Thread1 return value is Thread1
    Thread2
    Thread2 return value is Thread2
    end

# **线程池** #

java 线程池的顶级接口是 Executor, 只定义了一个方法 execute，ExecutorService继承接口Executor接口，并定义了其他的相关的接口，ExecutorService下面有一个继承的抽象类AbstractExecutorService和实现了定时调度的接口ScheduledExecutorService，而又有两个类继承抽象类，分别是ThreadPoolExecutor和ForkJoinPool，而ScheduledThreadPoolExecutor则实现接口实现ScheduledExecutorService接口和继承ThreadPoolExecutor类，基本的关系就是这样的，而且基本常用的封装方法都在Executors类中，只要 Executors.xxxx 就可以应用对应的封装的方法。

## Executor ##

Executor 管理多个异步任务的执行，而无需程序员显式地管理线程的生命周期。这里的异步是指多个任务的执行互不干扰，不需要进行同步操作。

## **线程池种类** ##

*  ### **newSingleThreadExecutor** ###

一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    public class ThreadPool {
        public static void main(String[] args) {
            ExecutorService pool = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 10; i++) {
                pool.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
                });
            }
        }
    }

pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....
    pool-1-thread-1    开始发车啦....

*  ### **newFixedThreadPool** ###

创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    public class ThreadPool {
        public static void main(String[] args) {
            ExecutorService pool = Executors.newFixedThreadPool(10);
            for (int i = 0; i < 10; i++) {
                pool.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
                });
            }
        }
    }

pool-1-thread-1    开始发车啦....
    pool-1-thread-4    开始发车啦....
    pool-1-thread-3    开始发车啦....
    pool-1-thread-2    开始发车啦....
    pool-1-thread-6    开始发车啦....
    pool-1-thread-7    开始发车啦....
    pool-1-thread-5    开始发车啦....
    pool-1-thread-8    开始发车啦....
    pool-1-thread-9    开始发车啦....
    pool-1-thread-10 开始发车啦....

*  ### **newCachedThreadPool** ###

创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，那么就会回收部分空闲的线程，当任务数增加时，此线程池又添加新线程来处理任务。

public static void main(String[] args) {
            ExecutorService pool = Executors.newCachedThreadPool();
            for (int i = 0; i < 10; i++) {
                pool.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
                });
            }
        }

pool-1-thread-1	开始发车啦....
    pool-1-thread-2	开始发车啦....
    pool-1-thread-2	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-1	开始发车啦....
    pool-1-thread-7	开始发车啦....
    pool-1-thread-5	开始发车啦....
    pool-1-thread-6	开始发车啦....
    pool-1-thread-4	开始发车啦....

*  ### **newScheduledThreadPool** ###

此线程池支持定时以及周期性执行任务的需求。

import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.TimeUnit;
    public class ThreadPool {
        public static void main(String[] args) {
            ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);
            for (int i = 0; i < 10; i++) {
                pool.schedule(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
                }, 10, TimeUnit.SECONDS);
            }
        }
    }

pool-1-thread-1	开始发车啦....
    pool-1-thread-2	开始发车啦....
    pool-1-thread-1	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-7	开始发车啦....
    pool-1-thread-8	开始发车啦....
    pool-1-thread-6	开始发车啦....
    pool-1-thread-5	开始发车啦....
    pool-1-thread-2	开始发车啦....
    pool-1-thread-4	开始发车啦....

上面演示的是延迟10秒执行任务,如果想要执行周期性的任务可以用下面的方式，每秒执行一次。

public static void main(String[] args) {
            ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);
            //pool.scheduleWithFixedDelay也可以
            pool.scheduleAtFixedRate(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
            }, 1, 1, TimeUnit.SECONDS);
    }

pool-1-thread-1	开始发车啦....
    pool-1-thread-1	开始发车啦....
    pool-1-thread-2	开始发车啦....
    pool-1-thread-1	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-2	开始发车啦....
    pool-1-thread-4	开始发车啦....
    pool-1-thread-1	开始发车啦....
    pool-1-thread-5	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-3	开始发车啦....
    pool-1-thread-9	开始发车啦....
    pool-1-thread-9	开始发车啦....

*  ### **newWorkStealingPool** ###

newWorkStealingPool是jdk1.8才有的，会根据所需的并行层次来动态创建和关闭线程，通过使用多个队列减少竞争，底层用的ForkJoinPool来实现的。ForkJoinPool的优势在于，可以充分利用多cpu，多核cpu的优势，把一个任务拆分成多个“小任务”，把多个“小任务”放到多个处理器核心上并行执行；当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。

newWorkStealingPool适合使用在很耗时的操作，但是newWorkStealingPool不是ThreadPoolExecutor的扩展，它是新的线程池类ForkJoinPool的扩展，但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），所以适合使用在很耗时的任务中

package org.jeecg.modules.exam.entity;
    
    import org.junit.Test;
    
    import java.util.concurrent.*;
    
    public class WorkStealingPoolTest {
        // 线程数
        private static final int threads = 10;
        // 用于计数线程是否执行完成
        CountDownLatch countDownLatch = new CountDownLatch(threads);
    
        /**
         * newFixedThreadPool execute
         *
         * @throws ExecutionException
         * @throws InterruptedException
         */
        @Test
        public void test1() throws ExecutionException, InterruptedException {
            System.out.println("---- start ----");
            ExecutorService executorService = Executors.newWorkStealingPool();
            for (int i = 0; i < threads; i++) {
                executorService.execute(() -> {
                    try {
                        System.out.println(Thread.currentThread().getName());
                    } catch (Exception e) {
                        System.out.println(e);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            countDownLatch.await();
            System.out.println("---- end ----");
        }
    
        /**
         * newFixedThreadPool submit submit
         */
        @Test
        public void test2() throws InterruptedException {
            System.out.println("---- start ----");
            ExecutorService executorService = Executors.newWorkStealingPool();
            for (int i = 0; i < threads; i++) {
    //            Callable 带返回值
                executorService.submit(new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            System.out.println(Thread.currentThread().getName());
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            countDownLatch.countDown();
                        }
                    }
                }));
            }
            countDownLatch.await();
            System.out.println("---- end ----");
        }
        /**
         * newFixedThreadPool submit Callable
         *
         * @throws ExecutionException
         * @throws InterruptedException
         */
        @Test
        public void test3() throws ExecutionException, InterruptedException {
            System.out.println("---- start ----");
            ExecutorService executorService = Executors.newWorkStealingPool();
            for (int i = 0; i < threads; i++) {
    //          Runnable 带返回值
                FutureTask<?> futureTask = new FutureTask<>(new Callable<String>() {
                    /**
                     * call
                     * @return currentThreadName
                     */
                    @Override
                    public String call() {
                        return Thread.currentThread().getName();
                    }
                });
                executorService.submit(new Thread(futureTask));
                System.out.println(futureTask.get());
            }
            System.out.println("---- end ----");
        }
    }