juc并发编程

1、什么是JUC

JUC指的是：Java里的三个包
java.util.concurrent
java.util.concurrent.atomic：原子性
java.util.concurrent.locks：lock锁

2、线程和进程

进程：程序执行的一次过程，一个进程包含一个或多个线程。进程是资源分配的单位。

线程：可以指程序执行过程中，负责实现某个功能的单位。线程是CPU调度和执行的单位。

并发和并行

并发：同一时刻，多个线程交替执行。（一个CPU交替执行线程）
并行：同一时刻，多个线程同时执行。

# 获取cpu的核数（cpu密集型；io密集型）
System.out.println(Runtime.getRuntime().availableProcessors());

线程的状态：

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

wait/sleep

1.来自不同的类
wait——Object
sleep——Thread

2.关于锁的释放
wait会、sleep不会

3.使用范围
wait必须在同步代码块中

4.sleep不需要被唤醒（时间到了自动退出阻塞）；wait需要被唤醒

3、锁

公平锁和非公平锁（锁的底层）
公平锁：十分公平，不能插队。
非公平锁：十分不公平，可以插队。（默认非公平锁）

Lock锁

Lock锁是一个接口，他有三个实现类：
ReentrantLock类
ReentrantReadWriteLock.ReadLock
ReentrantReadWriteLock.WriteLock

Lock锁 & synchronized

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

生产者 & 消费者传统的synchronized

// 判断等待，业务，通知
class Data {
 private int i = 0;
 // +1
 public synchronized void increment() throws InterruptedException {
     if (i != 0) {
         this.wait();
     }
     i++;
     System.out.println(Thread.currentThread().getName() + "=>" + i);
     // 通知其他线程我+1完成
     this.notifyAll();
 }
 // -1
 public synchronized void decrement() throws InterruptedException {
     if (i==0){
         this.wait();
     }
     i--;
     System.out.println(Thread.currentThread().getName() + "=>" + i);
     // 通知其他线程，我-1完毕
     this.notifyAll();
 }
}

if改成while解决虚假唤醒

JUC版

class Data {
 private int i = 0;
 Lock lock = new ReentrantLock();
 Condition condition = lock.newCondition();
 // +1
 public  void increment() throws InterruptedException {
     lock.lock();
     try {
         while (i != 0) {
             condition.await();
         }
         i++;
         System.out.println(Thread.currentThread().getName() + "=>" + i);
         // 通知其他线程我+1完成
         condition.signalAll();
     } catch (InterruptedException e) {
         e.printStackTrace();
     } finally {
         lock.unlock();
     }
 }
 // -1
 public void decrement() throws InterruptedException {
     lock.lock();
     try {
         while (i==0){
             condition.await();
         }
         i--;
         System.out.println(Thread.currentThread().getName() + "=>" + i);
         // 通知其他线程，我-1完毕
         condition.signalAll();
     } catch (InterruptedException e) {
         e.printStackTrace();
     } finally {
         lock.unlock();
     }
 }
}

Condition实现精准通知唤醒

condition2.signal();

关于锁的八个问题

staic修饰符锁的是类
无static修饰符时锁的是方法的调用者

4、集合类不安全

java.util.ConcurrentModificationException 并发修改异常

ArryList

        //CopyOnWrite 写入时复制 一种优化策略

HashSet

//Hashset不安全
Set set= Collections.synchronizedSet(new HashSet<>());
Set set= new CopyOnWriteArraySet<>();

hashset集合的底层是hashmap的key

HashMap

Map map=new ConcurrentHashMap<>();

5.Callable

Callable接口类似于Runnable接口，线程第三种创建方式。

可以抛出异常。
可以有返回值。
方法不同与Runnable接口。Call方法

futureTask

FutureTask类实现了RunnableFuture接口，继承了Runnable接口和Future接口。它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。

//两个构造器
public FutureTask(Callable callable) {
}
public FutureTask(Runnable runnable, V result) {
}

public class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyThread thread=new MyThread();
        FutureTask futureTask=new FutureTask<>(thread);
        new Thread(futureTask,"A").start();
        new Thread(futureTask,"B").start();// 打印一个Call，结果会被缓存，提高效率

        System.out.println(futureTask.get());  //获取callable返回接口,get方法可能会产生阻塞
    }
}

class MyThread implements Callable{

    @Override
    public String call() throws Exception {
        return "1024";
    }
}

6.常用辅助类 CountDownLatch 减法计数器

public static void main(String[] args) throws InterruptedException {
        //总数6
        CountDownLatch countDownLatch = new CountDownLatch(6);
        for (int i = 0; i < 6; i++) {
            new  Thread(()->{
                System.out.println(Thread.currentThread().getName() + "GO out");
                countDownLatch.countDown();
            },String.valueOf(i)).start();
        }
        countDownLatch.await();
        System.out.println("close door");
    }

原理：countDownLatch.countDown(); //数量-1

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

CyclicBarrier 加法计数器

例子：七龙珠唤神龙

public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3,()->{
            System.out.println("召唤神龙成功");// 如果计数器为7，线程只有6个，则会等待，不进行召唤神龙
        });

        for (int i = 1; i <= 7; 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();
                }
                System.out.println(Thread.currentThread().getName());
            }).start();

        }
    }

