JUC并发编程学习笔记
总的来说还可以，学到一些新知识，但很多是学过的了，深入的部分不多。
线程与进程
进程：一个程序，程序的集合，比如一个音乐播发器，QQ程序等。一个进程往往包含多个线程，且至少包含一个线程。
线程：进程中的实际运作单位。
Java默认有几个线程？
2个，main和GC
Java创建线程的方式？
3种，继承Thread，实现Runnable，实现Callable.
Java真的可以开启线程么？
Java不能自己开启线程，而是通过调用本地方法,调用C++代码开启线程，Java不能直接操作硬件。
并发与并行
并发：多个线程操作同一个资源。模拟多条线程交替执行，并发编程的本质是充分利用CPU资源，多个线程争抢CPU执行时间片段。
并行：CPU多核，多个线程同时执行。
public static void main(String[] args) {
//8 可用的计算资源
System.out.println(Runtime.getRuntime().availableProcessors());
}
注意：Runtime.getRuntime().availableProcessors()返回的是可用的计算资源，而不是CPU物理核心数，对于支持超线程的CPU来说，单个物理处理器相当于拥有两个逻辑处理器，能够同时执行两个线程。
Callable
- 有返回值
- 可以跑出异常
- 重写call()方法
细节
- 同一个Callable被两条线程执行，会只执行一次，结果会被缓存。
- get()方法会阻塞等待线程执行完返回值才继续往下走，一般把获取结果放在最后一行或者用异步通信获取结果。
代码示例
public class CallableTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutureTask<String> futureTask = new FutureTask<>(() -> {
// TimeUnit.SECONDS.sleep(3);
System.out.println("execute call");
return "hello callable";
});
//FutureTask为适配器模式
new Thread(futureTask).start();
//callable只执行一次，结果会被缓存
new Thread(futureTask).start();
//阻塞等待线程执行完返回值才继续往下走，一般把获取结果放在最后一行或者用异步通信获取结果
System.out.println(futureTask.get());
}
/**输出：
* execute call
* hello callable
*/
}
Java线程状态
Java线程一共有6种状态。
public enum State {
//新建
NEW,
//运行中
RUNNABLE,
//阻塞
BLOCKED,
//等待
WAITING,
//超时等待
TIMED_WAITING,
//结束
TERMINATED;
}
wait()和sleep()方法的区别？
- 来着不同的类,wait()属于Object类，而sleep()属于Thread类。
- wait()会释放锁，而sleep()不会释放锁。
- 使用范围不同，wait()在同步代码块中使用，而sleep()可以在任何地方使用。
Lock锁
synchronized锁
synchronized本质是队列加锁。
注意线程是个单独的资源类，使用不用有任何附属的操作。
synchronized代码示例
public class TicketTest {
public static void main(String[] args) {
Ticket ticket = new Ticket(20);
new Thread(() -> {
for (int i = 0; i < 20; i++) {
ticket.sale();
}
}, "Mike").start();
new Thread(() -> {
for (int i = 0; i < 20; i++) {
ticket.sale();
}
}, "Tom").start();
new Thread(() -> {
for (int i = 0; i < 20; i++) {
ticket.sale();
}
}, "Jane").start();
}
}
@Data
@AllArgsConstructor
class Ticket {
private Integer number;
public synchronized void sale() {
if (number <= 0) {
return;
}
System.out.println(Thread.currentThread().getName() + "买到了第" + (number--) + "张票，剩余票数" + number);
}
}
/**
* 输出：要有顺序的输出才行
* Mike买到了第20张票，剩余票数19
* Mike买到了第19张票，剩余票数18
* Mike买到了第18张票，剩余票数17
* Mike买到了第17张票，剩余票数16
* Mike买到了第16张票，剩余票数15
* Tom买到了第15张票，剩余票数14
* Tom买到了第14张票，剩余票数13
* Tom买到了第13张票，剩余票数12
* Tom买到了第12张票，剩余票数11
* Tom买到了第11张票，剩余票数10
* Tom买到了第10张票，剩余票数9
* Tom买到了第9张票，剩余票数8
* Tom买到了第8张票，剩余票数7
* Tom买到了第7张票，剩余票数6
* Tom买到了第6张票，剩余票数5
* Tom买到了第5张票，剩余票数4
* Tom买到了第4张票，剩余票数3
* Tom买到了第3张票，剩余票数2
* Tom买到了第2张票，剩余票数1
* Tom买到了第1张票，剩余票数0
*/
Lock锁示例
public class TicketTest2 {
public static void main(String[] args) {
Ticket2 ticket = new Ticket2(20,new ReentrantLock());
new Thread(() -> {
for (int i = 0; i < 20; i++) {
ticket.sale();
}
}, "Mike").start();
new Thread(() -> {
for (int i = 0; i < 20; i++) {
ticket.sale();
}
}, "Tom").start();
new Thread(() -> {
for (int i = 0; i < 20; i++) {
ticket.sale();
}
}, "Jane").start();
}
}
/**
* Lock使用三部曲
* 1、创建 ReentrantLock
* 2、加锁 lock.lock();
* 3、finally块中释放锁 lock.unlock();
*/
@Data
@AllArgsConstructor
class Ticket2 {
private Integer number;
private Lock lock;
public void sale() {
lock.lock();
try {
if (number <= 0) {
return;
}
System.out.println(Thread.currentThread().getName() + "买到了第" + (number--) + "张票，剩余票数" + number);
} finally {
lock.unlock();
}
}
}
synchronized与Lock的区别?
- synchronized是java内置关键字，Lock是一个Java类。
- synchronized无法判断获取锁的状态，Lock可以判断是否取到了锁。
- synchronized自动加锁释放锁，而Lock需手动加锁释放锁，否则会造成死锁。
- synchronized在假设有两个线程去获取锁时，一个线程获取到锁但阻塞了，第二个线程尝试获取锁时会一直等待；而Lock锁不一定会，因为Lock锁可以使用tryLock方法尝试获取锁。
- synchronized可重入，不可中断，非公平锁；而Lock是可重入，可响应中断，可设置是否公平锁。
读写锁——ReadWriteLock
实现类ReentrantReadWriteLock——读写分离思想，主要用在读多写少的并发场景下
| 操作 | 操作 | 结果 |
|---|---|---|
| 读 | 读 | 共存 |
| 读 | 写 | 不共存，互斥 |
| 写 | 写 | 不共存，互斥 |
- 写锁——独占锁——一次只能被一个线程占有
- 读锁——共享锁——多个线程可以同时占有
//示例桶Lock锁
线程通信问题——生产者消费者问题
步骤
- 判断等待
- 业务逻辑
- 通知其他线程
虚假唤醒
多个线程，即超过2个线程通信时的虚假唤醒问题，这时候不能用if判断，而要用while循环判断。
synchronized版线程通信
public class WaitNotifyTest {
public static void main(String[] args) {
Bowl bowl = new Bowl();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
bowl.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "一乐大叔").start();
new Thread(() -> {
for (int i = 0; i < 5; i++) {
try {
bowl.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "佐助").start();
new Thread(() -> {
for (int i = 0; i < 5; i++) {
try {
bowl.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "鸣人").start();
}
}
class Bowl {
//注意如果用Integer包装类的话则要使用构造函数等方法给number赋初值，否则默认为null会导致错误
private int number;
public synchronized void increment() throws InterruptedException {
//注意多个线程，即超过2个线程通信时的虚假唤醒问题，这时候不能用if判断，而要用while循环判断
/* if (number != 0) {
this.wait();
}*/
while (number != 0) {
this.wait();
}
number++;
System.out.println(Thread.currentThread().getName() + "生产了" + number + "碗拉面");
this.notifyAll();
}
public synchronized void decrement() throws InterruptedException {
while (number == 0) {
this.wait();
}
number--;
System.out.println(Thread.currentThread().getName() + "消费了1碗拉面,还剩" + number);
this.notifyAll();
}
/**输出
* 一乐大叔生产了1碗拉面
* 佐助消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 佐助消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 佐助消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 佐助消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 鸣人消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 佐助消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 鸣人消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 鸣人消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 鸣人消费了1碗拉面,还剩0
* 一乐大叔生产了1碗拉面
* 鸣人消费了1碗拉面,还剩0
*/
}
Lock版线程通信
public class ConditionTest {
public static void main(String[] args) {
BigBowl bowl = new BigBowl();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
bowl.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "一乐大叔").start();
new Thread(() -> {
for (int i = 0; i < 5; i++) {
try {
bowl.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "佐助").start();
new Thread(() -> {
for (int i = 0; i < 5; i++) {
try {
bowl.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "鸣人").start();
}
}
class BigBowl {
//注意如果用Integer包装类的话则要使用构造函数等方法给number赋初值，否则默认为null会导致错误
private int number;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
public void increment() throws InterruptedException {
/**
*
在使用阻塞等待获取锁的方式中，必须在try代码块之外，并且在加锁方法与try代码块之间没有任何可能抛出异常的方法调用，避免加锁成功后，在finally中无法解锁。
说明一：如果在lock方法与try代码块之间的方法调用抛出异常，那么无法解锁，造成其它线程无法成功获取锁。
说明二：如果lock方法在try代码块之内，可能由于其它方法抛出异常，导致在finally代码块中，unlock对未加锁的对象解锁，它会调用AQS的tryRelease方法（取决于具体实现类），抛出IllegalMonitorStateException异常。
说明三：在Lock对象的lock方法实现中可能抛出unchecked异常，产生的后果与说明二相同。 java.concurrent.LockShouldWithTryFinallyRule.rule.desc
Positive example：
Lock lock = new XxxLock();
// ...
lock.lock();
try {
doSomething();
doOthers();
} finally {
lock.unlock();
}
Negative example：
Lock lock = new XxxLock();
// ...
try {
// If an exception is thrown here, the finally block is executed directly
doSomething();
// The finally block executes regardless of whether the lock is successful or not
lock.lock();
doOthers();
} finally {
lock.unlock();
}
*/
lock.lock();
try {
//注意多个线程，即超过2个线程通信时的虚假唤醒问题，这时候不能用if判断，而要用while循环判断
while (number != 0) {
condition.await();
}
number++;
System.out.println(Thread.currentThread().getName() + "生产了" + number + "碗拉面");
condition.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void decrement() throws InterruptedException {
lock.lock();
try {
while (number == 0) {
condition.await();
}
number--;
System.out.println(Thread.currentThread().getName() + "消费了1碗拉面,还剩" + number);
condition.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
Condition——顺序打印ABC
Condition用于精准地通知和唤醒线程。
注意：任何一种新的技术，绝对不是仅仅覆盖了原有的技术，还包含对原有技术的补充和增加优势改善。
顺序打印ABC,一共打印3遍。
代码示例
public class PrintTest {
public static void main(String[] args) {
//Line是线程操作的共同资源
Line line = new Line(1);
new Thread(() -> {
for (int i = 0; i < 3; i++) {
try {
line.printA();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 3; i++) {
try {
line.printB();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 3; i++) {
try {
line.printC();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "C").start();
}
}
class Line {
private int state;
private Lock lock = new ReentrantLock();
private Condition conditionA = lock.newCondition();
private Condition conditionB = lock.newCondition();
private Condition conditionC = lock.newCondition();
public Line(int state) {
this.state = state;
}
public void printA() throws InterruptedException {
//判断等待，业务执行，唤醒其他线程
lock.lock();
try {
//注意多个线程，即超过2个线程通信时的虚假唤醒问题，这时候不能用if判断，而要用while循环判断
while (state != StateEnum.FIRST.getCode()) {
conditionA.await();
}
state = StateEnum.SECOND.getCode();
System.out.print("A");
conditionB.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printB() throws InterruptedException {
lock.lock();
try {
//注意多个线程，即超过2个线程通信时的虚假唤醒问题，这时候不能用if判断，而要用while循环判断
while (state != StateEnum.SECOND.getCode()) {
conditionB.await();
}
state = StateEnum.THIRD.getCode();
System.out.print("B");
conditionC.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printC() throws InterruptedException {
lock.lock();
try {
//注意多个线程，即超过2个线程通信时的虚假唤醒问题，这时候不能用if判断，而要用while循环判断
while (state != StateEnum.THIRD.getCode()) {
conditionC.await();
}
state = StateEnum.FIRST.getCode();
System.out.print("C");
conditionA.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
/**
* 输出
* ABCABCABC
*/
}
public enum StateEnum {
FIRST(1,"FIRST"),
SECOND(2,"SECOND"),
THIRD(3,"THIRD");
private int code;
private String name;
StateEnum(int code, String name) {
this.code = code;
}
public int getCode() {
return code;
}
public String getName() {
return name;
}
}
线程锁执行问题
synchronized锁的对象是方法的调用者，即对象本身，当两个同步方法被调用时，谁先拿到锁先执行。
代码示例
public class OrderTest {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(() -> {
phone.sendSMS();
}).start();
new Thread(() -> {
phone.call();
}).start();
}
}
class Phone {
public synchronized void call() {
System.out.println("call");
}
public synchronized void sendSMS() {
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("sendSMS");
}
/**
* 输出：等待3秒后才输出，虽然同时开启了2个线程调用不同方法，但是因为都是调用了phone对象的同步方法，
* 争抢同一把锁，会有执行的先后顺序
* sendSMS
* call
*/
}
同步方法与普通方法被同时调用，普通方法是不会被同步方法锁阻塞的，会直接执行。
两个对象，两个同步方法被同时调用时，对应两把锁，不会互相争抢锁，各自直接执行。
被static修饰的静态同步方法，锁的是Class，跟创建的对象无关，只有Class这一把锁。
一个是被static修饰的静态同步方法，一个是普通同步方法，因为两个的锁不一样，不会争抢锁，static同步方法锁的是Class，而普通同步方法锁的是调用者对象。
线程安全集合类
CopyOnWriteArrayList
public class ListTest {
public static void main(String[] args) {
//多线程下ArrayList边遍历边修改增加会报ConcurrentModificationException
// List<String> list = new ArrayList<>();
// Vector 是同步集合但效率不高
// List<String> list = new Vector<>();
// 集合工具类同步
// List<String> list = Collections.synchronizedList(new ArrayList<>());
//JUC的读写分离集合类，写入时复制
List<String> list = new CopyOnWriteArrayList<>();
for (int i = 0; i < 10; i++) {
new Thread(()->{
list.add(UUID.randomUUID().toString().substring(0,2));
System.out.println(list);
}).start();
}
}
}
CopyOnWriteArraySet
public class SetTest {
public static void main(String[] args) {
/**
* HashSet本质上是HashMap
* 利于了HashMap的KEY不重复的功能实现元素唯一性，值是固定的PRESENT
* private static final Object PRESENT = new Object();
*/
//多线程下HashSet边遍历边修改增加会报ConcurrentModificationException
// Set<String> set = new HashSet<>();
// 集合工具类同步
// Set<String> set = Collections.synchronizedSet(new HashSet<>());
//JUC的读写分离集合类，写入时复制
Set<String> set = new CopyOnWriteArraySet<>();
for (int i = 0; i < 10; i++) {
new Thread(() -> {
set.add(UUID.randomUUID().toString().substring(0, 2));
System.out.println(set);
}).start();
}
}
}
ConcurrentHashMap
原理
BlockingQueue
阻塞队列，指的是当生产者往队列写入生产或者消费者从队列中获取消费时，如果发现队列是满的或者是空的，则写入或者取出的操作将阻塞着，直到有队列有空间继续写入或者有元素可以取出消费。
写入：如果队列满了，就必须阻塞等待消费
取出：如果队列是空的，就必须阻塞等待生产
阻塞队列使用场景：多线程并发处理，线程池
四组写入取出API
- 抛出异常
- 不会抛出异常，有返回值
- 阻塞等待
- 超时等待
| 方式 | 抛出异常 | 不会抛出异常，有返回值 | 阻塞等待 | 超时等待 |
|---|---|---|---|---|
| 添加 | add() | offer() | put() | offer(E e, long t, TimeUnit t) |
| 移除 | remove() | poll() | take() | poll(long timeout, TimeUnit unit) |
| 检查队首元素 | element() | peek() | - | - |
代码示例
public class BlockingQueueTest {
public static void main(String[] args) throws InterruptedException {
test4();
}
/**
* 抛出异常API
*/
public static void test1(){
ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
//写入成功返回true，内部其实是调用offer(e)方法
System.out.println(queue.add("one"));
System.out.println(queue.add("two"));
System.out.println(queue.add("three"));
//IllegalStateException: Queue full
//当队列满时继续写入会抛IllegalStateException: Queue full 异常
// System.out.println(queue.add("four"));
//检查队首元素，有值返回元素但不移除元素，内部其实是调用 peek()方法
//如果队列为空检查不到队首元素则抛出NoSuchElementException 异常
System.out.println(queue.element());
System.out.println("--------------------------------------");
//取出成功返回元素，内部其实是调用poll()方法
System.out.println(queue.remove());
System.out.println(queue.remove());
System.out.println(queue.remove());
//NoSuchElementException
//当队列为空时继续取出会抛NoSuchElementException 异常
// System.out.println(queue.remove());
}
/**
* 输出：
* true
* true
* true
* one
* --------------------------------------
* one
* two
* three
*/
/**
* 不会抛出异常，有返回值 API
*/
public static void test2(){
ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
//写入成功返回true
System.out.println(queue.offer("one"));
System.out.println(queue.offer("two"));
System.out.println(queue.offer("three"));
//当队列满时继续写入会返回false
System.out.println(queue.offer("four"));
//检查队首元素，有值返回元素但不移除元素
//如果队列为空检查不到队首元素则返回null 空
System.out.println(queue.peek());
System.out.println("--------------------------------------");
//取出成功返回元素
System.out.println(queue.poll());
System.out.println(queue.poll());
System.out.println(queue.poll());
//当队列为空时继续取出会则返回null 空
System.out.println(queue.poll());//null
System.out.println(queue.peek());//null
}
/**
* 输出：
* true
* true
* true
* false
* one
* --------------------------------------
* one
* two
* three
* null
* null
*/
/**
* 阻塞等待 API ，因为阻塞等待所以不需要检查队首元素
*/
public static void test3() throws InterruptedException {
ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
//写入成功返回true
queue.put("one");
queue.put("two");
queue.put("three");
//当队列满时会阻塞等待，直到队列有空闲空间可以继续写入
// queue.put("four");
System.out.println("--------------------------------------");
//取出成功返回元素
System.out.println(queue.take());
System.out.println(queue.take());
System.out.println(queue.take());
//当队列为空时继续取出会阻塞等待，直到队列有元素时继续取出
// System.out.println(queue.take());
}
/**
* --------------------------------------
* one
* two
* three
*/
/**
* 超时等待 API
*/
public static void test4() throws InterruptedException {
ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
//写入成功返回true
System.out.println(queue.offer("one",2, TimeUnit.SECONDS));
System.out.println(queue.offer("two",2, TimeUnit.SECONDS));
System.out.println(queue.offer("three",2, TimeUnit.SECONDS));
//当队列满时继续写入会返回进行超时等待，如果在指定时间后写入不成功返回false
System.out.println(queue.offer("four",2, TimeUnit.SECONDS));
System.out.println("--------------------------------------");
//取出成功返回元素
System.out.println(queue.poll(2, TimeUnit.SECONDS));
System.out.println(queue.poll(2, TimeUnit.SECONDS));
System.out.println(queue.poll(2, TimeUnit.SECONDS));
//当队列为空时继续取出会进行超时等待，如果在指定时间后取出不成功返回null
System.out.println(queue.poll(2, TimeUnit.SECONDS));//null
}
/**输出：
* true
* true
* true
* false
* --------------------------------------
* one
* two
* three
* null
*/
}
SynchronousQueue
SynchronousQueue不存储元素，put写入一个元素后，必须从里面先take取出元素，否则无法再继续put写入元素。
//示例不好待补充
常用辅助类
CountDownLatch
CountDownLatch允许一个或多个线程等待直到其他一组线程都执行完操作后再继续执行的同步辅助类。
代码示例
public class CountDownLatchTest {
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(5);
for (int i = 0; i < 5; i++) {
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"召唤完毕！");
countDownLatch.countDown();
},"火精灵"+i).start();
}
//主线程等待其他线程执行完毕后再继续执行
countDownLatch.await();
System.out.println("绿儿指挥所有火精灵发起进攻！");
}
/**输出：
* 火精灵0召唤完毕！
* 火精灵4召唤完毕！
* 火精灵3召唤完毕！
* 火精灵2召唤完毕！
* 火精灵1召唤完毕！
* 绿儿指挥所有火精灵发起进攻！
*/
}
CyclicBarrier
代码示例
public class CyclicBarrierTest {
public static void main(String[] args) {
//第二个Runnable barrierAction 参数是在CyclicBarrier完成等待后会执行的任务
CyclicBarrier cyclicBarrier = new CyclicBarrier(7, () -> {
System.out.println("集齐7颗龙珠，神龙降世！");
});
for (int i = 1; i <= 7; i++) {
new Thread(() -> {
System.out.println("找到第" + Thread.currentThread().getName() + "颗龙珠了！");
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}, i + "").start();
}
}
/**
* 找到第1颗龙珠了！
* 找到第4颗龙珠了！
* 找到第3颗龙珠了！
* 找到第2颗龙珠了！
* 找到第7颗龙珠了！
* 找到第6颗龙珠了！
* 找到第5颗龙珠了！
* 集齐7颗龙珠，神龙降世！
*/
}
Semaphore
代码示例
public class SemophoreTest {
public static void main(String[] args) {
//信号量，限流
Semaphore semaphore = new Semaphore(3);
for (int i = 1; i <= 8; i++) {
new Thread(()->{
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"抢到了停车位了！");
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
semaphore.release();
System.out.println(Thread.currentThread().getName()+"离开停车位！");
}
},"司机"+i).start();
}
/**输出：
* 司机1抢到了停车位了！
* 司机3抢到了停车位了！
* 司机2抢到了停车位了！
* 司机1离开停车位！
* 司机4抢到了停车位了！
* 司机3离开停车位！
* 司机2离开停车位！
* 司机6抢到了停车位了！
* 司机5抢到了停车位了！
* 司机4离开停车位！
* 司机7抢到了停车位了！
* 司机8抢到了停车位了！
* 司机5离开停车位！
* 司机6离开停车位！
* 司机7离开停车位！
* 司机8离开停车位！
*/
}
}
线程池
线程池：Executors创建线程池的三大方法、7大参数、4种拒绝策略
池化技术
好处:
1、降低资源的消耗
2、提高响应的速度
3、方便管理。
线程复用、可以控制最大并发数、管理线程
Executors创建线程池的三大方法
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
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;
}
线程池运行逻辑参考银行办理业务或者工厂生产时的处理策略
4种拒绝策略
/**
* new ThreadPoolExecutor.AbortPolicy() // 银行满了，还有人进来，不处理这个人的，抛出异常
* new ThreadPoolExecutor.CallerRunsPolicy() // 哪来的去哪里！ 调用方自己执行任务
* new ThreadPoolExecutor.DiscardPolicy() //队列满了，丢掉任务，不会抛出异常！
* new ThreadPoolExecutor.DiscardOldestPolicy() //队列满了，尝试去和最早的竞争，丢弃任务，也不会
抛出异常！
*/
线程池大小设置
- CPU密集型——n+1 ，可以保持CPU的效率最高
- IO密集型——2n
注意获取n的方法是通过代码获取Runtime.getRuntime().availableProcessors()
代码示例
public class ExecutorTest {
public static void main(String[] args) {
//CPU密集型
ThreadPoolExecutor executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors() + 1,
60, TimeUnit.SECONDS, new LinkedBlockingQueue(3), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());
try {
System.out.println(Runtime.getRuntime().availableProcessors());
for (int i = 0; i < 13; i++) {
//使用线程池提交任务执行
executor.execute(() -> {
System.out.println(Thread.currentThread().getName() + "执行完毕！");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//关闭线程池
executor.shutdown();
}
}
/**输出：
* 8
* pool-1-thread-1执行完毕！
* pool-1-thread-3执行完毕！
* pool-1-thread-2执行完毕！
* pool-1-thread-4执行完毕！
* pool-1-thread-5执行完毕！
* pool-1-thread-6执行完毕！
* pool-1-thread-8执行完毕！
* pool-1-thread-6执行完毕！
* main执行完毕！
* pool-1-thread-6执行完毕！
* pool-1-thread-7执行完毕！
* pool-1-thread-9执行完毕！
* pool-1-thread-8执行完毕！
*/
}
ForkJoin
并行执行任务！提高效率。大数据量！ 如果任务拆分地不好时反而不如单线程执行，所以要注意。
大数据：Map Reduce （把大任务拆分为小任务）
ForkJoin 特点：任务拆分和工作窃取
三种任务执行效果对比，代码示例
public class ForkJoinTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
test3();
}
public static void test3() throws ExecutionException, InterruptedException {
long start = System.currentTimeMillis();
MyForkJoinTask myForkJoinTask = new MyForkJoinTask(0L, 10_0000_0000L);
ForkJoinPool forkJoinPool = new ForkJoinPool();
ForkJoinTask<Long> task = forkJoinPool.submit(myForkJoinTask);
Long sum = task.get();
long end = System.currentTimeMillis();
System.out.println("输出结果："+sum+"耗时："+(end-start));
}
//输出结果：500000000500000000耗时：2007
public static void test2(){
long start = System.currentTimeMillis();
long sum = 0;
for (long i = 0; i <= 10_0000_0000L; i++) {
sum+=i;
}
long end = System.currentTimeMillis();
System.out.println("输出结果："+sum+"耗时："+(end-start));
}
//输出结果：500000000500000000耗时：341
public static void test1(){
long start = System.currentTimeMillis();
long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0L, Long::sum);
long end = System.currentTimeMillis();
System.out.println("输出结果："+sum+"耗时："+(end-start));
}
//输出结果：500000000500000000耗时：192
}
//没有返回值的话继承RecursiveAction
public class MyForkJoinTask extends RecursiveTask<Long> {
private Long start;
private Long end;
public MyForkJoinTask(Long start, Long end) {
this.start = start;
this.end = end;
}
// 临界值
private Long temp = 10000L;
@Override
protected Long compute() {
if ((end - start) > temp) {
//拆分任务 递归
Long middle = (start+end)/2;
MyForkJoinTask forkJoinTaskLeft = new MyForkJoinTask(start, middle);
forkJoinTaskLeft.fork();
MyForkJoinTask forkJoinTaskRight = new MyForkJoinTask(middle+1, end);
forkJoinTaskRight.fork();
return forkJoinTaskLeft.join()+forkJoinTaskRight.join();
} else {
Long sum = 0L;
for (long i = start; i <= end; i++) {
sum += i;
}
return sum;
}
}
}
CompletableFuture
步骤
- 异步执行
- 成功回调
- 失败回调
代码示例
public class CompletableFutureTest {
public static void main1(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> completableFuture = new CompletableFuture<>();
//静态方法
CompletableFuture.runAsync(() -> {
System.out.println(Thread.currentThread().getName() + "thenRunAsync");
return;
});
System.out.println(Thread.currentThread().getName() + " end");
}
/**
* main end
* ForkJoinPool.commonPool-worker-1thenRunAsync
*/
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("supplyAsync ok");
//报错
System.out.println(10/0);
return 200;
});
CompletableFuture<Integer> completableFuture = future.whenComplete((u, e) -> {
System.out.println(u);// 正常的返回结果
System.out.println(e);//错误信息
}).exceptionally(e -> {
System.out.println(e.getMessage());
return 500;
});
System.out.println(completableFuture.get());
}
/**
* 没报错：
* supplyAsync ok
* 200
* null
* 200
*
* 报错：
* supplyAsync ok
* null
* java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
* java.lang.ArithmeticException: / by zero
* 500
*/
}
JMM
什么是JMM
JMM是Java内存模型，是一种概念，约定，不是一种存在的东西。
JMM的同步概念
- 线程解锁前必须把共享变量立刻刷会主内存。
- 线程加锁前必须读取主内存中的最新值到工作内存中。
- 加锁和解锁必须是同一把锁。
JMM的8种原子操作
内存交互操作有8种，虚拟机实现必须保证每一个操作都是原子的，不可在分的（对于double和long类型的变量来说，load、store、read和write操作在某些平台上允许例外）
- 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关键字，是虚拟机提供的一种轻量级的同步机制。
- 保证可见性
- 不保证原子性
- 禁止指令重排
保证可见性
即只要我们的共享变量加上volatile关键字修饰，则可以保证该共享变量在各个工作内存中获取到的值时最新的值，通过CPU总线嗅探机制来实现。
不保证原子性
原子性指操作不可再分，不可分割。
即当线程在执行任务是不可被打扰，不可分割的，要么同时成功，要么同时失败。
代码示例
public class AtomicTest2 {
//注意变量命令为int还是Integer在底层的汇编指令操作是不同的
private static volatile int num = 0;
//原子类修饰
// private static volatile AtomicInteger num = new AtomicInteger();
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(10);
for (int i = 0; i < 10; i++) {
new Thread(()->{
for (int j = 0; j < 1000; j++) {
add();
}
latch.countDown();
}).start();
}
latch.await();
System.out.println(num);
}
public static void add(){
num++;
}
/* public static synchronized void add(){
num++;
}
public static void add(){
num.getAndIncrement();
}*/
/**
* 只加volatile修饰
* 9600
* 加synchronized修饰
* 10000
* 更改为原子操作类计算
* 10000
*/
}
反编译class文件查看底层原子操作
反编译命令如下：
D:springboot-demo argetclassescomlaispringbootdemojmm>javap -c AtomicTest2.class
Compiled from "AtomicTest2.java"
public class com.lai.springbootdemo.jmm.AtomicTest2 {
public com.lai.springbootdemo.jmm.AtomicTest2();
Code:
0: aload_0
1: invokespecial #1 // Method java/lang/Object."<init>":()V
4: return
public static void main(java.lang.String[]) throws java.lang.InterruptedException;
Code:
0: new #2 // class java/util/concurrent/CountDownLatch
3: dup
4: bipush 10
6: invokespecial #3 // Method java/util/concurrent/CountDownLatch."<init>":(I)V
9: astore_1
10: iconst_0
11: istore_2
12: iload_2
13: bipush 10
15: if_icmpge 40
18: new #4 // class java/lang/Thread
21: dup
22: aload_1
23: invokedynamic #5, 0 // InvokeDynamic #0:run:(Ljava/util/concurrent/CountDownLatch;)Ljava/lang/Runnable;
28: invokespecial #6 // Method java/lang/Thread."<init>":(Ljava/lang/Runnable;)V
31: invokevirtual #7 // Method java/lang/Thread.start:()V
34: iinc 2, 1
37: goto 12
40: aload_1
41: invokevirtual #8 // Method java/util/concurrent/CountDownLatch.await:()V
44: getstatic #9 // Field java/lang/System.out:Ljava/io/PrintStream;
47: getstatic #10 // Field num:I
50: invokevirtual #11 // Method java/io/PrintStream.println:(I)V
53: return
public static void add();
Code:
0: getstatic #10 // Field num:I
3: iconst_1
4: iadd
5: putstatic #10 // Field num:I
8: return
static {};
Code:
0: iconst_0
1: putstatic #10 // Field num:I
4: return
指令重排
什么是指令重排？
我们写的代码逻辑，先后顺序，计算机并不是按照代码的顺序去执行的，计算机会对我们写的代码进行指令重排，优化执行效率和速度。
源代码--》编译器优化的重排--》指令并行也可能会重排--》内存系统也会重排--》最后才到执行
处理器在进行指令重排时会考虑数据之间依赖性，即happen-before原则。
volatile避免指令重排是通过内存屏障实现的。
CPU内存屏障的作用：
- 保证了特定操作的执行顺序。
- 可以保证某些操作的内存可见性。
CAS
比较并替换。比较当前工作内存中的值和主内存中的值，如果这个值是期望的，也就是相等的，那么就执行替换操作，否则就一直循环自旋。
原子类进行CAS操作是自旋锁，当比较失败后会重新比较，知道成功为止才跳出循环。
缺点
- 循环会耗时
- 一次只能保证一个共享变量的原子性
- ABA问题
Unsafe类
Unsafe类的出现是因为Java无法操作内存，这时候Java类想要操作内存是通过本地方法，调用C++的代码来操作内存，因为C++代码可以操作内存。Unsafe类相当于Java的后门。比较并替换时是获取内存地址中的值，这种直接内存操作效率高。
ABA问题——通过乐观锁解决
代码示例
public class ABATest {
public static void main(String[] args) {
/**
* 注意AtomicStampedReference<Integer>泛型类型比较，如果是包装类，则要注意是否在缓存-128——127之间，超过这个
* 范围的值就不是读取缓存，而是在堆空间中直接创建一个包装类对象，这时候compareAndSet方法里是用==比较，比较的是地址，
* 即是否为同一个对象，会由此导致错误无法执行成功。
*/
//// 正常在业务操作，这里面比较的都是一个个对象
AtomicStampedReference<Integer> stampedReference = new AtomicStampedReference<Integer>(1,1);
new Thread(()->{
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(stampedReference.compareAndSet(1, 2, stampedReference.getStamp(), stampedReference.getStamp() + 1));
System.out.println("A stamp"+stampedReference.getStamp());
System.out.println(stampedReference.compareAndSet(2, 1, stampedReference.getStamp(), stampedReference.getStamp() + 1));
System.out.println("A stamp"+stampedReference.getStamp());
},"A").start();
new Thread(()->{
int stamp = stampedReference.getStamp();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(stampedReference.compareAndSet(1, 21111111, stamp, stamp + 1));
System.out.println("B stamp"+stampedReference.getStamp());
},"B").start();
}
/**
* true
* A stamp2
* true
* A stamp3
* false
* B stamp3
*/
}
注意比较并替换源码是用==比较。
自旋锁
自定义锁，通过自旋实现锁同步——代码示例
public class SpinLockTest {
public static void main(String[] args) {
MySpinLock lock = new MySpinLock();
new Thread(()->{
lock.lock();
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
},"T1").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(()->{
lock.lock();
try {
TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
},"T2").start();
}
/**
* T1 lock
* T2 lock
* T1 unlock
* T2 unlock
*/
}
class MySpinLock {
private AtomicReference<Thread> reference = new AtomicReference<>();
public void lock() {
System.out.println(Thread.currentThread().getName() + " lock");
while (!reference.compareAndSet(null, Thread.currentThread())) {
}
}
public void unlock() {
reference.compareAndSet(Thread.currentThread(), null);
System.out.println(Thread.currentThread().getName() + " unlock");
}
}
公平锁、非公平锁
Lock类通过传参可以实现公平锁、非公平锁。
可重入锁
死锁
//查看进程号
jps -l
//jstack查看线程情况 进程号
jstack pid
扩展
单例模式
枚举自带单例模式，当我们使用反射去破坏单例时，会抛出异常：IllegalArgumentException("Cannot reflectively create enum objects");
代码示例
正常DCL
public class DCLTest {
//私有化构造函数
private DCLTest() {
}
/**
* 我们之所以要对双重检查加锁的单例模式的单例对象加volatile关键字，
* 是因为我们在进行创建单例对象时并不是一个原子操作-》 INSTANCE = new DCLTest();
* 创建一个对象有大概一下3个主要步骤，如果我们没有加volatile关键字修饰禁止指令重排的话，可能拿到的是一个还没有完成构件的对象
* 1. 分配内存空间
* 2、执行构造方法，初始化对象
* 3、把这个对象指向这个空间
* 123 * 132 A * B
* 此时DCLTest还没有完成构造
* */
private static volatile DCLTest INSTANCE;
public static DCLTest getInstance() {
if (INSTANCE == null) {
synchronized (DCLTest.class) {
if (INSTANCE == null) {
INSTANCE = new DCLTest();// 不是一个原子性操作
}
}
}
return INSTANCE;
}
public static void main1(String[] args) {
DCLTest instance1 = DCLTest.getInstance();
DCLTest instance2 = DCLTest.getInstance();
System.out.println(instance1);
System.out.println(instance2);
}
/**
* com.lai.springbootdemo.pattern.DCLTest@a09ee92
* com.lai.springbootdemo.pattern.DCLTest@a09ee92
*/
}
DCL加反射
public class DCLReflectTest {
private static boolean singletonFlag = false;
//私有化构造函数
private DCLReflectTest() {
//在构造器上加约束，避免反射破坏单例
synchronized (DCLReflectTest.class) {
if (singletonFlag == false) {
singletonFlag = true;
} else {
throw new RuntimeException("禁止使用反射破坏单例！");
}
}
}
/* //私有化构造函数
private DCLReflectTest() {
//在构造器上加约束，避免反射破坏单例
synchronized (DCLReflectTest.class) {
if (INSTANCE != null) {
throw new RuntimeException("禁止使用反射破坏单例！");
}
}
}*/
private static volatile DCLReflectTest INSTANCE;
public static DCLReflectTest getInstance() {
if (INSTANCE == null) {
synchronized (DCLReflectTest.class) {
if (INSTANCE == null) {
INSTANCE = new DCLReflectTest();
}
}
}
return INSTANCE;
}
public static void main1(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
DCLReflectTest instance1 = DCLReflectTest.getInstance();
//DCL单例模式被反射破坏
Constructor<DCLReflectTest> constructor = DCLReflectTest.class.getDeclaredConstructor(null);
constructor.setAccessible(true);
DCLReflectTest instance2 = constructor.newInstance();
System.out.println(instance1);
System.out.println(instance2);
}
/**
* 输出：
* com.lai.springbootdemo.pattern.DCLReflectTest@a09ee92
* com.lai.springbootdemo.pattern.DCLReflectTest@30f39991
*/
public static void main2(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
// DCLReflectTest instance1 = DCLReflectTest.getInstance();
//DCL单例模式被反射破坏
Constructor<DCLReflectTest> constructor = DCLReflectTest.class.getDeclaredConstructor(null);
constructor.setAccessible(true);
//当我们全部使用反射创建单例对象时，前面构造函数对INSTANCE是否为空的判断便没有意义，可以继续不断创建单例对象
DCLReflectTest instance2 = constructor.newInstance();
DCLReflectTest instance3 = constructor.newInstance();
System.out.println(instance3);
System.out.println(instance2);
}
/**
* com.lai.springbootdemo.pattern.DCLReflectTest@a09ee92
* com.lai.springbootdemo.pattern.DCLReflectTest@30f39991
*/
public static void main(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException, NoSuchFieldException {
// DCLReflectTest instance1 = DCLReflectTest.getInstance();
//DCL单例模式被反射破坏
Constructor<DCLReflectTest> constructor = DCLReflectTest.class.getDeclaredConstructor(null);
constructor.setAccessible(true);
//当我们全部使用反射创建单例对象时，前面构造函数我们可以设置一个变量或者秘钥控制只有一个线程才能创建成功单例对象
DCLReflectTest instance2 = constructor.newInstance();
//但是当然我们对这个变量继续破坏后又可以破坏了单例模式
//拿到私有域，并设置修改权限为true
Field singletonFlag = DCLReflectTest.class.getDeclaredField("singletonFlag");
singletonFlag.setAccessible(true);
singletonFlag.set(instance2,false);
DCLReflectTest instance3 = constructor.newInstance();
System.out.println(instance3);
System.out.println(instance2);
}
/**
* 破坏私有域前：
* Caused by: java.lang.RuntimeException: 禁止使用反射破坏单例！
*
* 破坏私有域后：
* com.lai.springbootdemo.pattern.DCLReflectTest@30f39991
* com.lai.springbootdemo.pattern.DCLReflectTest@452b3a41
*/
}
枚举单例——不能被反射破坏单例
public class EnumSingletonTest {
public static void main(String[] args) throws IllegalAccessException, InvocationTargetException, InstantiationException, NoSuchMethodException {
System.out.println(SingleEnum.INSTANCE);
// System.out.println(SingleEnum.INSTANCE);
//NoSuchMethodException: com.lai.springbootdemo.pattern.SingleEnum.<init>()
//枚举类不是无参构造函数，而是SingleEnum(Sting ,int)类型的构造函数，通过jad反编译得知
// Constructor<SingleEnum> constructor = SingleEnum.class.getDeclaredConstructor(null);
Constructor<SingleEnum> constructor = SingleEnum.class.getDeclaredConstructor(String.class,int.class);
constructor.setAccessible(true);
//通过反射创建枚举对象时会被限制：IllegalArgumentException: Cannot reflectively create enum objects
constructor.newInstance();
}
/**
* INSTANCE
* INSTANCE
*/
}
enum SingleEnum {
INSTANCE;
}
Runnable没有返回值，效率与Callable相比较低。原因