JUC的世界II

港控/mmm° 2022-10-27 05:13 124阅读 0赞

> 勤奋者废寝忘食，懒惰人总没有时间。  
> 你好，我是梦阳辰，期待与你相遇！

### 文章目录 ###

*  01.实现多线程的第三种方式（Callable)
 *  02.CountDownLatch线程计数器（辅助类）
 *  03.CyclicBarrier计数（人到齐了开会)（辅助类）
 *  04.Semaphore（计数信号量）（辅助类）
 *  05.ReadWriteLock读写锁
 *  06.BlockingQueue阻塞队列

# 01.实现多线程的第三种方式（Callable) #

第一种，基础Thread类。  
第二种，实现Runable接口。

**第三种：实现Callable接口**

callable接口与runnable接口的区别?

1）是否有返回值（泛型）

2）是否抛异常

3）落地方法不一样，一个是run，一个是call

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70]  
报错，因为Thread构造方法需要的是Runable接口，没有Callable接口。

解决方法：我没有那个女孩子的联系方式，但是我知道她朋友的联系方式，你说该怎么解决。

找到Runable的子接口`RunnableFuture <V>`并找到其实现类：`java.util.concurrent.FutureTask`，发现其构造方法：  
![在这里插入图片描述][20210203154822873.png]

FutureTask(Callable<V> callable) 
    创建一个 FutureTask ，它将在运行时执行给定的 Callable 。 
    
    FutureTask(Runnable runnable, V result) 
    创建一个 FutureTask ，将在运行时执行给定的 Runnable ，并安排 get将在成功完成后返回给定的结果。

来：

package Part3;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    class MyThread implements Callable<String>{ 
    
        @Override
        public String call() throws Exception { 
            System.out.println("you can very dance!");
            return "ok";
        }
    }
    public class CallableThread { 
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            FutureTask futureTask = new FutureTask(new MyThread());//future这个对象实现了Runable接口
    
            new Thread(futureTask,"t1").start();//多态
            System.out.println(futureTask.get());
        }
    }

![在这里插入图片描述][20210203155344333.png]  
可以获取线程执行的返回值。

一个fureTask对象，只会调用一次线程。

package Part3;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    class MyThread implements Callable<String>{ 
    
        @Override
        public String call() throws Exception { 
            System.out.println("you can very dance!");
            return "ok";
        }
    }
    public class CallableThread { 
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            FutureTask futureTask = new FutureTask(new MyThread());//future这个对象实现了Runable接口
    
            new Thread(futureTask,"t1").start();//多态
            new Thread(futureTask,"t2").start();//多态
            System.out.println(futureTask.get());
        }
    }

结果只有一个。

# 02.CountDownLatch线程计数器（辅助类） #

java.util.concurrent.CountDownLatch

允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助。

A CountDownLatch用给定的计数初始化。 await方法阻塞，直到由于countDown()方法的调用而导致当前计数达到零，之后所有等待线程被释放，并且任何后续的await 调用立即返回。 这是一个一次性的现象 - 计数无法重置。 如果您需要重置计数的版本，请考虑使用CyclicBarrier 。

CountDownLatch主要有两个方法，当一个或多个线程调用await方法时，这些线程会阻塞。其它线程调用countDown方法会将计数器减1(调用countDown方法的线程不会阻塞)，  
当计数器的值变为0时，因await方法阻塞的线程会被唤醒，继续执行。

CountDownLatch(int count) 
    构造一个以给定计数 CountDownLatch CountDownLatch。

例如:  
让其它线程做完后，主线程才离开。