Semaphore：信号量

 public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < 6; i++) {
            new Thread(()->{
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"抢到车位");
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println(Thread.currentThread().getName()+"离开车位");

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    semaphore.release();
                }
            }).start();
        }
    }

原理：
semaphore.acquire(); //获取，
semaphore.release(); //释放，唤醒在等待的线程
作用：多个共享资源的互斥使用。并发限流，控制最大的线程数

7.读写锁

独占锁（写锁）：一次只能被一个线程占有，不能读和写
共享锁（读锁）：多个线程可以同时占有，可以同时读

public static void main(String[] args) {
        MyCacheLock myCache = new MyCacheLock();
        // 5个线程写
        for (int i = 1; i <= 5; i++) {
            final int temp = i;
            new Thread(() -> {
                myCache.put(temp + "", temp + "");
            }, String.valueOf(i)).start();
        }
        // 5个线程读
        for (int i = 1; i <= 5; i++) {
            final int temp = i;
            new Thread(() -> {
                myCache.get(temp + "");
            }, String.valueOf(i)).start();
        }
    }
}
    
    class MyCacheLock {
        private volatile Map map = new HashMap<>();
        // 读写锁，更加细粒度的控制
        private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

        // 写，同时只有一个线程写
        public void put(String key, Object obj) {
            readWriteLock.writeLock().lock();
            try {
                System.out.println(Thread.currentThread().getName() + "写入");
                map.put(key, obj);
                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() + "读取");
                map.get(key);
                System.out.println(Thread.currentThread().getName() + "读取OK");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                readWriteLock.readLock().unlock();
            }
        }
    }

8.阻塞队列BlockingQueue

使用场景：多线程并发处理，线程池

// 阻塞，超时等待
    public static void test4() throws InterruptedException {
        // 队列的大小为3
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
        // offer返回boolean值
        boolean flag1 = blockingQueue.offer("a");
        boolean flag2 = blockingQueue.offer("b");
        boolean flag3 = blockingQueue.offer("c");
        // offer添加元素超过队列的长度会返回false；并且等待指定时间后推出，向下执行
        boolean flag4 = blockingQueue.offer("d", 2, TimeUnit.SECONDS);
        System.out.println("=========");
        // poll()返回本次移除的元素
        Object poll1 = blockingQueue.poll();
        Object poll2 = blockingQueue.poll();
        Object poll3 = blockingQueue.poll();
        // 队列中没有元素仍继续移除元素会打印出null，等待指定之间后退出。
        Object poll4 = blockingQueue.poll(2,TimeUnit.SECONDS);
    }

SynchronousQueue同步队列
进去一个元素，必须等待取出这个元素后，才能放下一个元素。put()、take()

 public static void main(String[] args) {
        BlockingQueue 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(3);
                System.out.println(Thread.currentThread().getName()+blockingQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+blockingQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        },"T2").start();
    }

9.线程池

池化技术：事先准备好一些资源，有人使用来我这里拿，用完归还。

好处：
1、降低资源的消耗
2、提高响应的速度
3、方便管理

3大方法

//        ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程
//        ExecutorService threadPool =Executors.newFixedThreadPool(5);    //固定线程池大小
        ExecutorService threadPool =Executors.newCachedThreadPool();

7大参数

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }

三大方法的本质ThreadPoolExecutor()

public ThreadPoolExecutor(int corePoolSize,//核心线程池大小
                          int maximumPoolSize,//最大核心线程池大小
                          long keepAliveTime,//超时释放
                          TimeUnit unit,//超时单位
                          BlockingQueue workQueue,//阻塞队列
                          ThreadFactory threadFactory,//创建线程的工厂
                          RejectedExecutionHandler handler//拒绝策略) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

自定义创建线程池

ExecutorService threadPool=new ThreadPoolExecutor(2,5,
                3, TimeUnit.SECONDS,
                            new LinkedBlockingQueue<>(3),
                            Executors.defaultThreadFactory(),
                            new ThreadPoolExecutor.AbortPolicy());//银行满了，有人进，抛出异常

4大拒绝策略

new ThreadPoolExecutor.AbortPolicy(); // 抛出异常
new ThreadPoolExecutor.CallerRunsPolicy();// 哪来的去哪（主线程来的，就回去让主线程执行）
new ThreadPoolExecutor.DiscardPolicy();// 丢掉任务，不抛出异常
new ThreadPoolExecutor.DiscardOldestPolicy();// 尝试和最早的竞争，竞争失败了也丢掉任务，也不抛出异常

最大线程定义 CPU密集型

最大线程数，CPU几核的就是几，可以保持CPU效率最高。

IO密集型

判断程序中十分耗IO的线程数量，大于这个数，一般是这个数的两倍。

10.四大函数式接口

新时代程序员必须掌握：lambda表达式、链式编程、函数式接口、stream流式计算
函数式接口：只有一个方法的接口。

Function函数型接口

有一个输入参数，有一个输出（返回值）。

predicate断定型接口

有一个输入参数，返回值只能是boolean值。

consumer消费型接口

有一个输入参数，没有返回值。

supplier供给型接口

没有输入参数，有一个输出（返回值）。

juc并发编程

Java相关栏目本月热门文章