JUC并发编程（一）多线程使用和线程池

HcodeBlogger

发布于 2020-07-14 10:54:53

6700

发布于 2020-07-14 10:54:53

线程和进程

一个进程往往可以包含多个线程，至少包含一个！ Java默认有几个线程？ 2 个： mian、GC 对于Java而言：Thread、Runnable、Callable 三种实现线程的方式。 JAVA不能开启线程，是调用本地方法，查看start方法可以看到底层是C++来开启线程的

并发和并行

并发（多线程操作同一个资源）：CPU 一核，模拟出来多条线程，天下武功，唯快不破，快速交替。
并行（同时多条线程同时执行）：CPU 多核，多个线程可以同时执行；线程池。

并发编程的本质：充分利用CPU的资源

线程的6种状态

COPYpublic enum State {    
    // 新生    
    NEW,
    // 运行    
  RUNNABLE,
    // 阻塞    
    BLOCKED,        
    // 等待，死死地等    
    WAITING,
    // 超时等待    
    TIMED_WAITING,
    // 终止    
    TERMINATED;
}

wait和sleep的区别

来自不同的类 wait => Object sleep => Thread
关于锁的释放 wait 会释放锁，sleep 睡觉了，抱着锁睡觉，不会释放！
范围不同：wait必须在同步代码块，sleep可以任意地方。

synchronized和lock

synchronized方式上锁

COPY/**
 * 真正的多线程开发，公司中的开发，降低耦合性
 * 线程就是一个单独的资源类，没有任何附属的操作！
 * 1、 属性、方法
 */
public class SaleTicketDemo01 {
    public static void main(String[] args) {
        // 并发：多线程操作同一个资源类, 把资源类丢入线程
        Ticket ticket = new Ticket();

        // @FunctionalInterface 函数式接口，jdk1.8  lambda表达式 (参数)->{ 代码 }
        new Thread(()->{
            for (int i = ; i <  ; i++) {
                ticket.sale();
            }
        },"A").start();

        new Thread(()->{
            for (int i = ; i <  ; i++) {
                ticket.sale();
            }
        },"B").start();

        new Thread(()->{
            for (int i = ; i <  ; i++) {
                ticket.sale();
            }
        },"C").start();


    }
}

// 资源类 OOP
class Ticket {
    // 属性、方法
    private int number = ;

    // 卖票的方式
    // synchronized 本质: 队列，锁
    public synchronized void sale(){
        if (number>){
            System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余："+number);
        }
    }

}

lock方式上锁

COPYpublic class SaleTicketDemo02  {
    public static void main(String[] args) {

        // 并发：多线程操作同一个资源类, 把资源类丢入线程
        Ticket2 ticket = new Ticket2();


        // @FunctionalInterface 函数式接口，jdk1.8  lambda表达式 (参数)->{ 代码 }
        new Thread(()->{for (int i = ; i <  ; i++) ticket.sale();},"A").start();
        new Thread(()->{for (int i = ; i <  ; i++) ticket.sale();},"B").start();
        new Thread(()->{for (int i = ; i <  ; i++) ticket.sale();},"C").start();


    }
}

// Lock三部曲
// 1、 new ReentrantLock();
// 2、 lock.lock(); // 加锁
// 3、 finally=>  lock.unlock(); // 解锁
class Ticket2 {
    // 属性、方法
    private int number = ;

    Lock lock = new ReentrantLock();

    public void sale(){

        lock.lock(); // 加锁

        try {
           // 业务代码

            if (number>){
                System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余："+number);
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock(); // 解锁
        }
    }

}

两者的区别

Synchronized 内置的Java关键字， Lock 是一个Java类
Synchronized 无法判断获取锁的状态，Lock 可以判断是否获取到了锁
Synchronized 会自动释放锁，lock 必须要手动释放锁！如果不释放锁，死锁
Synchronized 线程 1（获得锁，阻塞）、线程2（等待，傻傻的等）；Lock锁就不一定会等待下去；
Synchronized 可重入锁，不可以中断的，非公平；Lock ，可重入锁，可以判断锁，非公平（可以自己设置）；
Synchronized 适合锁少量的代码同步问题，Lock 适合锁大量的同步代码

生产者消费者模式两种实现

await 和 notify的实现方式

COPYpublic class B  {
    public static void main(String[] args) {
        Data2 data = new Data2();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"A").start();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"B").start();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"C").start();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"D").start();

    }
}