package Part3;
    
    import java.util.concurrent.CountDownLatch;
    
    public class CountDownLatchDemo { 
        public static void main(String[] args) throws InterruptedException { 
    
            CountDownLatch countDownLatch = new CountDownLatch(6);
            //创建6个线程
            for(int i=1;i<=6;i++){ 
                int finalI = i;
                new Thread(()->{ 
                    System.out.println(finalI +"同学离开教室！");
                    countDownLatch.countDown();//减一
                },String.valueOf(i)).start();
            }
    
            countDownLatch.await();
            //班长离开锁门
            System.out.println("班长锁门离开教室！");
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 1]

# 03.CyclicBarrier计数（人到齐了开会)（辅助类） #

允许一组线程全部等待彼此达到共同屏障点的同步辅助。 循环阻塞在涉及固定大小的线程方的程序中很有用，这些线程必须偶尔等待彼此。 屏障被称为循环 ，因为它可以在等待的线程被释放之后重新使用。  
A CyclicBarrier支持一个可选的Runnable命令，每个屏障点运行一次，在派对中的最后一个线程到达之后，但在任何线程释放之前。 在任何一方继续进行之前，此屏障操作对更新共享状态很有用。

**等待线程的数量达到一定时：执行特定线程。**

构造方法:

CyclicBarrier(int parties) 
    创建一个新的 CyclicBarrier ，当给定数量的线程（线程）等待它时，它将跳闸，并且当屏障跳闸时不执行预定义的动作。  
    
    
    CyclicBarrier(int parties, Runnable barrierAction) 
    创建一个新的 CyclicBarrier ，当给定数量的线程（线程）等待时，它将跳闸，当屏障跳闸时执行给定的屏障动作，由最后一个进入屏障的线程执行。

package Part3;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class CyclicBarrierDemo { 
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{ 
                System.out.println("人到齐了，可以开会了！");
            });
    
            for(int i=1;i<=7;i++){ 
                int temp = i;
                new Thread(()->{ 
                    System.out.println(temp+"号员工到了！");
                    try { 
                        cyclicBarrier.await();//等一下其它员工
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) { 
                        e.printStackTrace();
                    }
                },String.valueOf(i)).start();
            }
        }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 2]

# 04.Semaphore（计数信号量）（辅助类） #

java.lang.Object 
    java.util.concurrent.Semaphore

一个计数信号量。 在概念上，信号量维持一组许可证。 如果有必要，每个acquire()都会阻塞，直到许可证可用，然后才能使用它。 每个release()添加许可证，潜在地释放阻塞获取方。 但是，没有使用实际的许可证对象; Semaphore只保留可用数量的计数，并相应地执行。

Semaphore(int permits) 
    创建一个 Semaphore与给定数量的许可证和非公平公平设置。
      
    Semaphore(int permits, boolean fair) 
    创建一个 Semaphore与给定数量的许可证和给定的公平设置。

**在信号量上我们定义两种操作:**

acquire(获取)当一个线程调用acquire操作时，它要么通过成功获取信号量（信号量减1），要么一直等下去，直到有线程释放信号量，或超时。

release（释放）实际上会将信号量的值加1，然后唤醒等待的线程。

信号量主要用于两个目的，一个是用于多个共享资源的互斥使用，另一个用于并发线程数的控制。

**例如:7辆车抢3个座位。**  
当只有一个车位就相当于加了synchronized锁

