（2）JUC并发实用工具的学习与应用

傷城~ 2023-07-11 08:42 10阅读 0赞

*   1，线程按序交替
 *   2，ReadWriteLock读写锁
 *  3，线程八锁
 *   4，线程池
 *   5，线程调度
 *   6，ForkJoinPool分支/合并框架 工作窃取

#### 线程按序交替 ####

*  **例题**：编写一个程序，开启 3 个线程，这三个线程的 ID 分别为A、B、C，每个线程将自己的 ID 在屏幕上打印 10 遍，要求输出的结果必须按顺序显示。如：ABCABCABC…… 依次递归，代码如下：

public static void main(String[] args) { 
    
            AlternateDemo ad = new AlternateDemo();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    for (int i = 1; i < 11; i++) { 
                        ad.loopA(i);
                    }
                }
            }, "A").start();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    for (int i = 1; i < 11; i++) { 
                        ad.loopB(i);
                    }
                }
            }, "B").start();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    for (int i = 1; i < 11; i++) { 
                        ad.loopC(i);
                    }
                }
            }, "C").start();
        }
    }
    
    class AlternateDemo { 
    
        private int number = 1;//当前正在执行线程的标记
        private Lock lock = new ReentrantLock();
        private Condition condition1 = lock.newCondition();
        private Condition condition2 = lock.newCondition();
        private Condition condition3 = lock.newCondition();
    
        /** * @param totalLoop 循环第几轮 */
        public void loopA(int totalLoop) { 
            lock.lock();
            try { 
                //1，判断
                if (number != 1) { 
                    condition1.await();
                }
                //2，打印
                for (int i = 1; i < 2; i++) { 
                    System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
                }
                //3，唤醒
                number = 2;
                condition2.signal();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                lock.unlock();
            }
        }
    
        public void loopB(int totalLoop) { 
            lock.lock();
            try { 
                //1，判断
                if (number != 2) { 
                    condition2.await();
                }
                //2，打印
                for (int i = 1; i < 2; i++) { 
                    System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
                }
                //3，唤醒
                number = 3;
                condition3.signal();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                lock.unlock();
            }
        }
    
        public void loopC(int totalLoop) { 
            lock.lock();
            try { 
                //1，判断
                if (number != 3) { 
                    condition3.await();
                }
                //2，打印
                for (int i = 1; i < 2; i++) { 
                    System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
                }
                //3，唤醒
                number = 1;
                condition1.signal();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                lock.unlock();
            }
        }
    }

#### ReadWriteLock读写锁 ####

*  ReadWriteLock 维护了一对相关的锁，一个用于只读操作，另一个用于写入操作。只要没有 writer，读取锁可以由多个 reader 线程同时保持。写入锁是独占的。

//写写要互斥，读读不需要
        public static void main(String[] args) { 
            ReadWriteLockDemo rw = new ReadWriteLockDemo();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    rw.set((int) (Math.random()*101));
                }
            },"Write").start();
    
            for (int i = 0; i < 100; i++) { 
                new Thread(new Runnable() { 
                    @Override
                    public void run() { 
                        rw.get();
                    }
                },"Read").start();
            }
        }
    }
    
    class ReadWriteLockDemo { 
    
        private int number = 0;
        private ReadWriteLock lock = new ReentrantReadWriteLock();
    
        //读
        public void get() { 
            lock.readLock().lock();//上锁
            try { 
                System.out.println(Thread.currentThread().getName() + ":" + number);
            } finally { 
                lock.readLock().unlock();
            }
        }
    
        //写
        public void set(int number) { 
            lock.writeLock().lock();
            try { 
                System.out.println(Thread.currentThread().getName());
                this.number = number;
            } finally { 
                lock.writeLock().unlock();
            }
        }
    }

#### 线程八锁 ####

*  1，两个普通同步方法，两个线程，标准打印，打印 //one two
 *  2，新增Thread.sleep()给getOne()，打印 //one two
 *  3，新增普通方法getThree，打印 //three one two
 *  4，两个普通的同步方法，两个Number对象，打印 //two one
 *  5，修改getOne()为静态同步方法，打印 //two one
 *  6，修改两个方法均为静态同步方法，一个Number对象，打印 //one two
 *  7，一个静态同步方法，一个非静态同步方法，两个Number对象，打印//two one
 *  8，两个静态同步方法，两个Number对象，打印//one two

**线程八锁的关键**：

1，非静态方法的锁默认为this，静态方法的锁为对呀的Class实例  
2，某一时刻，只能有一个线程持有锁，无论几个方法。

代码如下：

public class JucTest { 
    
        public static void main(String[] args) { 
            Number number = new Number();
            Number number1 = new Number();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    number.getOne();
                }
            }).start();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    number1.getTwo();
                }
            }).start();
    // new Thread(new Runnable() { 
    // @Override
    // public void run() { 
    // number.getThree();
    // }
    // }).start();
        }
    }
    
    class Number{ 
        public static synchronized void getOne(){ 
            try { 
                Thread.sleep(3000);
            } catch (InterruptedException e) { 
            }
            System.out.println("one");
        }
        public synchronized void getTwo(){ 
            System.out.println("two");
        }
    
        public void getThree(){ 
            System.out.println("three");
        }
    }

