JUC(并发编程)
java.util.concurrent 包是在并发编程中使用的工具类,有以下三个包:
java.util 工具包
业务: 普通的线程代码 Thread
Runnable 没有返回值、效率相比于Callable相对较低!)
(
线程和进程
进程:一个程序,qq.exe,music.exe ,程序的集合
一个进程往往包含了多个线程,至少包含一个
Java默认有两个线程? 2个 main线程、GC线程
线程: 开了一个进程Typora ,写字,自动保存(线程负责的)
对于java而言: Thread、Runnable、Callable
Java 真的可以开启线程吗? 开不了的!
public synchronized void start() { /** * This method is not invoked for the main method thread or "system" * group threads created/set up by the VM. Any new functionality added * to this method in the future may have to also be added to the VM. * * A zero status value corresponds to state "NEW". */ if (threadStatus != 0) throw new IllegalThreadStateException(); /* Notify the group that this thread is about to be started * so that it can be added to the group's list of threads * and the group's unstarted count can be decremented. */ group.add(this); boolean started = false; try { start0(); started = true; } finally { try { if (!started) { group.threadStartFailed(this); } } catch (Throwable ignore) { /* do nothing. If start0 threw a Throwable then it will be passed up the call stack */ } } } private native void start0(); // 本地方法,调用底层的c++,java无法直接操作硬件!
并发和并行
并发编程: 并发和并行
并发(多线程操作同一个资源)
并行 (多个人一起行走)
代码查看cpu核数
package com.kuang.juc.demo01; public class Test { public static void main(String[] args) { // 获取CPU的核数 // CPU密集型、IO密集型 System.out.println(Runtime.getRuntime().availableProcessors()); } }
并发编程的本质:充分利用CPU资源
线程的几个状态
public enum State { /** * Thread state for a thread which has not yet started. */ NEW, // 新生 /** * Thread state for a runnable thread. A thread in the runnable * state is executing in the Java virtual machine but it may * be waiting for other resources from the operating system * such as processor. */ RUNNABLE, // 运行 /** * Thread state for a thread blocked waiting for a monitor lock. * A thread in the blocked state is waiting for a monitor lock * to enter a synchronized block/method or * reenter a synchronized block/method after calling * {@link Object#wait() Object.wait}. */ BLOCKED, // 阻塞 /** * Thread state for a waiting thread. * A thread is in the waiting state due to calling one of the * following methods: * <ul> * <li>{@link Object#wait() Object.wait} with no timeout</li> * <li>{@link #join() Thread.join} with no timeout</li> * <li>{@link LockSupport#park() LockSupport.park}</li> * </ul> * * <p>A thread in the waiting state is waiting for another thread to * perform a particular action. * * For example, a thread that has called <tt>Object.wait()</tt> * on an object is waiting for another thread to call * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on * that object. A thread that has called <tt>Thread.join()</tt> * is waiting for a specified thread to terminate. */ WAITING, // 等待(死死地等),阻塞 /** * Thread state for a waiting thread with a specified waiting time. * A thread is in the timed waiting state due to calling one of * the following methods with a specified positive waiting time: * <ul> * <li>{@link #sleep Thread.sleep}</li> * <li>{@link Object#wait(long) Object.wait} with timeout</li> * <li>{@link #join(long) Thread.join} with timeout</li> * <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li> * <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li> * </ul> */ TIMED_WAITING, // 超时等待,过期不候 /** * Thread state for a terminated thread. * The thread has completed execution. */ TERMINATED; // 终止 }
wait/sleep 区别
1、来自不同的类
wait => Object
sleep => Thread
2、关于锁的释放
wait 会释放锁,sleep抱着锁睡觉,不会释放!
3、使用的范围是不同的
wait必须在同步代码块中使用
sleep可以在任何地方睡,可以在任何地方使用
4、是否需要捕获异常
wait 也需要捕获异常
sleep 必须要捕获异常
传统 synchonized
package com.kuang.juc.demo01; // 基本的买票例子 /** * 真正的多线程开发,降低耦合性! * 线程就是一个单独的资源类,没有任何的附属操作! * 1.属性、方法 */ public class SaleTicketDemo01 { public static void main(String[] args) { // 并发:多线程操作同一个资源类,把资源类丢入线程就可以了! Ticket ticket = new Ticket(); new Thread(()->{ for (int i = 1; i < 40; i++) { ticket.sale(); } },"A").start(); new Thread(()->{ for (int i = 1; i < 40; i++) { ticket.sale(); } },"B").start(); new Thread(()->{ for (int i = 1; i < 40; i++) { ticket.sale(); } },"C").start(); } } // 资源类 OOP class Ticket{ // 属性,方法 private int nums = 30; // 卖票的方式 // synchronized 本质:队列 + 锁 public synchronized void sale(){ if (nums>0){ System.out.println(Thread.currentThread().getName()+"卖出了"+nums--+"票。还剩: "+nums+"张票"); } } }
结果:
)
总结
加了synchonized
关键字后,变为同步方法(队列加锁)。相当于你去学校食堂去买饭,买饭窗口排成一队,一个一个来,排到的人过去把碗给阿姨之后,相当于给自己加了一把锁,阿姨打完饭,再给碗给你,相当于这把锁就释放了。释放了之后,后面的人就可以再次排队买饭了。
Lock 接口
)
)
)
公平锁:十分公平,有先来后到顺序 3h 3s
非公平锁:十分不公平,可以插队(默认)
lock实现买票
package com.kuang.juc.demo01; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class SaleTicketDemo02 { public static void main(String[] args) { // 并发:多线程操作同一个资源类,把资源类丢入线程就可以了! Ticket2 ticket = new Ticket2(); new Thread(()->{ for (int i = 1; i < 40; i++) { ticket.sale(); } },"A").start(); new Thread(()->{ for (int i = 1; i < 40; i++) { ticket.sale(); } },"B").start(); new Thread(()->{ for (int i = 1; i < 40; i++) { ticket.sale(); } },"C").start(); } } // lock class Ticket2{ // 属性,方法 private int nums = 30; Lock lock = new ReentrantLock(); public void sale(){ lock.lock(); //加锁 try { // 业务代码 if (nums>0){ System.out.println(Thread.currentThread().getName()+"卖出了"+nums--+"票。还剩: "+nums+"张票"); } } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } }
Synchonized 和 Lock 区别
锁是什么?如何判断锁的是谁?
面试: 单例模式、排序算法、生产者消费者问题、死锁
生产者消费者问题 Synchonized 版
package com.kuang.juc.pc; /** * 线程之间的通信问题:生产者和消费者问题! 等待唤醒,通知唤醒 * 线程交替执行 A B 操作同一个变量 num = 0 * A num+1 * B num-1 */ public class A { public static void main(String[] args) { Data data = new Data(); new Thread(()->{ for (int i = 0; i < 10; i++) { data.increment(); } },"A").start(); new Thread(()->{ for (int i = 0; i < 10; i++) { data.decrement(); } },"B").start(); } } // 口诀:判断等待,业务,通知! class Data{ // 数字 资源类 private int num = 0; //+1 public synchronized void increment(){ if (num != 0){ // 0 // 等待 try { this.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } num++; System.out.println(Thread.currentThread().getName()+"==>"+num); // 通知其他线程,我+1完毕了 this.notifyAll(); } //-1 public synchronized void decrement(){ if (num == 0){ //1 // 等待 try { this.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } num--; System.out.println(Thread.currentThread().getName()+"==>"+num); // 通知其他线程,我-1完毕了 this.notifyAll(); } }
问题存在, A B C D 4个线程
)
出现问题,发现出现了2和3
解决
)
将原先的if判断 ==> while判断,防止虚假唤醒!
注意点: 因为if只会判断一次(第二个线程进入之后不判断了,直接+1,出现了2的情况,不安全!),而使用while每次进入该方法都会进行判断, 非常安全!
JUC版的 生产者消费者问题
)
通过lock找到 Condition
)
代码实现
package com.kuang.juc.pc; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * 线程之间的通信问题:生产者和消费者问题! 等待唤醒,通知唤醒 * 线程交替执行 A B 操作同一个变量 num = 0 * A num+1 * B num-1 */ public class B { public static void main(String[] args) { Data2 data = new Data2(); new Thread(()->{ for (int i = 0; i < 10; i++) { try { data.increment(); } catch (InterruptedException e) { e.printStackTrace(); } } },"A").start(); new Thread(()->{ for (int i = 0; i < 10; i++) { try { data.decrement(); } catch (InterruptedException e) { e.printStackTrace(); } } },"B").start(); new Thread(()->{ for (int i = 0; i < 10; i++) { try { data.increment(); } catch (InterruptedException e) { e.printStackTrace(); } } },"C").start(); new Thread(()->{ for (int i = 0; i < 10; i++) { try { data.decrement(); } catch (InterruptedException e) { e.printStackTrace(); } } },"D").start(); } } class Data2{ // 数字 资源类 private int num = 0; Lock lock = new ReentrantLock(); Condition condition = lock.newCondition(); //condition.await(); //等待 //condition.signalAll(); //唤醒 //+1 public void increment() throws InterruptedException { lock.lock(); //加锁 try { // 业务代码 while (num != 0){ // 0 // 等待 condition.await(); } num++; System.out.println(Thread.currentThread().getName()+"==>"+num); // 通知其他线程,我+1完毕了! condition.signalAll(); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } //-1 public void decrement() throws InterruptedException { lock.lock(); // 加锁 try { // 业务代码 while (num == 0){ //1 condition.await(); } num--; System.out.println(Thread.currentThread().getName()+"==>"+num); // 通知其他线程,我-1完毕了! condition.signalAll(); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } }
结果:
)
问题: 如果想会让A执行完 B执行 C执行 D执行
任何一个新的技术绝对不是仅仅覆盖了原来的技术,有自己的优势以及补充!
Condition 精准通知和唤醒线程
代码实现:
package com.kuang.juc.pc; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * A B C 3个线程顺序执行 * A-->B-->C-->A...... */ public class C { public static void main(String[] args) { Data3 data = new Data3(); new Thread(()->{ for (int i = 0; i < 10; i++) { data.printA(); } },"A").start(); new Thread(()->{ for (int i = 0; i < 10; i++) { data.printB(); } },"B").start(); new Thread(()->{ for (int i = 0; i < 10; i++) { data.printC(); } },"C").start(); } } class Data3 { // 资源类 private Lock lock = new ReentrantLock(); private Condition condition1 = lock.newCondition(); //同步监视器 private Condition condition2 = lock.newCondition(); private Condition condition3 = lock.newCondition(); private int flag = 1; // 1A 2B 3C public void printA() { lock.lock(); try { // 业务代码,判断等待、执行、通知,三部曲 while (flag != 1) { //等待 condition1.await(); } System.out.println(Thread.currentThread().getName() + "=>AAAAAAAAAA"); // 唤醒,唤醒指定的人,B flag = 2; condition2.signal(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } public void printB() { lock.lock(); try { while (flag != 2) { condition2.await(); } System.out.println(Thread.currentThread().getName() + "=>BBBBBBBBBB"); // 唤醒,唤醒指定的人,C flag = 3; condition3.signal(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } public void printC() { lock.lock(); try { while (flag != 3) { condition3.await(); } System.out.println(Thread.currentThread().getName() + "=>CCCCCCCCCC"); // 唤醒,唤醒指定的人,A flag = 1; condition1.signal(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } }
查看结果:
如何判断锁的是谁?永远的知道什么是锁,锁的到底是谁?
对象,class
深刻理解我们的锁
package com.kuang.juc.lock8; import java.util.concurrent.TimeUnit; /** * 8锁就是关于锁的8个问题 * 1、标准情况下,两个线程先打印 打电话还是发短信? 1/发短信 2/打电话 * 2、发短信延迟4s,两个线程先打印 打电话还是发短信? 1/发短信 2/打电话 产生原因:只拿到了调用者的一把锁! */ public class Test1 { public static void main(String[] args) { Phone phone = new Phone(); // 锁的存在 new Thread(()->{ phone.sendSms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(()->{ phone.call(); },"B").start(); } } class Phone{ // synchronized 锁的对象是方法调用者! // 两个方法用的是同一个锁,谁先拿到谁执行! public synchronized void sendSms(){ try { TimeUnit.SECONDS.sleep(4); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } public synchronized void call(){ System.out.println("打电话"); } }
package com.kuang.juc.lock8; import java.util.concurrent.TimeUnit; /** * 3、增加了一个普通方法后,是先打印发短信还是hello? 1/hello 2/发短信 普通方法 * 4、两个对象,两个同步方法,是先打印发短信还是打印打电话? 1/打电话 2/发短信 * */ public class Test2 { public static void main(String[] args) { // 两个对象 两个调用者,两把锁! Phone2 phone1 = new Phone2(); Phone2 phone2 = new Phone2(); // 锁的存在 new Thread(()->{ phone1.sendSms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(()->{ phone2.call(); },"B").start(); } } class Phone2{ // synchronized 锁的对象是方法调用者! public synchronized void sendSms(){ try { TimeUnit.SECONDS.sleep(4); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } public synchronized void call(){ System.out.println("打电话"); } // 这里没有锁! 不是同步方法,不受锁的影响! public void hello(){ System.out.println("hello"); } }
package com.kuang.juc.lock8; import java.util.concurrent.TimeUnit; /** * 5、增加两个静态的同步方法,只有一个对象,先打印 发短信还是打电话? 1/发短信 2/打电话 因为锁的是class,类模板只有一个,一把锁! * 6、两个对象,两个静态的同步方法,先打印 发短信还是打电话? 1/发短信 2/打电话 因为锁的是class,类模板只有一个,一把锁! * */ public class Test3 { public static void main(String[] args) { // 两个对象的class类模板只有一个,static,锁的是class Phone3 phone1 = new Phone3(); Phone3 phone2 = new Phone3(); // 锁的存在 new Thread(()->{ phone1.sendSms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(()->{ phone2.call(); },"B").start(); } } // Phone3只有唯一的一个 class 对象 class Phone3{ // synchronized 锁的对象是方法调用者! // static 静态方法,类一加载就有了! 锁的是Class // 锁的Class 模板! public static synchronized void sendSms(){ try { TimeUnit.SECONDS.sleep(4); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } public static synchronized void call(){ System.out.println("打电话"); } // 这里没有锁! 不是同步方法,不受锁的影响! public void hello(){ System.out.println("hello"); } }
package com.kuang.juc.lock8; import java.util.concurrent.TimeUnit; /** * 1、1个静态同步方法,1个普通同步方法 ,一个对象, 先打印发短信还是打电话? 1/打电话 2/发短信 * 产生1的原因: 调用打电话一把锁,锁的是调用者;调用发短信另一把锁,锁的类模板class! * 2、1个静态同步方法,1个普通同步方法 ,两个对象, 先打印发短信还是打电话? 1/打电话 2/发短信 * 产生2的原因: 调用打电话一把锁,锁的是调用者phone2;调用发短信另一把锁,锁的类模板class! */ public class Test4 { public static void main(String[] args) { // 拿到的锁是类模板class Phone4 phone1 = new Phone4(); // 拿到的锁是调用者phone2 Phone4 phone2 = new Phone4(); // 锁的存在 new Thread(()->{ phone1.sendSms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(()->{ phone2.call(); },"B").start(); } } class Phone4{ // 静态同步方法 锁的是class类模板 public static synchronized void sendSms(){ try { TimeUnit.SECONDS.sleep(4); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } // 普通的同步方法, 锁的是调用者 public synchronized void call(){ System.out.println("打电话"); } }
总结
上述例子中发现:
synchronized实现同步的基础:java中的每一个对象都可以作为锁!
具体的表现为以下三种形式:
List不安全
package com.kuang.juc.unsafe; import java.lang.reflect.Array; import java.util.*; import java.util.concurrent.CopyOnWriteArrayList; // java.util.ConcurrentModificationException 并发修改异常! public class ListTest { public static void main(String[] args) { // 并发下 ArrayList 是不安全的吗? synchonized; /** * 解决方案: * 1、 List<String> list = new Vector<>(); * 2、 List<String> list = Collections.synchronizedList(new ArrayList<>()); * 3、 List<String> list = new CopyOnWriteArrayList<>(); */ // CopyOnWrite 写入时复制 COW 计算机程序设计领域的一种优化策略! // 多个线程调用的时候,list,读取的时候,固定的,写入 (可能存在覆盖) // 在写入的时候避免覆盖造成数据问题! // CopyOnWriteArrayList 比 Vector 厉害在哪里? // Vector(add方法用到了synchonized,效率较低) // CopyOnWriteArrayList(add方法用到了lock锁,效率比vector高) List<String> list = new CopyOnWriteArrayList<>(); for (int i = 1; i <= 10; i++) { new Thread(()->{ list.add(UUID.randomUUID().toString().substring(0,5)); System.out.println(list); },String.valueOf(i)).start(); } } }
查看CopyOnWriteArrayList的add方法:
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-a5AaUiNI-1625379008053)()]
学习方法: 1、先会用 2、货比三家 ,寻找其他解决方法 3、分析源码
Set 不安全
package com.kuang.juc.unsafe; import java.util.Set; import java.util.UUID; import java.util.concurrent.CopyOnWriteArraySet; // java.util.ConcurrentModificationException 并发修改异常 public class TestSet { public static void main(String[] args) { /** * 解决方案: * 1、Set<String> set = Collections.synchronizedSet(new HashSet<>()); 通过工具类转换为 synchonized * 2、Set<String> set = new CopyOnWriteArraySet<>(); 写入时复制,保证性能和安全! */ Set<String> set = new CopyOnWriteArraySet<>(); for (int i = 1; i <= 30; i++) { new Thread(()->{ set.add(UUID.randomUUID().toString().substring(0,5)); System.out.println(set); }).start(); } } }
HashSet底层是什么?
public HashSet() { map = new HashMap<>(); } // add. ashset的本质 就是 map的key,因为key是无法重复的! public boolean add(E e) { return map.put(e, PRESENT)==null; } // PRESENT. 不变的值! private static final Object PRESENT = new Object();
Map不安全
回顾map的基本操作
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cEPap1zO-1625379008055)()]
package com.kuang.juc.unsafe; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; // java.util.ConcurrentModificationException 并发修改异常! public class MapTest { public static void main(String[] args) { // map 是这样用的吗? 不是,多线程情况下 工作中不用 HashMap // 默认等价于什么? // 加载因子,初始化容量 /** * 解决方法: * 1、 Map<String, String> map = Collections.synchronizedMap(new HashMap<>()); 工具类装换为安全的map集合 * 2、 Map<String, String> map = new ConcurrentHashMap<>(); **/ Map<String, String> map = new ConcurrentHashMap<>(); for (int i = 1; i <= 30 ; i++) { new Thread(()->{ map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0,5)); System.out.println(map); },String.valueOf(i)).start(); } } }
1、可以有返回值
2、可以抛出异常
3、方法是不同的,run() / call()
package com.kuang.juc.callable; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; public 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 myThread = new MyThread(); // 实现了Callable接口 FutureTask futureTask = new FutureTask<>(myThread); // 适配类 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() throws Exception { // 如果有耗时的操作的话 System.out.println("call()"); return 1024; } }
结果:
细节:
1、有缓存
2、结果可能需要等待,会阻塞!
案例:晚上大爷关自习室门
package com.kuang.juc.add; import java.util.concurrent.CountDownLatch; // 计数器 public class CountDownLatchDemo { public static void main(String[] args) throws InterruptedException { // 总数是6 教室中6个人 // 必须要执行任务 的时候再使用 CountDownLatch countDownLatch = new CountDownLatch(6); for (int i = 1; i <= 6; i++) { new Thread(()->{ System.out.println(Thread.currentThread().getName()+"go out"); countDownLatch.countDown(); // 数量 -1 },String.valueOf(i)).start(); } countDownLatch.await(); //等待计数器归零,然后再向下执行! System.out.println("我是你大爷,我要关门了!"); } }
结果:
原理:
countDownLatch.countDown()
// 数量 -1
countDownLatch.await()
//等待计数器归零,然后再向下执行!
每次有线程调用**countDown()**时候数量-1,假设计数器变为0;**countDownLatch.await()**就会被唤醒,继续向下执行!
相当于加法计数器
案例:集齐7颗龙珠召唤神龙
package com.kuang.juc.add; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; /** * eg: 集齐7颗龙珠召唤神龙! */ public class CyclicBarrierDemo { public static void main(String[] args) { // 召唤龙珠的线程 CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{ System.out.println("召唤神龙成功!"); }); for (int i = 0; i < 7; i++) { // lamabda能操作到变量i吗? no 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(); } } }
案例:抢车位
6车 ---- 3个停车位置
package com.kuang.juc.add; 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 <= 6; i++) { new Thread(()->{ // acquire() 得到 release() 释放 try { semaphore.acquire(); System.out.println(Thread.currentThread().getName()+"抢到车位"); TimeUnit.SECONDS.sleep(2); // 模拟停车2s System.out.println(Thread.currentThread().getName()+"离开车位"); } catch (InterruptedException e) { e.printStackTrace(); }finally { semaphore.release(); } },String.valueOf(i)+"号车").start(); } } }
结果:
原理:
semaphore.acquire()
获得,假设已经满了,等待,等待被释放为止
semaphore.release()
释放,会将当前的信号量释放+1,然后唤醒等待的线程!
使用场景:
1、多个共享资源互斥的时候使用!
2、并发限流,控制最大的线程数!
读可以被多线程同时读
写的时候只能由一个线程去写
package com.kuang.juc.rw; import java.util.HashMap; import java.util.Map; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; /** * ReadWriteLock * 读--读 可以共存! * 读--写 不能共存! * 写--写 不能共存! */ public class ReadWriteLockDemo { public static void main(String[] args) { //MyCache myCache = new MyCache(); MyCacheLock myCache = new MyCacheLock(); // 写入 for (int i = 1; i <= 5; i++) { final int temp = i; new Thread(()->{ myCache.put(temp+"",temp+""); },String.valueOf(i)).start(); } // 读取 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<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()+"写入"+key); map.put(key,value); System.out.println(Thread.currentThread().getName()+"写入完毕"); } 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()+"读取完毕"); } 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()+"写入完毕"); } // 取,读 public void get(String key){ System.out.println(Thread.currentThread().getName()+"读取"+key); Object o = map.get(key); System.out.println(Thread.currentThread().getName()+"读取完毕"); } }
运行结果:
总结:
独占锁: 也就是这里的写锁,一次只能被一个线程占有!
共享锁: 也就是这里的读锁,一次可以被多个线程同时占有!
读写锁是一种更加 细粒度 的锁
阻塞
队列
BlockingQueue不是一个新的东西!
什么情况下我们会用到阻塞队列?
1、多线程并发处理
2、线程池
学会使用队列
添加、移除,4组API
方式 | 抛出异常 | 不抛异常,有返回值 | 阻塞等待 | 超时等待 |
---|---|---|---|---|
添加 | add() | offer() | put() | offer(,) |
移除 | remove() | poll() | take() | poll(,) |
判断队列首部元素 | element() | peek() | ~ | ~ |
/** * 抛出异常 */ public static void test1(){ // 队列大小 ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3); 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.remove()); System.out.println(blockingQueue.element()); // 查看队首元素 System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); // java.util.NoSuchElementException 没有元素,抛出异常! // System.out.println(blockingQueue.remove()); }
/** * 不抛出异常,有返回值! */ public static void test2(){ //队列大小 ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3); System.out.println(blockingQueue.offer("A")); System.out.println(blockingQueue.offer("B")); System.out.println(blockingQueue.offer("C")); // System.out.println(blockingQueue.offer("D")); // 返回false,不抛出异常! System.out.println(blockingQueue.peek()); //检测队首元素 System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); // System.out.println(blockingQueue.poll()); // 返回null,不抛出异常! }
/** * 等待,阻塞(一直阻塞) */ public static void test3(){ ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3); // 一直阻塞 try { 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()); // 没有这个元素,一直阻塞! } catch (InterruptedException e) { e.printStackTrace(); } }
/** * 等待,阻塞(超时等待) */ public static void test4() throws InterruptedException { ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3); blockingQueue.offer("A"); blockingQueue.offer("B"); blockingQueue.offer("C"); // blockingQueue.offer("D",2, TimeUnit.SECONDS); // 等待超过两秒就退出! System.out.println("================================"); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); blockingQueue.poll(2,TimeUnit.SECONDS); // 超过两秒钟就不等了! }
SynchronousQueue 同步队列
容量为1
进去一个元素,必须等待这个元素取出来之后,才能继续往里面放元素!
put、take
package com.kuang.juc.synbq; import java.util.concurrent.BlockingQueue; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; /** * 同步队列 * 和其他的BlockingQueue 不一样 , SynchronousQueue 不存储元素 * put了一个元素,必须从里面先take取出来,否则不能在put进去值! */ public class SynchronousQueueDemo { public static void main(String[] args) { BlockingQueue<String> blockingQueue = new SynchronousQueue<String>(); // 同步队列 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(); } }
运行结果:
注意 :
同步队列 和 其他的 BlockingQueue 不一样
SynchronousQueue 不存储元素
put了一个元素,必须先从里面先take取出来,否则不能再put进去值!
池化技术
程序的运行,本质 : 占用系统的资源!
想要优化资源的使用! ==> 池化技术
线程池、连接池、内存池、对象池… 它的创建和销毁十分消耗系统资源!
池化技术: 事先准备好一些资源放在池子里,有人要用,就来我这里拿,用完之后再还给我!
线程池的好处:
1、降低资源消耗
2、提高相应的速度
3、方便管理
一句话总结: 线程实现了复用、可以控制最大并发数、方便管理线程
线程池:三大方法
阿里巴巴开发手册规定:
三大方法测试
package com.kuang.juc.pool; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; // Executors 工具类 ,3大方法 public class Demo01 { public static void main(String[] args) { ExecutorService threadPool = Executors.newSingleThreadExecutor(); // 单个线程 // ExecutorService threadPool = Executors.newFixedThreadPool(5); // 创建一个固定线程池的大小 // ExecutorService threadPool = Executors.newCachedThreadPool(); // 有弹性的,遇强则强,遇弱则弱 try { for (int i = 1; i <= 100; i++) { // 使用了线程池之后,就是用线程池来创建线程 threadPool.execute(()->{ System.out.println(Thread.currentThread().getName()+ " ok"); }); } } catch (Exception e) { e.printStackTrace(); } finally { // 线程池用完,程序结束,要将线程池关闭! threadPool.shutdown(); } } }
所谓3大方法,就是创建线程池的三种方法:
7大参数
源码分析:
// 1、Executors.newSingleThreadExecutor() public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); } // 2、Executors.newFixedThreadPool(5) public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } // 3、Executors.newCachedThreadPool() public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, // 21亿 OOM 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } // 本质调用了ThreadPoolExecutor(7大参数) public ThreadPoolExecutor(int corePoolSize, // 核心线程池大小 int maximumPoolSize, // 最大线程池大小 long keepAliveTime, // 超时了没有人调用就会释放 TimeUnit unit, // 超时的单位 BlockingQueue<Runnable> 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; }
7大参数图解(模拟银行取钱)
上面图解小故事说明:当一个人去银行取钱的时候,发现只有两个窗口开着的,这两个窗口相当于核心线程池大小
然后它去1号办理业务了,又进来一个人他看到2号窗口空着,就过去办理业务了,
来了第三个人,他发现三个窗口没营业,又发现两个窗户正在有人办理业务
就去候客区的椅子上等着,进来第4个人,也去候客区等着,进来第五个人,也去候客区等着
这个时候候客区满了,就会触发银行打电话叫3,4,5号窗口营业员赶紧回来工作
这时候3,4,5号营业员回来了
这里的候客区相当于阻塞队列,当阻塞队列已满时,就会触发最大线程池机制,也就是3,4,5号以及前面1,2号窗口共同组成最大线程池
这个时候3,4,5号营业员回来了,候客区的这三个人就过去办理业务了,此时窗口已经满员,达到了最大线程池大小
再后来,又进来三个人,发现窗口已满,进入候客区
这个时候,窗口已满,候客区已满
突然又进来一个人,他没地方可待了,就会触发拒绝策略
这个人要选择继续等待,还是哪里来回哪里去,还是把1号位的人赶走自己上位
到了下午,人非常少,1号,2号窗口的人寥寥无几,3,4,5号窗口等了两个小时都一直没人,就先关闭窗口了,这里下班了相当于线程池中超时了没有人调用就会释放( keepAliveTime ),超时不候
手动创建线程池
// 自定义线程池! 工作常用 new ThreadPoolExecutor() ThreadPoolExecutor threadPool = new ThreadPoolExecutor( 2, 5, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>(3), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了, 抛弃队列中最老的那个,代替他的位置进入到队列中,也不会抛出异常! );
4种拒绝策略
package com.kuang.juc.pool; import java.util.concurrent.*; /** * new ThreadPoolExecutor.AbortPolicy() // 银行满了,还有人进来,就不处理这个人的了,抛出异常! * new ThreadPoolExecutor.CallerRunsPolicy() // 哪里来的去哪里! 我不受理,让main线程去受理! * new ThreadPoolExecutor.DiscardPolicy() // 队列满了,不会抛出异常。 丢掉任务,不执行这个线程! * new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了, 抛弃队列中最老的那个,代替他的位置进入到队列中,也不会抛出异常! */ public class Demo01 { public static void main(String[] args) { // 自定义线程池! 工作常用 new ThreadPoolExecutor() ThreadPoolExecutor threadPool = new ThreadPoolExecutor( 2, 5, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>(3), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了, 抛弃队列中最老的那个,代替他的位置进入到队列中,也不会抛出异常! ); try { // 最大 8 , 最大承载 = Deque + max // java.util.concurrent.RejectedExecutionException 超出了最大承载抛出异常! for (int i = 1; i <= 9; i++) { // 使用了线程池之后,就是用线程池来创建线程 threadPool.execute(()->{ System.out.println(Thread.currentThread().getName()+ " ok"); }); } } catch (Exception e) { e.printStackTrace(); } finally { // 线程池用完,程序结束,要将线程池关闭! threadPool.shutdown(); } } }
四种拒绝策略总结:
new ThreadPoolExecutor.AbortPolicy() // 银行满了,还有人进来,就不处理这个人的了,抛出异常**(默认)**
new ThreadPoolExecutor.CallerRunsPolicy() // 哪里来的去哪里! 我不受理,让main线程去受理!
new ThreadPoolExecutor.DiscardPolicy() // 队列满了,不会抛出异常。 丢掉任务,不执行这个线程!
new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了, 抛弃队列中最老的那个,代替他的位置进入到队列中,也不会抛出异常!
小结和拓展
问题 :线程池的最大的大小如何设置?
了解IO密集型和CPU密集型:(调优)
// 最大线程池到底应该如何规定? // 1、CPU密集型 ,几核,就是几 ,可以保证cpu效率最高! // 2、IO密集型 > 判断你程序中十分耗io资源的线程! // 程序 15个大型任务 io十分占用资源! System.out.println(Runtime.getRuntime().availableProcessors()); // 获取电脑cpu的核数 ThreadPoolExecutor threadPool = new ThreadPoolExecutor( 2, Runtime.getRuntime().availableProcessors(), 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>(3), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了, 抛弃队列中最老的那个,代替他的位置进入到队列中,也不会抛出异常! );
新时代的程序员 : lambda表达式、链式编程、函数式接口、Stream流式计算!
函数式接口 :只有一个方法的接口
@FunctionalInterface public interface Runnable { public abstract void run(); } // 超级多的FunctionalInterface // lambda表达式、链式编程、函数式接口、Stream流式计算!(jdk8新特性) // 简化编程模型,在新版本的框架中大量应用! // foreach(参数为消费者类型的函数式接口)
代码测试:
1、Function 函数式接口
package com.kuang.juc.function; import java.util.function.Function; /** * Function 函数型接口, 有一个输入参数,有一个输出参数 * 只要是 函数式接口 就可以使用lambda表达式简化! */ public class Demo01 { public static void main(String[] args) { // 工具方法: 输出输入的值! // Function<String,String> function = new Function<String, String>() { // @Override // public String apply(String str) { // return str; // } // }; Function<String,String> function = (str)->{ return str;}; System.out.println(function.apply("lac")); } }
2、Predicate 断定型接口
package com.kuang.juc.function; import java.util.function.Predicate; /** * 断定型接口: 有一个输入参数,返回值只能是 布尔值! */ public class Demo02 { public static void main(String[] args) { // 判断字符串是否为空 // Predicate<String> predicate = new Predicate<String>() { // @Override // public boolean test(String str) { // return str.isEmpty(); // } // }; Predicate<String> predicate = (str)->{ return str.isEmpty(); }; System.out.println(predicate.test("")); } }
3、Consumer 消费型接口
package com.kuang.juc.function; import java.util.function.Consumer; /** * 消费者接口 Consumer: 只有输入,没有返回值! */ public class Demo03 { public static void main(String[] args) { // Consumer<Integer> consumer = new Consumer<Integer>() { // @Override // public void accept(Integer integer) { // System.out.println(integer); // } // }; Consumer<Integer> consumer = (i)->{ System.out.println(i); }; consumer.accept(1024); } }
4、Supplier 供给型接口
package com.kuang.juc.function; import java.util.function.Supplier; /**、 * 供给型接口 Supplier : 只有返回值,没有参数 */ public class Demo04 { public static void main(String[] args) { // Supplier<String> supplier = new Supplier<String>() { // @Override // public String get() { // return "xx-x"; // } // }; Supplier<String> supplier = ()->{ return "xx-x"; }; System.out.println(supplier.get()); } }
什么是流式计算
大数据 : 存储 + 计算
存储:mysql / 集合 本质是存储东西的;
计算都交给流去计算!
package com.kuang.juc.stream; import java.util.Arrays; import java.util.List; /** * 题目计算:一分钟内完成此题,只能用一行代码实现! * 现在有5个用户!筛选: * 1、ID 必须为偶数 * 2、年龄必须大于23岁 * 3、用户名转换为大写字母 * 4、用户名字母倒着排序 * 5、只输出一个用户的名字! * 6、输出该用户所有信息! */ public class Test { public static void main(String[] args) { User u1 = new User(1, "a", 21); User u2 = new User(2, "b", 22); User u3 = new User(3, "c", 23); User u4 = new User(4, "d", 24); User u5 = new User(6, "e", 25); // 转换为集合 , 集合就是管存储的 List<User> list = Arrays.asList(u1, u2, u3, u4, u5); // 计算交给stream流 // lambda表达式、链式编程、函数式接口、Stream流式计算!这道题中显示的淋漓尽致! // 5、只输出一个用户的名字! // list.stream() // .filter((u)->{return u.getId()%2 == 0;}) // .filter((u)->{return u.getAge()>23;}) // .map((u)->{return u.getName().toUpperCase();}) // .sorted((uu1,uu2)->{return uu2.compareTo(uu1);}) // .limit(1) // .forEach(System.out::println); // 6、输出该用户所有信息! list.stream() .filter(u->{return u.getId()%2 == 0;}) .filter(u->{return u.getAge()>23;}) .map(u->{ u.setName(u.getName().toUpperCase()); return u; }) .sorted((uu1,uu2)->{return uu2.getName().compareTo(uu1.getName());}) .limit(1) .forEach(System.out::println); } }
查看5的输出结果:
查看6的输出结果:
分支合并
什么是ForkJoin
ForkJoin 在 JDK 1.7 ,4, 并行执行任务!提高效率,大数据量!
大数据 :Mapreduce 将大任务拆分成一些些小的任务
Forkjoin特点: 工作窃取
这个里面维护的都是 双端队列 !
ForkJoin
package com.kuang.juc.forkjoin; import java.util.concurrent.RecursiveTask; /** * 求和计算的任务! * 3000 6000 (ForkJoin) 9000 (Stream并行流) * 如何使用forkjoin * 1、ForkJoinPool 通过他来执行 * 2、计算任务 ForkJoinPool.execute(ForkJoinTask task) * 3、计算类要 继承 ForkJoinTask */ public class ForkJoinDemo extends RecursiveTask<Long> { private Long start; // 0 private Long end; // 10_0000_0000 // 临界值 private Long temp = 10000L; public ForkJoinDemo(Long start, Long end) { this.start = start; this.end = end; } // 计算方法 @Override protected Long compute() { if (end-start<temp){ Long sum = 0L; for (Long i = start; i < end; i++) { sum += i; } return sum; }else { //forkjoin 递归 long middle = (start + end)/2; ForkJoinDemo task1 = new ForkJoinDemo(start, middle); task1.fork(); // 拆分任务。把任务压入线程队列 ForkJoinDemo task2 = new ForkJoinDemo(middle+1, end); task2.fork(); // 拆分任务。把任务压入线程队列 return task1.join() + task2.join(); } } }
测试类:
package com.kuang.juc.forkjoin; import java.util.concurrent.ExecutionException; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.ForkJoinTask; import java.util.stream.IntStream; import java.util.stream.LongStream; // 3000 6000 (ForkJoin) 9000 (Stream并行流) public class Test { public static void main(String[] args) throws ExecutionException, InterruptedException { // test1(); //时间: 5417 // test2(); //时间: 3667 // test3(); //时间: 182 } // 普通程序员 public static void test1(){ Long sum = 0L; long startTime = System.currentTimeMillis(); for (Long i = 1L; i < 10_0000_0000L; i++) { sum += i; } long endTime = System.currentTimeMillis(); System.out.println("sum=>"+sum+"时间: "+(endTime-startTime)); } // ForkJoin public static void test2() throws ExecutionException, InterruptedException { long startTime = System.currentTimeMillis(); ForkJoinPool forkJoinPool = new ForkJoinPool(); ForkJoinTask<Long> task = new ForkJoinDemo(0L, 10_0000_0000L); //具体的任务 ForkJoinTask ForkJoinTask<Long> submit = forkJoinPool.submit(task); Long sum = submit.get(); long endTime = System.currentTimeMillis(); System.out.println("sum=>"+sum+"时间: "+(endTime-startTime)); } // Stream并行流 public static void test3(){ long startTime = System.currentTimeMillis(); long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);//(] long endTime = System.currentTimeMillis(); System.out.println("sum=>"+sum+"时间: "+(endTime-startTime)); } }
ForkJoin使用场景: 大数据量下使用! 小数据量没必要!
Future 设计的初衷 : 对未来的某个事件的结果进行建模
package com.kuang.juc.future; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; /** * 异步调用 Ajax * // 异步执行 * // 成功回调 * // 失败回调 * */ public class Demo01 { public static void main(String[] args) throws ExecutionException, InterruptedException { // 没有返回值的异步回调 // CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{ //执行异步任务,没有返回值 // try { // TimeUnit.SECONDS.sleep(2); // } catch (InterruptedException e) { // e.printStackTrace(); // } // System.out.println(Thread.currentThread().getName()+"runAsync=>Void"); // }); // // System.out.println("0V0"); // System.out.println(completableFuture.get()); // 有返回值的异步回调 // ajax 成功和失败的回调 // 返回的是错误信息 CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> { //执行异步任务,有返回值 System.out.println(Thread.currentThread().getName()+"=>supplyAsync=>String"); int i = 10/0; return "以雷霆击碎黑暗"; }); // 编译成功! System.out.println(completableFuture.whenComplete((t, u) -> { System.out.println("t=>" + t); // t代表正常的返回结果 System.out.println("u=>" + u); // u代表错误信息 }).exceptionally((e) -> { //编译失败 System.out.println(e.getMessage()); return "二杯"; }).get()); } }
查看结果:没有异常,正确的输出结果:
若程序中有异常,再次运行:
请你谈谈你对Volatile的理解
Volatile 是Java 虚拟机提供的轻量级的同步机制
1、保证可见性
2、不保证原子性
3、禁止指令重排
什么是JMM
JMM:是Java内存模型,不存在的东西,是一种概念、模型!
关于JMM的一些同步的约定:
1、线程解锁前,必须把线程的共享变量立刻刷回主存
2、线程加锁前,必须读取主存中的最新值到自己的工作内存中
3、加锁和解锁是同一把锁
线程 工作内存 主内存
JMM中8种操作
内存交互操作有8种,虚拟机实现必须保证每一个操作都是原子的,不可在分的(对于double和long类型的变量来说,load、store、read和write操作在某些平台上允许例外)
JMM对这八种指令的使用,制定了如下规则:
问题:程序不知道主内存中的值已经发生了变化
1、保证可见性
package com.kuang.juc.Tvolatile; import java.util.concurrent.TimeUnit; public class JMMDemo { // 不加volatile,程序就会死循环 // 可以保证可见性! private volatile static int num = 0; public static void main(String[] args) { //main new Thread(()->{ //线程1 对主内存的变化是不知道的! while (num==0){ } },"1").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } num = 1; System.out.println(num); } }
2、不保证原子性
原子性:不可分割
线程A在执行任务的时候是不能被打扰的,要么同时成功,要么同时失败!
package com.kuang.juc.Tvolatile; // 不保证原子性 public class VDemo02 { // volatile 不保证原子性! private volatile static int num = 0; public static void add(){ num++; } public static void main(String[] args) { // 理论上输出应该为 2w for (int i = 1; i <= 20; i++) { new Thread(()->{ for (int j = 0; j < 1000; j++) { add(); } }).start(); } while (Thread.activeCount()>2){ //main gc // 礼让,重新竞争资源! Thread.yield(); } System.out.println(Thread.currentThread().getName()+" "+num); } }
输出不到2w,有问题!
问题:如果不加synchonized和lock,怎么样保证原子性?
使用原子类,来解决原子性问题
package com.kuang.juc.Tvolatile; import java.util.concurrent.atomic.AtomicInteger; // 不保证原子性 public class VDemo02 { private static AtomicInteger num = new AtomicInteger(); //原子类的Integer public static void add(){ num.getAndIncrement(); // AtomicInteger +1 方法 , 底层用的CAS } public static void main(String[] args) { // 理论上输出应该为 2w for (int i = 1; i <= 20; i++) { new Thread(()->{ for (int j = 0; j < 1000; j++) { add(); } }).start(); } while (Thread.activeCount()>2){ //main gc // 礼让,重新竞争资源! Thread.yield(); } System.out.println(Thread.currentThread().getName()+" "+num); } }
查看运行结果:
这些类的底层都直接和操作系统挂钩! 在内存中直接修改值!
Unsafe类是一个很特殊的存在!
指令重排
什么是指令重排:你写的程序计算机并不是按照你写的顺序去执行的!
源代码—> 编译器优化重排–> 指令并行也可能重排–>内存系统也可能出现重排–>执行
处理器在执行指令重排的时候,考虑:数据之间的依赖问题
int x = 1; //1 int y = 2; //2 x = x + 5; //3 y = x * x; //4 //我们所期望的:1234 //程序执行的时候可能会是: 1324 2134 2413 //可不可能是:4123
可能造成影响的结果: 前提 a、b、x、y值默认都为0
线程A | 线程B |
---|---|
x=a | y=b |
b=1 | a=2 |
正常的结果: x=0 , y=0;
如果发生指令重排:
线程A | 线程B |
---|---|
b=1 | a=2 |
x=a | y=b |
指令重排出现了诡异的结果:x=2,y=1;
指令重排理解
加了volatile
关键字就可以避免指令重排!
内存屏障。也就是CPU指令。它的作用:
1、保证特定的操作的执行顺序!
2、可以保证某些变量的内存可见性 (利用这些特性,volatile就实现了可见性)
Volatile是可以保持 可见性,不能保证原子性,由于内存屏障,可以保证避免指令重排的现象的产生
饿汉式、DCL懒汉式,深究!
饿汉式
package com.kuang.juc.single; // 饿汉式单例 public class Hungry { // 可能会浪费空间 private byte[] data1 = new byte[1024*1024]; private byte[] data2 = new byte[1024*1024]; private byte[] data3 = new byte[1024*1024]; private byte[] data4 = new byte[1024*1024]; private Hungry(){ } private final static Hungry HUNGRY = new Hungry(); public static Hungry getInstance(){ return HUNGRY; } }
DCL懒汉式
package com.kuang.juc.single; import com.sun.org.apache.regexp.internal.RE; import java.lang.reflect.Constructor; import java.lang.reflect.Field; // 懒汉式单例模式 public class LazyMan { // 标志位 private static boolean panghu = false; // 加锁,防止反编译 private LazyMan(){ synchronized (LazyMan.class){ if (panghu==false){ panghu = true; }else { throw new RuntimeException("不要试图使用反射破坏异常"); } } } private volatile static LazyMan lazyMan; // 避免指令重排 // 双重检测锁模式的 懒汉式单例 DCL懒汉式 private static LazyMan getInstance(){ if (lazyMan==null){ synchronized (LazyMan.class){ if (lazyMan==null){ lazyMan = new LazyMan(); // 不是原子性操作 /** * 1、分配内存空间 * 2、执行构造方法,初始化对象 * 3、把这个对象指向这个空间 * * 123 * 132 A * B //此时LazgyMan还没有完成构造 */ } } } return lazyMan; } //反射! public static void main(String[] args) throws Exception { // 1、用反射破坏,解决私有构造再加锁 // LazyMan instance1 = LazyMan.getInstance(); // //1、获得反射对象 // Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null); // declaredConstructor.setAccessible(true); //无视私有的构造器 // LazyMan instance2 = declaredConstructor.newInstance(); //通过反射创建对象 // // System.out.println(instance1); // System.out.println(instance2); // 2、直接通过反射连续创建两个对象 解决加标志位解决! // Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null); // declaredConstructor.setAccessible(true); //无视私有的构造器 // LazyMan instance1 = declaredConstructor.newInstance(); //通过反射创建对象 // LazyMan instance2 = declaredConstructor.newInstance(); //通过反射创建对象 // // System.out.println(instance1); // System.out.println(instance2); // 3、如果标志位被人反编译出来,知道了 ,再次破解了 Field panghu = LazyMan.class.getDeclaredField("panghu"); panghu.setAccessible(true); Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null); declaredConstructor.setAccessible(true); //无视私有的构造器 LazyMan instance1 = declaredConstructor.newInstance(); //通过反射创建对象 panghu.set(instance1,false); LazyMan instance2 = declaredConstructor.newInstance(); //通过反射创建对象 System.out.println(instance1); System.out.println(instance2); } } // 双重检查锁模式说明: // 这个方法首先判断变量是否被初始化,没有被初始化,再去获取锁。 // 获取锁之后,再次判断该变量是否被初始化。 // 第二次判断目的在于有可能其他线程获取过锁,已经初始化过变量。第二次检查通过了,才会真正初始化变量。 // 这个方法检查判定两次,并使用锁,所以形象称为双重检查锁模式。
静态内部类
package com.kuang.juc.single; //静态内存类 public class Holder { private Holder(){ } public static Holder getInstance(){ return InnerClass.holder; } public static class InnerClass{ private static final Holder holder = new Holder(); } }
单例不安全,存在反射
枚举登场了
package com.kuang.juc.single; import java.lang.reflect.Constructor; // 枚举本身也是一个class类 public enum EnumSingle { INSTANCE; public EnumSingle getInstance(){ return INSTANCE; } } class Test{ // 试图通过反射破解枚举 public static void main(String[] args) throws Exception { EnumSingle instance1 = EnumSingle.INSTANCE; Constructor<EnumSingle> declaredConstructor = EnumSingle.class.getDeclaredConstructor(String.class,int.class); declaredConstructor.setAccessible(true); EnumSingle instance2 = declaredConstructor.newInstance(); // 出现: java.lang.NoSuchMethodException: com.kuang.juc.single.EnumSingle.<init>() 没有空参构造器 System.out.println(instance1); System.out.println(instance2); } }
并不是无参构造,而是有两个参数的有参构造!
通过反编译发现它骗了我们
用jad.exe进行反编译
**枚举类型的最终反编译源码: ** 发现她有一个两个参数的构造方法,而不是空参构造
package com.kuang.juc.single; public final class EnumSingle extends Enum { public static EnumSingle[] values() { return (EnumSingle[])$VALUES.clone(); } public static EnumSingle valueOf(String name) { return (EnumSingle)Enum.valueOf(com/kuang/juc/single/EnumSingle, name); } private EnumSingle(String s, int i) { super(s, i); } public EnumSingle getInstance() { return INSTANCE; } public static final EnumSingle INSTANCE; private static final EnumSingle $VALUES[]; static { INSTANCE = new EnumSingle("INSTANCE", 0); $VALUES = (new EnumSingle[] { INSTANCE }); } }
再次通过两个参数构造方法,反射试图破解,发现:
什么是CAS
package com.kuang.juc.cas; import java.util.concurrent.atomic.AtomicInteger; public class CASDemo { // CAS compareAndSet: 比较并交换! public static void main(String[] args) { AtomicInteger atomicInteger = new AtomicInteger(2020); // 期望 、 更新 // public final boolean compareAndSet(int expect, int update) // 如果我期望的值达到了,那么就更新,否则就不更新! CAS 是CPU的并发原语!! System.out.println(atomicInteger.compareAndSet(2020, 2021)); System.out.println(atomicInteger.get()); System.out.println(atomicInteger.compareAndSet(2020, 2021)); System.out.println(atomicInteger.get()); } }
Unsafe类
CAS :比较当前工作内存中的值和主内存中的值,如果这个值是期望的,那么执行操作!如果不是就一直循环(底层是自旋锁)
缺点:
1、循环会耗时
2、一次性只能保证一个共享变量的原子性
3、存在ABA问题
CAS : ABA 问题(狸猫换太子)
package com.kuang.juc.cas; import java.util.concurrent.atomic.AtomicInteger; public class CASDemo { // CAS compareAndSet: 比较并交换! public static void main(String[] args) { AtomicInteger atomicInteger = new AtomicInteger(2020); // 对于我们平时写的sql来说: 乐观锁! // 期望 、 更新 // public final boolean compareAndSet(int expect, int update) // 如果我期望的值达到了,那么就更新,否则就不更新! CAS 是CPU的并发原语!! // =====================捣乱的线程=========================== System.out.println(atomicInteger.compareAndSet(2020, 2021)); System.out.println(atomicInteger.get()); System.out.println(atomicInteger.compareAndSet(2021, 2020)); System.out.println(atomicInteger.get()); // =====================期望的线程=========================== System.out.println(atomicInteger.compareAndSet(2020, 6666)); System.out.println(atomicInteger.get()); } }
如何解决这个问题: 使用原子引用
解决ABA问题,引入原子引用! 对应的思想乐观锁!
带版本号的原子操作!(相当于sql中的乐观锁)
注意:
package com.kuang.juc.cas; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicStampedReference; public class CASDemo { // CAS compareAndSet: 比较并交换! public static void main(String[] args) { // 坑 : AtomicStampedReference // 注意: 如果泛型是一个包装类,注意对象的引用问题! // 正常的业务操作的话,这里面比较的都是一个个对象! AtomicStampedReference<Integer> atomicReference = new AtomicStampedReference<Integer>(1,1); new Thread(()->{ int stamp = atomicReference.getStamp(); //获得初始的版本号 1 System.out.println("a1=>"+stamp); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } // version + 1 System.out.println(atomicReference.compareAndSet(1, 2, atomicReference.getStamp(), atomicReference.getStamp()+1)); System.out.println("a2=>"+ atomicReference.getStamp()); System.out.println(atomicReference.compareAndSet(2, 1, atomicReference.getStamp(), atomicReference.getStamp()+1)); System.out.println("a3=>"+ atomicReference.getStamp()); },"a").start(); // 和乐观锁的原理相同 new Thread(()->{ int stamp = atomicReference.getStamp(); //获得最初的版本号 1 System.out.println("b1=>"+stamp); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(atomicReference.compareAndSet(1, 6, stamp, stamp + 1)); System.out.println("b2=>"+atomicReference.getStamp()); },"b").start(); } }
查看结果:
公平锁:非常公平,不能够插队,必须先来后到!
非公平锁:非常不公平,可以插队(默认都是非公平的 3h 3s), 保证效率!
public ReentrantLock() { sync = new NonfairSync(); } public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }
可重入锁(递归锁)
Synchonized版 可重入锁
package com.kuang.juc.lock; import java.util.concurrent.TimeUnit; /** * Synchonized版的可重入锁 */ public class Demo01 { public static void main(String[] args) { Phone phone = new Phone(); new Thread(()->{ phone.sms(); },"A").start(); new Thread(()->{ phone.sms(); },"B").start(); } } class Phone{ public synchronized void sms(){ System.out.println(Thread.currentThread().getName()+" sms"); try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } call(); // 这里也有锁 } public synchronized void call(){ System.out.println(Thread.currentThread().getName()+" call"); } }
lock版 可重入锁
package com.kuang.juc.lock; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class Demo02 { public static void main(String[] args) { Phone2 phone = new Phone2(); new Thread(()->{ phone.sms(); },"A").start(); new Thread(()->{ phone.sms(); },"B").start(); } } class Phone2{ Lock lock = new ReentrantLock(); public void sms(){ lock.lock();// 细节问题 : lock.lock(); lock.unlock(); // lock锁必须配对,否则就会死在里面 try { System.out.println(Thread.currentThread().getName()+" sms"); TimeUnit.SECONDS.sleep(3); call(); // 这里也有锁 } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } public void call(){ lock.lock(); // 细节问题 : lock.lock(); lock.unlock(); try { System.out.println(Thread.currentThread().getName()+" call"); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } }
总结
可重入锁就是说某个线程已经获得了某个锁,可以再次获得锁而不会出现死锁!
什么是自旋锁 spinlock
我们来自己定义一个自旋锁
package com.kuang.juc.lock; import java.util.concurrent.atomic.AtomicReference; /** * 自定义自旋锁 */ public class Spinlock { // int 默认为0 // Thread 默认为null AtomicReference<Thread> atomicReference = new AtomicReference<>(); // 加锁 public void mylock(){ Thread thread = Thread.currentThread(); System.out.println(thread.getName()+"===> mylock"); // 自旋锁 while (!atomicReference.compareAndSet(null,thread)){ } } // 解锁 public void myUnlock(){ Thread thread = Thread.currentThread(); System.out.println(thread.getName()+"===> myUnlock"); // 自旋锁 atomicReference.compareAndSet(thread,null); } }
自定义自旋锁进行测试:
package com.kuang.juc.lock; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class TestSpinlock { public static void main(String[] args) throws InterruptedException { // Lock lock = new ReentrantLock(); // lock.lock(); // lock.unlock(); // 使用自定义的自旋锁,底层实现CAS Spinlock spinlock = new Spinlock(); new Thread(()->{ spinlock.mylock(); try { TimeUnit.SECONDS.sleep(5); } catch (Exception e) { e.printStackTrace(); } finally { spinlock.myUnlock(); } },"T1").start(); TimeUnit.SECONDS.sleep(1); new Thread(()->{ spinlock.mylock(); try { TimeUnit.SECONDS.sleep(3); } catch (Exception e) { e.printStackTrace(); } finally { spinlock.myUnlock(); } },"T2").start(); } }
结果:
说明:
1、上面的T1先拿到锁,拿到锁后,判断Thread对象为空,将对象变为当前线程对象,自己没有自旋
2、T2线程这是进来了,Thread对象不为空,while循环条件为true,T2就一直处于自旋状态
3、T1解锁后,T2发现了 Thread对象重新置为了空,while条件不满足了,自旋结束,T2再进行解锁!
死锁是什么
死锁测试,怎么排除死锁!
package com.kuang.juc.lock; import java.util.concurrent.TimeUnit; public class DeadLockDemo { public static void main(String[] args) { String LockA="A"; String LockB="B"; new Thread(new MyThread(LockA,LockB),"线程a").start(); new Thread(new MyThread(LockB,LockA),"线程b").start(); } } class MyThread implements Runnable{ private String LockA; private String LockB; public MyThread(String lockA, String lockB) { LockA = lockA; LockB = lockB; } @Override public void run() { synchronized (LockA){ System.out.println(Thread.currentThread().getName()+" Lock:" +LockA+"试图获取"+LockB); try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } synchronized (LockB){ System.out.println(Thread.currentThread().getName()+" Lock:" +LockB+"试图获取"+LockA); } } } }
结果:
出现了死锁
原因:
1、常量池中的LockA,LockB都加了锁
2、线程a进来之后给进行赋值,赋值之后,常量池中LockA=A,LockB=B都加了锁,线程a中LockA=A,拿到了A的锁,又试图拿到B的锁
3、线程b进来之后,线程b中LockA=B,拿到了B的锁,又试图拿到A的锁
4、这就出现了死锁!
解决问题
1、使用jps -l
定位进程号
2、使用jstack 进程号
查看死锁的堆栈信息:
面试或者工作中,排查问题:
1、看日志 9
2、看堆栈信息! 1
说明
本课程来自于狂神说java系列课程的JUC:https://www.bilibili.com/video/BV1B7411L7tE,课程优质,欢迎大家去学习。
这个笔记是我上课一笔一笔写出来了