package Part3;
    
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    public class SemaphoreDemo { 
        public static void main(String[] args) { 
            Semaphore semaphore = new Semaphore(3);
    
            for(int i=1;i<=7;i++){ 
                int temp = i;
                new Thread(()->{ 
                    try { 
                        semaphore.acquire();//抢车位
                        System.out.println(temp+"抢到了车位！");
                        TimeUnit.SECONDS.sleep(3);//停了3秒钟
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        semaphore.release();//释放车位
                    }
                }).start();
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 3]

# 05.ReadWriteLock读写锁 #

接口：

java.util.concurrent.locks 
    Interface ReadWriteLock

实现类:

java.lang.Object 
    java.util.concurrent.locks.ReentrantReadWriteLock

A ReadWriteLock维护一对关联的locks ，一个用于只读操作，一个用于写入。 read lock可以由多个阅读器线程同时进行，只要没有作者。 write lock是独家的。

所有ReadWriteLock实现必须保证的存储器同步效应writeLock操作（如在指定Lock接口）也保持相对于所述相关联的readLock 。 也就是说，一个线程成功获取读锁定将会看到在之前发布的写锁定所做的所有更新。

**读操作可以多个线程操作，多个线程同时读一个资源类没有问题，所以为了满足并发量，读取资源应该可以同时进行，但是如果有一个线程想去写共享资源，就不应该再有其它线程可以对该资源进行读写**。

例如:

package Part4;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.locks.ReadWriteLock;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    class MyCache{ 
        private volatile Map<String,Object> map = new HashMap<>();
        private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        public void put(String key,Object value){ 
    
            readWriteLock.writeLock().lock();
            try{ 
                System.out.println(Thread.currentThread().getName()+"\t 写入数据！");
                map.put(key,value);
                System.out.println(Thread.currentThread().getName()+"\t 写入成功！");
            }finally { 
                readWriteLock.writeLock().unlock();
            }
    
        }
    
        public void get(String key){ 
    
            readWriteLock.readLock().lock();
            try { 
                System.out.println(Thread.currentThread().getName()+"\t 读取数据！");
                map.get(key);
                System.out.println(Thread.currentThread().getName()+"\t 读取成功！");
            }finally { 
                readWriteLock.readLock().unlock();
            }
    
        }
    
    }
    
    public class ReadWriterLockDemo { 
        public static void main(String[] args) { 
    
            MyCache myCache = new MyCache();
    
            //写入线程
            for(int i = 1;i<=6;i++){ 
                String temp = String.valueOf(i);
                new Thread(()->{ 
                    myCache.put(temp,temp);
                },String.valueOf(i)).start();
            }
    
            //读取线程
            for(int i = 1;i<=6;i++){ 
                String temp = String.valueOf(i);
                new Thread(()->{ 
                    myCache.get(temp);
                },String.valueOf(i)).start();
            }
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 4]

# 06.BlockingQueue阻塞队列 #

java.util.concurrent 
    Interface BlockingDeque<E>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 5]  
试图从空的队列中获取元素的线程将会被阻塞，直到其他线程往空的队列插入新的元素。

试图向已满的队列中添加新元素的线程将会被阻塞，直到其他线程从队列中移除一个或多个元素或者完全清空，使队列变得空闲起来并后续新增。

在多线程领域:所谓阻塞，在某些情况下会挂起线程〈即阻塞），一旦条件满足，被挂起的线程又会自动被唤起

**为什么需要BlockingQueue**

好处是我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切BlockingQueue都给你一手包办了

在concurrent包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。

子接口:

java.util.concurrent 
    Interface BlockingQueue<E>

实现类：  
![在这里插入图片描述][20210205163424993.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 6]

ArrayBlockingQueue:由数组结构组成的有界阻塞队列。
    LinkedBlockingQueue:由链表结构组成的有界（但大小默认值为integer.MAX_VALUE)阻塞队列。
    PriorityBlockingQueue:支持优先级排序的无界阻塞队列。
    DelayQueue:使用优先级队列实现的延迟无界阻塞队列。
    SynchronousQueue:不存储元素的阻塞队列，也即单个元素的队列。
    LinkedTransferQueue:由链表组成的无界阻塞队列。
    LinkedBlocking Deque:由链表组成的双向阻塞队列。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 7]

package Part4;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    public class BlockingQueueDemo { 
        public static void main(String[] args) { 
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
    
            blockingQueue.add("1");
            blockingQueue.add("1");
            blockingQueue.add("1");
    
            System.out.println(blockingQueue.remove());
        }
    }

例二:

package Part4;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    public class BlockingQueueDemo { 
        public static void main(String[] args) throws InterruptedException { 
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
    
            System.out.println(blockingQueue.offer("1",3L, TimeUnit.SECONDS));
            System.out.println(blockingQueue.offer("1",3L, TimeUnit.SECONDS));
            System.out.println(blockingQueue.offer("1",3L, TimeUnit.SECONDS));
            System.out.println(blockingQueue.offer("1",3L, TimeUnit.SECONDS));
    
    
            System.out.println(blockingQueue.poll(2,TimeUnit.SECONDS));
        }
    }

> Time waits for no one. Treasure every moment you have.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 8]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 9]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70]: /images/20221024/d2486414bb8c45aa88f900966f7b840d.png
[20210203154822873.png]: /images/20221024/31892f17fe824c899f2b5bfd63ff4921.png
[20210203155344333.png]: /images/20221024/7c6cc173e0f74ea189530723d21d320d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 1]: /images/20221024/499beda4ce4f4dcfaf7e39bd12241754.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 2]: /images/20221024/03854d3d6d414674905902a169319585.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 3]: /images/20221024/f1e6474a08f04730872aec8bdfabfb49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 4]: /images/20221024/f676693d9c4f43b6aee9569211283c38.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 5]: /images/20221024/92710ebac9a440978e2a05a000dc7ca4.png
[20210205163424993.png]: /images/20221024/1d5d622275cb4df2b4827656e07de366.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 6]: /images/20221024/1ad443a5582c4e6196dcce93cae6e416.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 7]: /images/20221024/8db8e7696b34469b9f64a3c576814f69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 8]: /images/20221024/317b63dddcad4f2fb33436a87179c4da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWh1YW5f_size_16_color_FFFFFF_t_70 9]: /images/20221024/091851485bd644729baa5d8b223229db.png