读写锁+阻塞队列+线程池

1. 读写锁

// 独占锁（写锁）：一次只能被一个线程占有
// 共享锁（读锁）：可以被多个线程同时占有
// 运行结果分析：写锁保证操作的原子性，读锁不能保证操作的原子性
// 注意：lock锁一定要配对，不然可能会导致死锁
// 个人理解：写操作即为修改数据，会改变原有的数据，因此必须保证其写操作的完整性，
// 不能被中途打断，所以不能同时被多个线程调用
// 读操作即为查询数据，原有数据不会改变，因此可被多个线程同时调用

public class ReadWriteLockDemo1 {
    public static void main(String[] args) {
        MyCacheLock myCacheLock = new MyCacheLock();

        for (int i = 1; i <= 5; i++) {
            int temp = i;
            new Thread(() -> {
                myCacheLock.write(temp, "_" + temp + "_");
            }).start();
        }

        for (int i = 1; i <= 5; i++) {
            int temp = i;
            new Thread(() -> {
                myCacheLock.read(temp);
            }).start();
        }
    }
}

class MyCacheLock {
    private volatile Map<Integer, Object> map = new HashMap<>();
    // 创建读写锁
    ReadWriteLock lock = new ReentrantReadWriteLock();

    // 写
    public void write(Integer key, Object value) {
        // 写锁加锁
        lock.writeLock().lock();
        try {
            // 业务编码
            System.out.println(Thread.currentThread().getName() + "写入" + key);
            map.put(key, value);
            System.out.println(Thread.currentThread().getName() + "写入成功");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 释放写锁
            lock.writeLock().unlock();
        }
    }

    // 读
    public void read(Integer key) {
        // 读锁加锁
        lock.readLock().lock();
        try {
            // 业务编码
            System.out.println(Thread.currentThread().getName() + "读取" + key);
            Object value = map.get(key);
            System.out.println(Thread.currentThread().getName() + "读取结果" + value);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 释放读锁
            lock.readLock().unlock();
        }
    }
}

2. 阻塞队列

　　

 　　阻塞：当队列满时，还往里存，此时就会阻塞；　当队列为空时，还往外取，此时也会阻塞。

　　我们为什么要使用阻塞队列？

　　　　在多线程下，线程之间通信，我们需要关心的是线程的 —唤醒！

　　　　例如：在消息投递过程中，消息发送方和接受方 可通过消息中间件MQ 来异步解决阻塞问题。

// put(E e)  一直等待
// offer(E e, long timeout, TimeUnit unit) 超时等待，队列还是满即返回false
// offter(E e) 返回false，不报异常
// add(E e) 执行这行报异常 java.lang.IllegalStateException: Queue full
public class BlockingDemo {
    public static void main(String[] args) throws InterruptedException {
        // 参数：队列初始容量  设置为3
        ArrayBlockingQueue queue = new ArrayBlockingQueue<>(3);

        // 添加一个元素
        queue.put(1);
        // 取两次
        System.out.println(queue.take());
        System.out.println(queue.take()); // 执行到此，一直等待


        // put(E e)添加元素
        /*queue.put(1);
        queue.put(2);
        queue.put(3);
        queue.put(4); // 执行到此，一直等待

        System.out.println(queue.poll()); // 1
        System.out.println(queue.poll()); // 2
        System.out.println(queue.poll()); // 3
        System.out.println(queue.poll()); // null
        */

        // -----------------------------------------------------------
        // offer(E e, long timeout, TimeUnit unit)添加元素
        /*System.out.println(queue.offer(1, 3L, TimeUnit.SECONDS)); // true
        System.out.println(queue.offer(2, 3L, TimeUnit.SECONDS)); // true
        System.out.println(queue.offer(3, 3L, TimeUnit.SECONDS)); // true
        System.out.println(queue.offer(4, 3L, TimeUnit.SECONDS)); // 等待3秒，false (超时等待，队列还是满即返回false)
        */

        // ----------------------------------------------------------------
        // offter(E e)添加元素
        /*System.out.println(queue.offer(1)); // true
        System.out.println(queue.offer(2)); // true
        System.out.println(queue.offer(3)); // true
        System.out.println(queue.offer(4)); // 返回false，不报异常

        System.out.println(queue.poll()); // 1
        System.out.println(queue.poll()); // 2
        System.out.println(queue.poll()); // 3
        System.out.println(queue.poll()); // null
        */

        // -------------------------------------------------
        // add(E e)添加元素
        /*queue.add(1);
        queue.add(2);
        queue.add(3);
        // 添加第4个元素
        queue.add(4);  //执行这行报异常 java.lang.IllegalStateException: Queue full

        System.out.println(queue.remove());
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        */
    }
}

 　　特殊的阻塞队列——同步队列