#### 线程池 ####

*  **概念**：提供了一个线程队列，队列中保存着所有等待状态的线程，避免了创建与销毁额外开销，提高了响应的速度。
 *  **线程池的体系结构**：  
          java.util.concurrent.Executor：负责线程的使用与调度的根接口  
               ExecutorService子接口：线程池的主要接口  
                    ThreadPoolExecutor： 线程池的实现类  
                          ScheduledExceutorService子接口：负责线程的调度  
                                  ScheduledThreadPoolExecutor：继承ThreadPoolExector，实现ScheduledExecutorService

**工具类：Executors**

ExecutorService newFixedThreadPool()：创建固定大小的线程池

ExecutorService newCachedThreadPool()：缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量

ExecutorService newSingleThreadExecutor()：创建单个线程池。线程池中只有一个线程

ScheduledExecutorService newScheduledThreadPool()：创建固定大小的线程，可以延迟或定时的执行任务

线程池的代码使用如下：

public static void main(String[] args) throws Exception { 
            //1，创建线程池
            ExecutorService pool = Executors.newFixedThreadPool(5);
    
            Future<Integer> future = pool.submit(new Callable<Integer>() { 
    
                @Override
                public Integer call() throws Exception { 
                    int sum = 0;
                    for (int i = 0; i <= 100; i++) { 
                        sum += i;
                    }
                    return sum;
                }
            });
            System.out.println(future.get());
    /* ThreadPoolDemo td = new ThreadPoolDemo(); //2，为线程池中的线程分配任务 for (int i = 0; i < 10; i++) { pool.submit(td); }*/
            //3，关闭线程池
            pool.shutdown();
        }
    }
    
    class ThreadPoolDemo implements Runnable{ 
        private  int i =0;
    
        @Override
        public void run() { 
            while (i<=100){ 
                System.out.println(Thread.currentThread().getName()+":"+i++);
            }
        }
    }

#### 线程调度 ####

*  一个 ExecutorService，可安排在给定的延迟后运行或定期执行的命令。代码如下：

public static void main(String[] args) throws Exception { 
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);
        for (int i = 0; i < 5; i++) { 
            Future<Integer> future = pool.schedule(new Callable<Integer>() { 
                @Override
                public Integer call() throws Exception { 
                    int num = new Random().nextInt(100);
                    System.out.println(Thread.currentThread().getName() + ":" + num);
                    return num;
                }
            }, 2, TimeUnit.SECONDS);
            System.out.println(future.get());
        }
        pool.shutdown();
    }

#### ForkJoinPool分支/合并框架 工作窃取 ####

*  **Fork/Join 框架**：就是在必要的情况下，将一个大任务，进行拆分(fork)成若干个小任务（拆到不可再拆时），再将一个个的小任务运算的结果进行 join 汇总  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTMwMDA0_size_16_color_FFFFFF_t_70]
 *  采用 “工作窃取”模式（work-stealing）：当执行新的任务时它可以将其拆分分成更小的任务执行，并将小任务加到线程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队  
    列中。

public static void main(String[] args) { 
            Instant start = Instant.now();
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<Long> task = new ForkJoinSumCalculate(0L, 1000000000L);
            Long sum = pool.invoke(task);
            System.out.println(sum);
            Instant end = Instant.now();
            System.out.println("消耗时间为" + Duration.between(start, end).toMillis());//10829
        }
    
        //for循环测试
        @Test
        public void testFor() { 
            Instant start = Instant.now();
            Long sum = 0L;
            for (Long i = 0L; i <= 1000000000L; i++) { 
                sum += 1;
            }
            System.out.println(sum);
            Instant end = Instant.now();
            System.out.println("消耗时间为" + Duration.between(start, end).toMillis());//11471
        }
    
        //java8测试
        @Test
        public void testJava8() { 
            Instant start = Instant.now();
            Long sum = LongStream.rangeClosed(0L, 1000000000L).parallel().reduce(0L, Long::sum);
            System.out.println(sum);
            Instant end = Instant.now();
            System.out.println("消耗时间为" + Duration.between(start, end).toMillis());//756
        }
    }
    
    class ForkJoinSumCalculate extends RecursiveTask<Long> { 
    
        private Long start;
        private Long end;
        private static final Long THRSHOLD = 10000L;
    
        public ForkJoinSumCalculate(Long start, Long end) { 
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Long compute() { 
            Long length = end - start;
            if (length <= THRSHOLD) { 
                Long sum = 0L;
                for (Long i = start; i <= end; i++) { 
                    sum += i;
                }
                return sum;
            } else { 
                Long middle = (start + end) / 2;
                ForkJoinSumCalculate left = new ForkJoinSumCalculate(start, middle);
                left.fork();//进行拆分，同时压入线程队列
    
                ForkJoinSumCalculate right = new ForkJoinSumCalculate(middle + 1, end);
                right.fork();
    
                return left.join() + right.join();
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTMwMDA0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200304205524533.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTMwMDA0,size_16,color_FFFFFF,t_70