// 判断等待，业务，通知
class Data2{ // 数字 资源类

    private int number = ;

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    //condition.await(); // 等待
    //condition.signalAll(); // 唤醒全部
    //+1
    public void increment() throws InterruptedException {
        lock.lock();
        try {
            // 业务代码
            while (number!=){  //0
                // 等待
                condition.await();
            }
            number++;
            System.out.println(Thread.currentThread().getName()+"=>"+number);
            // 通知其他线程，我+1完毕了
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }

    //-1
    public synchronized void decrement() throws InterruptedException {
        lock.lock();
        try {
            while (number==){ // 1
                // 等待
                condition.await();
            }
            number--;
            System.out.println(Thread.currentThread().getName()+"=>"+number);
            // 通知其他线程，我-1完毕了
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

}

用JUC实现

COPYpublic class C {

    public static void main(String[] args) {
        Data3 data = new Data3();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                data.printA();
            }
        },"A").start();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                data.printB();
            }
        },"B").start();

        new Thread(()->{
            for (int i = ; i < ; i++) {
                data.printC();
            }
        },"C").start();
    }

}

class Data3{ // 资源类 Lock

    //几个线程就用几个Condition 精准的通知和唤醒线程
     private Lock lock = new ReentrantLock();
    private Condition condition1 = lock.newCondition();
    private Condition condition2 = lock.newCondition();
    private Condition condition3 = lock.newCondition();
    private int number = ; // 1A  2B  3C

    public void printA(){
        lock.lock();
        try {
            // 业务，判断-> 执行-> 通知
            while (number!=){
                // 等待
                condition1.await();
            }
            System.out.println(Thread.currentThread().getName()+"=>AAAAAAA");
            // 唤醒，唤醒指定的人，B
            number = ;
            condition2.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void printB(){
        lock.lock();
        try {
            // 业务，判断-> 执行-> 通知
            while (number!=){
                condition2.await();
            }
            System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
            // 唤醒，唤醒指定的人，c
            number = ;
            condition3.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public void printC(){
        lock.lock();
        try {
            // 业务，判断-> 执行-> 通知
            // 业务，判断-> 执行-> 通知
            while (number!=){
                condition3.await();
            }
            System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
            // 唤醒，唤醒指定的人，c
            number = ;
            condition1.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

}

使用JUC的集合

原生的List，Map，Set的实现类方法在多线程会出现不同步的问题，所以有以下解决的方案：

List

List list = new Vector<>(); 其实Vector底层也就是在add方法上加了synchronized。

List list = Collections.synchronizedList(new ArrayList<> ());
List list = new CopyOnWriteArrayList<>()；（推荐使用这个，效率高）：底层使用了可重入锁Lock，效率更高。

Set

Set set = Collections.synchronizedSet(new HashSet<>()); 使用这个锁的就是整个对象。

Set set = new CopyOnWriteArraySet<>();（推荐使用这个，效率高）：底层使用了可重入锁Lock，效率更高。

Map

Map map= Collections.synchronizedMap(new HashMap<>());
Map map = new ConcurrentHashMap<>();(推荐使用这个~)

使用Callable

执行相同的任务有存在缓存，效率高。
结果可能需要等待，会阻塞！

COPYpublic class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // new Thread(new Runnable()).start();
        // new Thread(new FutureTask<V>()).start();
        // new Thread(new FutureTask<V>( Callable )).start();
        new Thread().start(); // 怎么启动Callable

        MyThread thread = new MyThread();
        FutureTask futureTask = new FutureTask(thread); // 适配类

        new Thread(futureTask,"A").start();
        new Thread(futureTask,"B").start(); // 结果会被缓存，效率高

        Integer o = (Integer) futureTask.get(); //这个get 方法可能会产生阻塞！把他放到最后
        // 或者使用异步通信来处理！
        System.out.println(o);

    }
}

class MyThread implements Callable<Integer> {

    @Override
    public Integer call() {
        System.out.println("call()"); // 会打印几个call
        // 耗时的操作
        return ;
    }

}

GUC种的三大辅助类

CountDownLatch(减法计数器)

countDownLatch.countDown(); // 数量-1
countDownLatch.await(); // 等待计数器归零，然后再向下执行
每次有线程调用 countDown() 数量-1，假设计数器变为0，countDownLatch.await()就会被唤醒，继续执行

COPY// 计数器
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        // 总数是6，必须要执行任务的时候，再使用！
        CountDownLatch countDownLatch = new CountDownLatch();

        for (int i = ; i <= ; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+" Go out");
                countDownLatch.countDown(); // 数量-1
            },String.valueOf(i)).start();
        }