// 同步队列
public class SynchronousQueueDemo {
    public static void main(String[] args) {
        // 特殊的阻塞队列  容量固定为1
        SynchronousQueue<Object> queue = new SynchronousQueue<>();

        // 添加元素
        new Thread(() -> {
            try {
                // 延时 以便看效果
                TimeUnit.SECONDS.sleep(3);
                queue.put(1);
                TimeUnit.SECONDS.sleep(3);
                queue.put(2);
                TimeUnit.SECONDS.sleep(3);
                queue.put(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"A").start();

        // 取出元素
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(queue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(queue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"B").start();
    }
}

3. 线程池

　　池化技术：程序运行会占用系统资源，提高程序的使用率，降低我们一个性能消耗

　　　　线程池，连接池，内存池，对象池......

　　为什么要使用线程池？　　线程复用

　　线程池：

　　　　三大方法：

　　　　　　1.  Executors.newSingleThreadExecutor()  单例，只有一个线程

　　　　　　2.  Executors.newFixedThreadPool(5) 固定线程数  参数：线程数量

　　　　　　3.  Executors.newCachedThreadPool()  线程数可变，可伸缩

　　　　七大参数：

　　　　　　1.  int corePoolSize  // 核心池线程数

　　　　　　2.  int maximumPoolSize  // 最大线程数

　　　　　　3.  long  keepAliveTime   // 超时等待时间

　　　　　　4.  TimeUnit unit  // keepAliveTime参数的时间单位

　　　　　　5.  BlockingQueue< Runnable > workQueue  // 阻塞队列

　　　　　　6.  ThreadFactory threadFactory  // 线程工厂

　　　　　　7.  RejectedExecutionHandler handler  // 拒绝策略 （见下面）

　　　　四种拒绝策略：

　　　　　　1.  AbortPolicy()  // 直接报异常，丢弃任务

　　　　　　2.  DiscardOldestPolicy()  // 线程池关闭之前，尝试去获取

　　　　　　3.  DiscardPolicy() //  直接返回

　　　　　　4.  CallerRunsPolicy()  // 从哪来回哪去 

public class ThreadPoolExecutorDemo1 {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, // 核心池线程数
                5, // 最大线程数
                5L,   // 超时等待时间
                TimeUnit.SECONDS,  // keepAliveTime参数的时间单位
                new LinkedBlockingDeque<>(3), // 阻塞队列
                Executors.defaultThreadFactory(), // 线程工厂
                new ThreadPoolExecutor.AbortPolicy()  // 拒绝策略
        );

        /*
        * new ThreadPoolExecutor.CallerRunsPolicy()  // 从哪来回哪去
         * new ThreadPoolExecutor.DiscardOldestPolicy() // 线程池关闭之前，尝试去获取
        * new ThreadPoolExecutor.DiscardPolicy() //  直接返回
        * new ThreadPoolExecutor.AbortPolicy()// 直接报异常，丢弃任务
        */

        try {
            for (int i = 0; i < 15; i++) {
                executor.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "__ok");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭线程池
            executor.shutdown();
        }
    }
}

　　最大线程池 该如何设置？

　　　　CPU密集型： Runtime.getRuntime().availableProcessors() 获取当前CPU核数

　　　　IO密集型：  假设当前有50个线程都是进程经常操作大IO资源的，比较耗时！ 我们则需要保证设置的线程池数量大于 50