JUC并发编程
一、什么是JUC
使用到 java.util 工具包、包、分类
二、线程和进程
进程:一个正在运行的程序,QQ.exe Music.exe 程序的集合;
一个进程往往可以包含多个线程,至少包含一个!
Java默认有两个线程:main、GC
对于Java而言:Thread、Runnable、Callable
并发编程的本质:充分利用CPU的资源
wait与sleep区别
-
wait是Object类,sleep是Thread类
-
wait会释放锁,sleep不会释放锁
-
wait必须用在同步代码块中,sleep可以用在任何地方
-
wait不需要捕获异常,sleep必须要捕获异常
三、Lock锁(重点)
传统Synchronized
public class SaleTicketDemo01 {public static void main(String[] args) {Ticket ticket = new Ticket();new Thread(() -> {ticket.sale();}, "张三").start();new Thread(() -> {ticket.sale();}, "李四").start();new Thread(() -> {ticket.sale();}, "王五").start();}
}class Ticket {private int number = 30;public synchronized void sale() {while (true) {if (number > 0) {System.out.println(Thread.currentThread().getName() + "卖出了1张票,剩余:" + (--number));}}}
}
Lock接口
公平锁:十分公平:可以先来后到
非公平锁:十分不公平:可以插队 (默认)
public class SaleTicketDemo02 {public static void main(String[] args) {Ticket1 ticket = new Ticket1();new Thread(() -> {ticket.sale();}, "张三").start();new Thread(() -> {ticket.sale();}, "李四").start();new Thread(() -> {ticket.sale();}, "王五").start();}
}class Ticket1 {private int number = 30;Lock lock = new ReentrantLock();public synchronized void sale() {lock.lock();//加锁try {//业务代码while (true) {if (number > 0) {System.out.println(Thread.currentThread().getName() + "卖出了1张票,剩余:" + (--number));}}} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();//解锁}}
}
Synchronized和Lock区别
- Synchronized是内置的Java关键字;Lock是一个Java类
- Synchronized无法判断获取锁的状态;Lock可以判断是否获取到了锁
- Synchronized会自动释放锁;Lock必须要手动释放锁,如果不释放锁,死锁
- Synchronized如果线程1获得锁,线程2只会等待;Lock锁就不一定会等下去
- Synchronized是可重入锁,不可以中断,非公平;Lock是可重入锁,可以判断锁,非公平(可以自己设置)
- Synchronized适合锁少量的代码同步问题,Lock适合锁大量的同步代码
四、生产者和消费者问题
面试:单例模式、排序算法、生产者和消费者、死锁
生产者和消费者问题Syschronized版
/*** 线程之间的通信问题:生成者和消费者问题* 等待唤醒,通知唤醒* 线程交通执行 A和B操作同一个变量num*/
public class A {public static void main(String[] args) {Data data = new Data();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.increase();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "A").start();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.decrease();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "B").start();}
}//判断等待、业务、通知
class Data {private int num = 0;public synchronized void increase() throws InterruptedException {if (num != 0) {//等待,会自动释放锁this.wait();}num++;System.out.println(Thread.currentThread().getName() + "====>" + num);//通知其他线程this.notifyAll();}public synchronized void decrease() throws InterruptedException {if (num == 0) {this.wait();}num--;System.out.println(Thread.currentThread().getName() + "====>" + num);this.notifyAll();}
}
存在问题:
解决方法:if改为while判断
JUC版的生产者和消费者问题
代码实现:
public class B {public static void main(String[] args) {Data1 data = new Data1();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.increase();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "A").start();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.decrease();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "B").start();}
}//判断等待、业务、通知
class Data1 {private int num = 0;Lock lock = new ReentrantLock();Condition condition = lock.newCondition();public void increase() throws InterruptedException {lock.lock();try {while (num != 0) {//等待condition.await();}num++;System.out.println(Thread.currentThread().getName() + "====>" + num);//通知其他线程condition.signal();} catch (Exception e) {throw new RuntimeException(e);} finally {lock.unlock();}}public void decrease() throws InterruptedException {lock.lock();try {while (num == 0) {condition.await();}num--;System.out.println(Thread.currentThread().getName() + "====>" + num);condition.signal();} catch (Exception e) {throw new RuntimeException(e);} finally {lock.unlock();}}
}
Condition精准的通知和唤醒线程
/*** A执行完调用B,B执行完调用C,C执行完调用A*/
public class C {public static void main(String[] args) {Data2 data = new Data2();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.A();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "A").start();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.B();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "B").start();new Thread(() -> {for (int i = 0; i < 10; i++) {try {data.C();} catch (InterruptedException e) {throw new RuntimeException(e);}}}, "C").start();}
}//判断等待、业务、通知
class Data2 {private int num = 1;private Lock lock = new ReentrantLock();private Condition condition1 = lock.newCondition();private Condition condition2 = lock.newCondition();private Condition condition3 = lock.newCondition();public void A() throws InterruptedException {lock.lock();try {while (num != 1) {//等待condition1.await();}System.out.println(Thread.currentThread().getName() + "====>" + num);num++;//通知其他线程condition2.signal();} catch (Exception e) {throw new RuntimeException(e);} finally {lock.unlock();}}public void B() throws InterruptedException {lock.lock();try {while (num != 2) {condition2.await();}System.out.println(Thread.currentThread().getName() + "====>" + num);num++;condition3.signal();} catch (Exception e) {throw new RuntimeException(e);} finally {lock.unlock();}}public void C() throws InterruptedException {lock.lock();try {while (num != 3) {condition3.await();}System.out.println(Thread.currentThread().getName() + "====>" + num);num = 1;condition1.signal();} catch (Exception e) {throw new RuntimeException(e);} finally {lock.unlock();}}
}
五、八锁现象
哪八种锁?
1、一个对象,俩个同步方法
2、一个对象,俩个同步方法,一个方法延迟
3、两个对象,两个同步方法
4、一个对象,一个同步,一个普通方法
5、一个对象,俩个静态同步方法
6、两个对象,俩个静态同步方法
7、一个对象,一个静态的同步方法,一个同步方法
8、两个对象,一个静态的同步方法,一个同步方法
import java.util.concurrent.TimeUnit;/*** 八锁,就是关于锁的八个问题* 1、标准情况下,两个线程先执行发短信,再打电话*/
public class Test1 {public static void main(String[] args) {Phone phone = new Phone();// 线程A发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B打电话new Thread(()->{phone.call();},"B").start();}
}
class Phone{// 发短信public synchronized void sendSms(){System.out.println("发短信");}// 打电话public synchronized void call(){System.out.println("打电话");}
}
import java.util.concurrent.TimeUnit;/*** 八锁,就是关于锁的八个问题* 2、sendSms方法延时4秒,仍然是先执行发短信,再打电话*/
public class Test1 {public static void main(String[] args) {Phone phone = new Phone();// 线程A发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B打电话new Thread(()->{phone.call();},"B").start();}
}
class Phone{// 发短信public synchronized void sendSms(){// 休息四秒try {TimeUnit.SECONDS.sleep(4);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("发短信");}// 打电话public synchronized void call(){System.out.println("打电话");}
}
import java.util.concurrent.TimeUnit;/**
也可以使用Phone phone = new Phone();Phone phone2 = new Phone();phone在线程A中调用发短信,phone2在线程B中调用打电话,这里锁的是两个不同的调用者,所以他们是两把锁,所以互不影响,因为发短信中有4秒的线程休眠,故而一定是打电话被先调用*/
public class Test2 {public static void main(String[] args) {Phone2 phone = new Phone2();Phone2 phone1 = new Phone2();/*** 被synchronized 修饰的方式和普通方法 先执行sendSms() 还是 hello()* 答案: hello()* 解释:新增加的这个方法没有 synchronized 修饰,不是同步方法,不受锁的影响!*/// 线程A调用phone发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B使用phone1调用打电话new Thread(()->{phone1.call();},"B").start();}
}class Phone2{// 发短信public synchronized void sendSms(){// 休息四秒try {TimeUnit.SECONDS.sleep(4);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("发短信");}// 打电话public synchronized void call(){System.out.println("打电话");}
}
import java.util.concurrent.TimeUnit;/*** 被synchronized 修饰的方法和普通方法 先执行sendSms() 还是 hello()* 答案: hello()* 解释:新增加的这个方法没有 synchronized 修饰,不是同步方法,不受锁的影响!*/
public class Test2 {public static void main(String[] args) {Phone2 phone = new Phone2();// 线程A发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B调用未被synchronized修饰的方法hello()new Thread(()->{phone.hello();},"B").start();}
}class Phone2{// 发短信public synchronized void sendSms(){// 休息四秒try {TimeUnit.SECONDS.sleep(4);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("发短信");}// 打电话public synchronized void call(){System.out.println("打电话");}// say hello 没有被锁 也就是说不是同步方法、不受锁的影响,线程会直接把它来执行public void hello(){System.out.println("hello");}
}
import java.util.concurrent.TimeUnit;/*** 将两个同步方法设置为静态*/
public class Test3 {public static void main(String[] args) {// 线程A发短信new Thread(()->{Phone3.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B打电话new Thread(()->{Phone3.call();},"B").start();}
}class Phone3{/*** 在此处我们将 两个同步方法均设为静态方法,此两个方法在类加载是被加载* 所以,在此2个同步方法加载时,它们被全局唯一的Phone3.class对象加载,因为此Phone3.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("打电话");}
}
import java.util.concurrent.TimeUnit;/***/
public class Test3 {public static void main(String[] args) {// 因为sendSms和call方法均是静态方法,故而Phone3.sendSms()和new Phone3().sendSms()没有区别Phone3 phone = new Phone3();Phone3 phone1 = new Phone3();// 线程A发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B打电话new Thread(()->{phone1.call();},"B").start();}
}class Phone3{// 发短信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("打电话");}
}
import java.util.concurrent.TimeUnit;/***/
public class Test3 {public static void main(String[] args) {// 因为sendSms和call方法均是静态方法,故而Phone3.sendSms()和new Phone3().sendSms()没有区别Phone3 phone = new Phone3();// 线程A发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B打电话new Thread(()->{phone.call();},"B").start();}
}class Phone3{// 发短信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("打电话");}
}
import java.util.concurrent.TimeUnit;/***/
public class Test3 {public static void main(String[] args) {// 因为sendSms和call方法均是静态方法,故而Phone3.sendSms()和new Phone3().sendSms()没有区别Phone3 phone = new Phone3();Phone3 phone1 = new Phone3();// 线程A发短信new Thread(()->{phone.sendSms();},"A").start();// 休息一秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 线程B打电话new Thread(()->{phone1.call();},"B").start();}
}class Phone3{// 发短信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("打电话");}
}
总结
-
new对象时,this调用的是这个对象,是一个具体的对象
-
static,class是唯一的一个模板,即Class对象
-
synchronized锁的对象是方法的调用者
- 普通同步方法的调用者是类的实例(new 出来的对象)
- 静态同步方法的调用者是类的对象(类名.class 对象)
六、集合类不安全
List不安全
public class List {public static void main(String[] args) {/*** 并发下ArrayList是不安全的,没有Synchronized* ArrayList<String> list = new ArrayList<>();* 解决方法:* 1.List<String> list = new Vector<>();* Vector底层有Synchronized* 2.java.util.List<String> list = Collections.synchronizedList(new ArrayList<>());* 3.CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();*/CopyOnWriteArrayList<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 数据结构和 ArrayList 是一致的,底层是个数组,只不过CopyOnWriteArrayList 在对数组进行操作的时候,基本会分四步走:
-
加锁;
-
从原数组中拷贝出新数组;
-
在新数组上进行操作,并把新数组赋值给数组容器;
-
解锁
除了加锁之外,CopyOnWriteArrayList 的底层数组还被 volatile 关键字修饰,意思是一旦数组被修改,其它线程立马能够感知到。
整体上来说,CopyOnWriteArrayList 就是利用锁 + 数组拷贝 + volatile 关键字保证了 List 的线程安全。
Set不安全
public class Set {public static void main(String[] args) {/*** 并发下HashSet是不安全的,没有Synchronized* HashSet<String> set = new HashSet<>();* 解决方法:* 1.java.util.Set<String> set = Collections.synchronizedSet(new HashSet<>());* 2.CopyOnWriteArraySet<String> set = new CopyOnWriteArraySet<>();*/CopyOnWriteArraySet<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底层
Map不安全
public class Map {public static void main(String[] args) {/*** 并发下HashMap是不安全的,没有Synchronized* HashMap<String, String> map = new HashMap<>();* 解决方法:* 1.java.util.Map<String, String> map = Collections.synchronizedMap(new HashMap<>());* 2.ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();*/ConcurrentHashMap<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);}).start();}}
}
七、Callable
Callable可以有返回值,可以抛出异常,call()方法
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(); // 怎么启动CallableMyThread 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> {@Overridepublic Integer call() {System.out.println("call()"); // 会打印几个call
// 耗时的操作return 1024;}
}
细节
- 有缓存
- 结果可能需要等待,会阻塞!
八、常用的辅助类(必会)
8.1 CountDownLatch
//计数器
public class CountDownLatchDemo {public static void main(String[] args) throws InterruptedException {//总数是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("Close");}
}
原理:
countDownLatch.countDown();
数量-1
countDownLatch.await();
等待计数器归零,然后再向下执行
每次有线程调用countDown()数量-1,假设计数器变为0,countDownLatch.await()就会被唤醒,继续向下执行!
8.2 CyclicBarrier
public class CyclicBarrierDemo {public static void main(String[] args) {CyclicBarrier cyclicBarrier = new CyclicBarrier(7, () -> {System.out.println("召唤神龙成功");});for (int i = 1; i <= 7; i++) {int temp = i;new Thread(() -> {System.out.println(Thread.currentThread().getName() + "收集第" + temp + "个龙珠");try {cyclicBarrier.await();} catch (InterruptedException e) {throw new RuntimeException(e);} catch (BrokenBarrierException e) {throw new RuntimeException(e);}}).start();}}
}
8.3 Semaphore
Semaphore:信号量
public class SemaphoreDemo {public static void main(String[] args) {//3个停车位,6辆车Semaphore semaphore = new Semaphore(3);for (int i = 1; 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) {throw new RuntimeException(e);} finally {semaphore.release();}},String.valueOf(i)).start();}}
}
原理:
semaphore.acquire()
获得,如果已经满了,等待被释放为止
semaphore.release()
释放,会将当前的信号量释放+1,然后唤醒等待的线程
作用:
- 多个共享资源互斥的使用
- 并发限流,控制最大的线程数
九、读写锁
/*** 独占锁(写锁) 一次只能被一个线程占有* 共享锁(读锁) 多个线程可以同时占有* ReadWriteLock* 多个线程可以同时读,但只能一个线程写,也不能边写边读*/
public class ReadWriteLockDemo {public static void main(String[] args) {MyCache myCache = new MyCache();for (int i = 1; i <= 5; i++) {int temp = i;new Thread(() -> {myCache.put(temp + "", temp);}, String.valueOf(i)).start();}for (int i = 1; i <= 5; i++) {int temp = i;new Thread(() -> {myCache.get(temp + "");}, String.valueOf(i)).start();}}
}/*** 自定义缓存*/
class MyCache {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() + "写入成功" + key);} catch (Exception e) {throw new RuntimeException(e);} 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() + "读取成功" + key);} catch (Exception e) {throw new RuntimeException(e);} finally {readWriteLock.readLock().unlock();}}
}
十、阻塞队列
10.1 概述
阻塞队列:
阻塞队列使用场景:
- 多线程并发处理
- 线程池
10.2 ArrayBlockingQueue 使用
方式 | 抛出异常 | 不抛出异常 | 阻塞等待 | 超时等待 |
---|---|---|---|---|
添加 | add() | offer() | put() | offer(,) |
移除 | remove() | poll() | take() | poll(,) |
检查队首元素 | element() | peek() |
public class Test {/*** 抛出异常*/public static void test1() {//队列的大小:3ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);//add():在队列添加元素System.out.println(blockingQueue.add("a"));System.out.println(blockingQueue.add("b"));System.out.println(blockingQueue.add("c"));//返回队首元素,若没有队首元素,抛出异常Object element = blockingQueue.element();System.out.println(element);//System.out.println(blockingQueue.add("d")); 抛异常//remove():移除队首元素System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());//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//返回队首元素,若没有队首元素,输出nullObject peek = blockingQueue.peek();System.out.println(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() throws InterruptedException {ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);//put()方法没有返回值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()); 队列为空,取不到元素了,线程等待}/*** 超时等待*/public static void test4() throws InterruptedException {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", 2, TimeUnit.SECONDS)); //等待超过两秒就退出,并返回falseSystem.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());//System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS));等待超过两秒就退出,并返回false}
}
10.3 SynchronousQueue 同步队列
没有容量,进去一个元素,必须等待取出来之后,才能再往里面放一个元素。
/*** 同步队列* 和其他的BlockingQueue不一样,SynchronousQueue不存储元素* put了一个元素,必须从里面先take取出来,否则不能put进去值*/
public class SynchronousQueueTest {public 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) {throw new RuntimeException(e);}}).start();new Thread(() -> {try {TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName() + "take 1");blockingQueue.take();TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName() + "take 2");blockingQueue.take();TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName() + "take 3");blockingQueue.take();} catch (InterruptedException e) {throw new RuntimeException(e);}}).start();}
}
十一、线程池(重点)
池化技术
程序的运行本质:占用系统的资源
池化技术:优化资源的使用。事先准备好一些资源,有需要的时候就拿来用,用完了放回来
线程池的好处:
- 降低资源的消耗
- 提高响应的速度
- 方便管理
线程复用、可以控制最大并发数、管理线程
11.1 三大方法
//Executors工具类、三大方法
public class ExecutorsTest {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 <= 10; i++) {//使用线程池来创建线程threadPool.execute(() -> {System.out.println(Thread.currentThread().getName() + " ok");});}} catch (Exception e) {throw new RuntimeException(e);} finally {//线程池用完,程序结束,关闭线程池threadPool.shutdown();}}
}
11.2 七大参数
源码分析
手动创建一个线程池
public class ThreadPoolExecutorTest {public static void main(String[] args) {ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2,//核心线程池大小5,//最大核心线程池大小3,//超时了,没有人调用就会释放TimeUnit.SECONDS,//超时单位new LinkedBlockingQueue<>(3),//阻塞队列(等待的最大数量)Executors.defaultThreadFactory(),//线程工厂,创建线程的,一般不用修改new ThreadPoolExecutor.DiscardOldestPolicy()//拒绝策略);try {for (int i = 1; i <= 9; i++) {threadPool.execute(() -> {System.out.println(Thread.currentThread().getName() + " ok");});}} catch (Exception e) {throw new RuntimeException(e);} finally {threadPool.shutdown();}}
}
11.3 四大拒绝策略
/*** 四大拒绝策略:* new ThreadPoolExecutor.AbortPolicy() 银行满了,还有人进来,不处理这个人的,抛出异常* new ThreadPoolExecutor.CallerRunsPolicy() 满了,从哪个线程来再回哪个线程,让原线程去处理* new ThreadPoolExecutor.DiscardPolicy() 队列满了,丢掉任务,不会抛出异常* new ThreadPoolExecutor.DiscardOldestPolicy() 队列满了,尝试去和最早的竞争,不会抛出异常*/
11.4 CPU密集型与IO密集型
十二、四大函数式接口(必须掌握)
函数式接口:只有一个方法的接口
12.1 Function函数型接口
/*** Function 函数型接口,有一个输入参数,有一个输出* 只要是函数式接口,都可用lambda表达式简化*/
public class FunctionTest {public static void main(String[] args) {/*Function function = new Function<String, String>() {@Overridepublic String apply(String s) {return s;}};*//*Function function = (str) -> {return str;};*/Function function = str -> {return str;};System.out.println(function.apply("abc"));}
}
12.2 Predicate断定型接口
/*** 断定型接口:有一个输入参数,返回值只能是布尔值*/
public class PredicateTest {public static void main(String[] args) {/*Predicate<String> predicate = new Predicate<String>() {@Overridepublic boolean test(String s) {return s.isEmpty();}};*/Predicate<String> predicate = s -> {return s.isEmpty();};System.out.println(predicate.test("abc"));}
}
12.3 Consumer消费型接口
/*** Consumer 消费型接口,只有输入,没有返回值*/
public class ConsumerTest {public static void main(String[] args) {/*Consumer<String> consumer = new Consumer<String>(){@Overridepublic void accept(String str) {System.out.println(str);}};*/Consumer consumer = str -> {System.out.println(str);};consumer.accept("abc");}
}
10.4 Supplier供给型接口
/*** Supplier 供给型接口,没有参数,只有返回值*/
public class SupplierTest {public static void main(String[] args) {/*Supplier<Integer> supplier = new Supplier<Integer>() {@Overridepublic Integer get() {return 1024;}};*/Supplier<Integer> supplier = () -> {return 1024;};System.out.println(supplier.get());}
}
十三、Stream流式计算
集合、数据库本质就是存储东西的。
计算都应该交给流来操作。
/*** 题目要求:一分钟内完成此题,只能用一行代码实现!* 现在有5个用户!筛选:* 1、ID 必须是偶数* 2、年龄必须大于23岁* 3、用户名转为大写字母* 4、用户名字母倒着排序* 5、只输出一个用户!*/
public class StreamTest {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> users = Arrays.asList(u1, u2, u3, u4, u5);users.stream().filter(user -> {return user.getId()%2==0;}).filter(user -> {return user.getAge()>23;}).map(user -> {return user.getName().toUpperCase();}).sorted(((o1, o2) -> {return o2.compareTo(o1);})).limit(1).forEach(System.out::println);}
}
十四、ForkJoin
大数据:Map Reduce(把大任务拆分为小任务)
ForkJoin在JDK1.7,并行执行任务,提高效率,大数据量。
ForkJoin特点:工作窃取
A和B并行执行任务,假设B中的任务执行完了,A中任务还没有执行完,B可以帮助A一起执行剩余的任务。
该图为双端队列,队首和队尾都可以取数据。
/*** 求和计算的任务*/
public class ForkJoinTest extends RecursiveTask<Long> {private Long start;private Long end;//临界值private Long temp = 10000L;public ForkJoinTest(Long start, Long end) {this.start = start;this.end = end;}@Overrideprotected 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; // 中间值ForkJoinTest task1 = new ForkJoinTest(start, middle);task1.fork(); // 拆分任务,把任务压入线程队列ForkJoinTest task2 = new ForkJoinTest(middle+1, end);task2.fork(); // 拆分任务,把任务压入线程队列return task1.join() + task2.join();}}
}
public class Test {public static void main(String[] args) throws Exception {test2();}//普通方法public static void test1() {Long sum = 0L;Long start = System.currentTimeMillis();for (Long i = 1L; i <= 10_0000_0000; i++) {sum += i;}Long end = System.currentTimeMillis();System.out.println("sum=" + sum + "时间:" + (end - start));}/*** 如何使用 forkjoin* 1.forkjoinPool 通过它来执行* 2.计算任务 forkjoinPool.execute(ForkJoinTask task)* 3.计算类要继承 ForkJoinTask*/public static void test2() throws ExecutionException, InterruptedException {long start = System.currentTimeMillis();ForkJoinPool forkJoinPool = new ForkJoinPool();ForkJoinTask<Long> task = new ForkJoinTest(0L, 10_0000_0000L);ForkJoinTask<Long> submit = forkJoinPool.submit(task);//提交任务Long sum = submit.get();long end = System.currentTimeMillis();System.out.println("sum=" + sum + "时间:" + (end - start));}//Streampublic static void test3() {Long start = System.currentTimeMillis();Long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);Long end = System.currentTimeMillis();System.out.println("sum=" + sum + "时间:" + (end - start));}
}
十五、JMM
JMM:Java内存模型,是一种约定,实际不存在的东西。
关于JMM的一些同步的约定:
- 线程解锁前,必须把共享变量立刻刷新回主存。
- 线程加锁前,必须读取主存中的最新值到工作内存中。
- 加锁和解锁是同一把锁
8种操作
内存交互操作有8种,虚拟机实现必须保证每一个操作都是原子的,不可再分的
-
lock (锁定):作用于主内存的变量,把一个变量标识为线程独占状态
-
unlock (解锁):作用于主内存的变量,它把一个处于锁定状态的变量释放出来,释放后的变量才可以被其他线程锁定
-
read (读取):作用于主内存变量,它把一个变量的值从主内存传输到线程的工作内存中,以便随后的load动作使用
-
load (载入):作用于工作内存的变量,它把read操作从主存中变量放入工作内存中
-
use (使用):作用于工作内存中的变量,它把工作内存中的变量传输给执行引擎,每当虚拟机遇到一个需要使用到变量的值,就会使用到这个指令
-
assign (赋值):作用于工作内存中的变量,它把一个从执行引擎中接受到的值放入工作内存的变量副本中
-
store (存储):作用于主内存中的变量,它把一个从工作内存中一个变量的值传送到主内存中,以便后续的write使用
-
write (写入):作用于主内存中的变量,它把store操作从工作内存中得到的变量的值放入主内存的变量中
JMM对这八种指令的使用,制定了如下规则:
-
不允许read和load、store和write操作之一单独出现。即使用了read必须load,使用了store必须write
-
不允许线程丢弃他最近的assign操作,即工作变量的数据改变了之后,必须告知主存
-
不允许一个线程将没有assign的数据从工作内存同步回主内存
-
一个新的变量必须在主内存中诞生,不允许工作内存直接使用一个未被初始化的变量。就是对变量实施use、store操作之前,必须经过assign和load操作
-
一个变量同一时间只有一个线程能对其进行lock。多次lock后,必须执行相同次数的unlock才能解锁
-
如果对一个变量进行lock操作,会清空所有工作内存中此变量的值,在执行引擎使用这个变量前,必须重新load或assign操作初始化变量的值
-
如果一个变量没有被lock,就不能对其进行unlock操作。也不能unlock一个被其他线程锁住的变量
-
对一个变量进行unlock操作之前,必须把此变量同步回主内存
十六、Volatile
Volatile是Java虚拟机提供的轻量级的同步机制
- 保证可见性
- 不保证原子性
- 禁止指令重排
16.1 保证可见性
public class A {// 不加 volatile 程序就会死循环!// 加 volatile 可以保证可见性private volatile static int num = 0;public static void main(String[] args) { // mainnew Thread(() -> { // 线程 1 对主内存的变化不知道的while (num == 0) {}}).start();try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}num = 1;System.out.println(num);}
}
16.2不保证原子性
原子性:不可分割
线程A在执行任务的时候,不能被打扰的,也不能被分割。要么同时成功,要么同时失败。
public class A {// volatile 不保证原子性private volatile static int num = 0;public static void add() {num++;}public static void main(String[] args) {//理论上num结果应该为 2 万for (int i = 1; i <= 20; i++) {new Thread(() -> {for (int j = 0; j < 1000; j++) {add();}}).start();}while (Thread.activeCount() > 2) { // main gcThread.yield();}System.out.println(Thread.currentThread().getName() + " " + num);}
}
如果不加lock或synchronized,怎么保证原子性?
使用原子类,解决原子性问题
public class A {// volatile 不保证原子性private volatile static AtomicInteger num = new AtomicInteger();public static void add() {num.getAndIncrement();}public static void main(String[] args) {//理论上num结果应该为 2 万for (int i = 1; i <= 20; i++) {new Thread(() -> {for (int j = 0; j < 1000; j++) {add();}}).start();}while (Thread.activeCount() > 2) { // main gcThread.yield();}System.out.println(Thread.currentThread().getName() + " " + num);}
}
这些类的底层都直接和操作系统挂钩
在内存中修改值
Unsafe类是一个很特殊的存在
16.3 禁止指令重排
指令重排
什么是指令重排:你写的程序,计算机并不是按照你写的那样去执行的。
源代码–>编译器优化的重排–> 指令并行也可能会重排–> 内存系统也会重排—> 执行
处理器在进行指令重排的时候,考虑:数据之间的依赖性!
int x = 1; // 1
int y = 2; // 2
x = x + 5; // 3
y = x * x; // 4
我们所期望的:1234 但是可能执行的时候回变成 2134 1324
假设下表a b x y 默认都是0
正常的结果: x = 0;y = 0;但是可能由于指令重排
指令重排导致的诡异结果: x = 2;y = 1;
内存屏障
volatile有内存屏障,可避免指令重排
内存屏障:
- 保证特定的操作的执行顺序
- 可以保证某些变量的内存可见性(利用这些特性volatile实现了可见性)
Volatile是可以保证可见性,不能保证原子性,由于内存屏障,可以保证避免指令重排的现象产生。
十七、单例模式
饿汉式
// 饿汉式单例
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 懒汉式
// 懒汉式单例
// 道高一尺,魔高一丈!
public class LazyMan {private static boolean qinjiang = false;private LazyMan() {synchronized (LazyMan.class) {if (qinjiang == false) {qinjiang = true;} else {throw new RuntimeException("不要试图使用反射破坏异常");}}}private volatile static LazyMan lazyMan;// 双重检测锁模式的 懒汉式单例 DCL懒汉式public static LazyMan getInstance() {if (lazyMan == null) {synchronized (LazyMan.class) {if (lazyMan == null) {lazyMan = new LazyMan(); // 不是一个原子性操作}}}return lazyMan;}// 反射!public static void main(String[] args) throws Exception {
// LazyMan instance = LazyMan.getInstance();Field qinjiang = LazyMan.class.getDeclaredField("qinjiang");qinjiang.setAccessible(true);Constructor<LazyMan> declaredConstructor =LazyMan.class.getDeclaredConstructor(null);declaredConstructor.setAccessible(true);LazyMan instance = declaredConstructor.newInstance();qinjiang.set(instance, false);LazyMan instance2 = declaredConstructor.newInstance();System.out.println(instance);System.out.println(instance2);}
}
/*** 1. 分配内存空间* 2、执行构造方法,初始化对象* 3、把这个对象指向这个空间* <p>* 123* 132 A* B // 此时lazyMan还没有完成构造*/
静态内部类
// 静态内部类
public class Holder {private Holder(){}public static Holder getInstace(){return InnerClass.HOLDER;}public static class InnerClass{private static final Holder HOLDER = new Holder();}
}
枚举
// enum 是一个什么? 本身也是一个Class类
public enum EnumSingle {INSTANCE;public EnumSingle getInstance() {return INSTANCE;}
}class Test {public static void main(String[] args) throws NoSuchMethodException,IllegalAccessException, InvocationTargetException, InstantiationException {EnumSingle instance1 = EnumSingle.INSTANCE;Constructor<EnumSingle> declaredConstructor =EnumSingle.class.getDeclaredConstructor(String.class, int.class);declaredConstructor.setAccessible(true);EnumSingle instance2 = declaredConstructor.newInstance();
// NoSuchMethodException: com.kuang.single.EnumSingle.<init>()System.out.println(instance1);System.out.println(instance2);}
}
枚举类型的最终反编译源码:
package com.kuang.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 / 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
CompareAndSet
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());atomicInteger.getAndIncrement()System.out.println(atomicInteger.compareAndSet(2020, 2021));System.out.println(atomicInteger.get());}
}
Unsafe类
CAS:比较当前工作内存中的值和主内存中的值,如果这个值是期望的,那么执行操作,如果不是则一直循环。
缺点:
- 循环会耗时
- 一次性只能保证一个共享变量的原子性
- ABA问题
ABA问题
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(2021, 2020));System.out.println(atomicInteger.get());
// ============== 期望的线程 ==================System.out.println(atomicInteger.compareAndSet(2020, 6666));System.out.println(atomicInteger.get());}
}
十九、各种锁的理解
19.1 公平锁、非公平锁
公平锁:非常公平,不能插队,必须先来后到
非公平锁:非常不公平,可以插队(默认都是非公平)
public ReentrantLock() {sync = new NonfairSync();
}public ReentrantLock(boolean fair) {sync = fair ? new FairSync() : new NonfairSync();
}
19.2 可重入锁
Synchronized
public class demo1 {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");call(); // 这里也有锁}public synchronized void call(){System.out.println(Thread.currentThread().getName() + "call");}
}
Lock
public class demo2 {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 锁必须配对,否则就会死在里面lock.lock();try {System.out.println(Thread.currentThread().getName() + "sms");call(); // 这里也有锁} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();lock.unlock();}}public void call(){lock.lock();try {System.out.println(Thread.currentThread().getName() + "call");} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}
}
19.3 自旋锁
/*** 自旋锁*/
public class SpinlockDemo {// int 0
// Thread nullAtomicReference<Thread> atomicReference = new AtomicReference<>();// 加锁public void myLock() {Thread thread = Thread.currentThread();System.out.println(Thread.currentThread().getName() + "==> mylock");
// 自旋锁while (!atomicReference.compareAndSet(null, thread)) {}}// 解锁
// 加锁public void myUnLock() {Thread thread = Thread.currentThread();System.out.println(Thread.currentThread().getName() + "==> myUnlock");atomicReference.compareAndSet(thread, null);}
}
public class TestSpinLock {public static void main(String[] args) throws InterruptedException {
// ReentrantLock reentrantLock = new ReentrantLock();
// reentrantLock.lock();
// reentrantLock.unlock();
// 底层使用的自旋锁CASSpinlockDemo lock = new SpinlockDemo();new Thread(() -> {lock.myLock();try {TimeUnit.SECONDS.sleep(5);} catch (Exception e) {e.printStackTrace();} finally {lock.myUnLock();}}, "T1").start();TimeUnit.SECONDS.sleep(1);new Thread(() -> {lock.myLock();try {TimeUnit.SECONDS.sleep(1);} catch (Exception e) {e.printStackTrace();} finally {lock.myUnLock();}}, "T2").start();}
}
19.4 死锁
多个线程都占用了对方的资源,都不肯相让,导致了死锁,在编程中一定要避免死锁的发生。
public class DeadLock_ {public static void main(String[] args) {//模拟死锁现象DeadLockDemo A = new DeadLockDemo(true);A.setName("A线程");DeadLockDemo B = new DeadLockDemo(false);B.setName("B线程");A.start();B.start();}
}//线程
class DeadLockDemo extends Thread {static Object o1 = new Object();// 保证多线程,共享一个对象,这里使用staticstatic Object o2 = new Object();boolean flag;public DeadLockDemo(boolean flag) {//构造器this.flag = flag;}@Overridepublic void run() {//下面业务逻辑的分析//1. 如果flag 为 T, 线程A 就会先得到/持有 o1 对象锁, 然后尝试去获取 o2 对象锁//2. 如果线程A 得不到 o2 对象锁,就会Blocked//3. 如果flag 为 F, 线程B 就会先得到/持有 o2 对象锁, 然后尝试去获取 o1 对象锁//4. 如果线程B 得不到 o1 对象锁,就会Blockedif (flag) {synchronized (o1) {//对象互斥锁, 下面就是同步代码System.out.println(Thread.currentThread().getName() + " 进入1");synchronized (o2) { // 这里获得li对象的监视权System.out.println(Thread.currentThread().getName() + " 进入2");}}} else {synchronized (o2) {System.out.println(Thread.currentThread().getName() + " 进入3");synchronized (o1) { // 这里获得li对象的监视权System.out.println(Thread.currentThread().getName() + " 进入4");}}}}
}