        countDownLatch.await(); // 等待计数器归零，然后再向下执行

        System.out.println("Close Door");

    }
}

CyclicBarrier (加法计数器)

只有等待线程数达到特定值才会结束。

COPYpublic class CyclicBarrierDemo {
    public static void main(String[] args) {
        /**
         * 集齐7颗龙珠召唤神龙
         */
        // 召唤龙珠的线程
        CyclicBarrier cyclicBarrier = new CyclicBarrier(,()->{
            System.out.println("召唤神龙成功！");
        });

        for (int i = ; i <= ; i++) {
            final int temp = i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"收集"+temp+"个龙珠");
                try {
                    cyclicBarrier.await(); // 等待
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

Semaphore(信号量)

规定有几个线程就最多只能有几个线程执行。 - semaphore.acquire(); 获得，假设如果已经满了，等待，等待被释放为止 - semaphore.release();释放，会将当前的信号量释放 + 1，然后唤醒等待的线程！

作用：多个共享资源互斥的使用！并发限流，控制大的线程数！

COPYpublic class SemaphoreDemo {
    public static void main(String[] args) {
        // 线程数量：停车位! 限流！
        Semaphore semaphore = new Semaphore();

        for (int i = ; i <= ; i++) {
            new Thread(()->{
                // acquire() 得到
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"抢到车位");
                    TimeUnit.SECONDS.sleep();
                    System.out.println(Thread.currentThread().getName()+"离开车位");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release(); // release() 释放
                }

            },String.valueOf(i)).start();
        }

    }
}

读写锁

ReadLock 读锁，多线程可同时占有执行。 writeLock() 写锁，多线程中只能一个线程占有执行。

COPY/**
 * 独占锁（写锁） 一次只能被一个线程占有
 * 共享锁（读锁） 多个线程可以同时占有
 * ReadWriteLock
 * 读-读  可以共存！
 * 读-写  不能共存！
 * 写-写  不能共存！
 */
public class ReadWriteLockDemo {
    public static void main(String[] args) {
        MyCache myCache = new MyCache();

        // 写入
        for (int i = ; i <=  ; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.put(temp+"",temp+"");
            },String.valueOf(i)).start();
        }

        // 读取
        for (int i = ; i <=  ; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.get(temp+"");
            },String.valueOf(i)).start();
        }

    }
}

// 加锁的
class MyCacheLock{

    private volatile Map<String,Object> map = new HashMap<>();
    // 读写锁： 更加细粒度的控制
    private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private Lock lock = new ReentrantLock();

    // 存，写入的时候，只希望同时只有一个线程写
    public void put(String key,Object value){
        readWriteLock.writeLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"写入"+key);
            map.put(key,value);
            System.out.println(Thread.currentThread().getName()+"写入OK");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readWriteLock.writeLock().unlock();
        }
    }

    // 取，读，所有人都可以读！
    public void get(String key){
        readWriteLock.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"读取"+key);
            Object o = map.get(key);
            System.out.println(Thread.currentThread().getName()+"读取OK");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
    }

}

/**
 * 自定义缓存
 */
class MyCache{

    private volatile Map<String,Object> map = new HashMap<>();

    // 存，写
    public void put(String key,Object value){
        System.out.println(Thread.currentThread().getName()+"写入"+key);
        map.put(key,value);
        System.out.println(Thread.currentThread().getName()+"写入OK");
    }

    // 取，读
    public void get(String key){
        System.out.println(Thread.currentThread().getName()+"读取"+key);
        Object o = map.get(key);
        System.out.println(Thread.currentThread().getName()+"读取OK");
    }
}

阻塞队列

抛出异常，add==>removeCOPYpublic static void test1(){ // 队列的大小 ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(); System.out.println(blockingQueue.add("a")); System.out.println(blockingQueue.add("b")); System.out.println(blockingQueue.add("c")); // IllegalStateException: Queue full 抛出异常！ // System.out.println(blockingQueue.add("d")); System.out.println("=-==========="); System.out.println(blockingQueue.element()); // 查看队首元素是谁 System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); // java.util.NoSuchElementException 抛出异常！ // System.out.println(blockingQueue.remove()); }
有返回值，无抛出异常 offer==>poll peel（获取队首元素）

COPY public static void test2(){
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>();

        System.out.println(blockingQueue.offer("a"));
        System.out.println(blockingQueue.offer("b"));
        System.out.println(blockingQueue.offer("c"));

        System.out.println(blockingQueue.peek());
        // System.out.println(blockingQueue.offer("d")); // false 不抛出异常！
        System.out.println("============================");
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll()); // null  不抛出异常！
    }

3. 死死的阻塞等待 put==>take

COPYpublic static void test3() throws InterruptedException {
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>();

        // 一直阻塞
        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        // blockingQueue.put("d"); // 队列没有位置了，一直阻塞
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take()); // 没有这个元素，一直阻塞

    }

等待，阻塞（设置等待超时）依旧是offer==>poll

COPY public static void test4() throws InterruptedException {
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>();

        blockingQueue.offer("a");
        blockingQueue.offer("b");
        blockingQueue.offer("c");
        // blockingQueue.offer("d",2,TimeUnit.SECONDS); // 等待超过2秒就退出
        System.out.println("===============");
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        blockingQueue.poll(,TimeUnit.SECONDS); // 等待超过2秒就退出

    }

同步队列

和其他的BlockingQueue 不一样， SynchronousQueue 不存储元素
put了一个元素，必须从里面先take取出来，否则不能在put进去值！意思就是队列只能存一个值，取一个值。

COPYpublic static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>(); // 同步队列

        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+" put 1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+" put 2");
                blockingQueue.put("2");
                System.out.println(Thread.currentThread().getName()+" put 3");
                blockingQueue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"T1").start();


        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep();
                System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
                TimeUnit.SECONDS.sleep();
                System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
                TimeUnit.SECONDS.sleep();
                System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"T2").start();
    }

线程池

线程池的好处: 1. 降低资源的消耗 2. 提高响应的速度 3. 方便管理。

线程复用、可以控制最大并发数、管理线程

三大创建方法

ExecutorService threadPool = Executors.newSingleThreadExecutor();// 单个线程
ExecutorService threadPool = Executors.newFixedThreadPool(10);// 创建一个固定的线程池的大小
ExecutorService threadPool = Executors.newCachedThreadPool(); // 由执行的线程数决定大小。

具体过程：创建线程池对象--->执行---->关闭线程池

COPY try {
        for (int i = ; i <= ; i++) {
                // 使用了线程池之后，使用线程池来创建线程
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName());
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 线程池用完，程序结束，关闭线程池
            threadPool.shutdown();
        }

七大参数

一般开发中使用线程池最好不使用Executors去创建对象。如阿里巴巴开发手册所说的：OOM（内存溢出）

所以应该使用ThreadPoolExecutor来创建线程池查看源码得到以下： - 当执行线程数大于核心线程锁时，其它休眠的线程位置将会启用，最多不超过最大线程数。 - 当超过最大线程锁的线程任务会在阻塞队列中进行等待。 - 当此时并发执行的线程任务超过了最大线程数+阻塞队列长度时，拒绝策略起效。

拒绝策略有四种

new ThreadPoolExecutor.AbortPolicy() // 同时并发的线程数大于最大线程数+阻塞队列长度时，不处理超过部分的线程任务，直接抛出异常。
new ThreadPoolExecutor.DiscardPolicy() //同时并发的线程数大于最大线程数+阻塞队列长度时，直接丢掉超过部分的线程任务，不抛出异常。
new ThreadPoolExecutor.CallerRunsPolicy() // 同时并发的线程数大于最大线程数+阻塞队列长度时，超过部分的线程任务回归main主线程执行。
new ThreadPoolExecutor.DiscardOldestPolicy() //阻塞队列满了，尝试去和最先执行的线程发起竞争，也不会抛出异常！

COPYpublic ThreadPoolExecutor(int corePoolSize, //核心线程数
                              int maximumPoolSize, //最大线程数。最多几个线程并发。
                              long keepAliveTime, //当线程无任务时，几秒后结束该线程
                              TimeUnit unit,//线程结束的时间单位
                              BlockingQueue<Runnable> workQueue,//阻塞队列，限制等候线程数
                              ThreadFactory threadFactory,//线程工厂
                              RejectedExecutionHandler handler)//拒绝策略

例子如下：

COPYExecutorService threadPool = new ThreadPoolExecutor(
                ,
                Runtime.getRuntime().availableProcessors(),
                ,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());  //队列满了，尝试去和最早的竞争，也不会抛出异常！