- 1.知识点回顾及延伸
- 2.生产者消费者问题
- 3. 八锁问题
- 4.集合类线程不安全解决
- 5.Callable再理解
- 6.三大常用辅助类
- 6.1、 CountDownLatch
- 6.2、CyclicBarrier
- 6.3、Semaphore
- 7.读写锁
- 8.阻塞队列
- 8.1、了解阻塞队列
- 8.2、阻塞队列核心方法
- 8.3、同步队列SynchronousQueue
- 9.线程池
- 9.1、三大方法（不建议）
- 9.2、七大参数
- 9.3、四种拒绝策略
- 9.4、最大线程数设定方式
- 10.四大函数式接口
- 11.Stream流式计算
- 12.ForkJoin
- 13.异步回调Future
- 14.JMM
- 15.Volatile
- 16.原子类
- 17.CAS
- 18.各种锁
1.知识点回顾及延伸
- java其实无法真正的开启一个线程，而是通过native关键字去调用本地方法！！
- 并发和并行：
- 并发：多个线程通过快速交替，模拟出多线程同时执行，单核CPU使用
- 并行：真正意义上的多线程同时进行，多核CPU才能完成
public class model01 {
public static void main(String[] args) {
//获取CPU核数
System.out.println(Runtime.getRuntime().availableProcessors());
}
}
- 研究并发编程的意义在于充分利用CPU资源！！
- 代码中线程的六种状态：
//新生
NEW,
//运行
RUNNABLE,
//阻塞
BLOCKED
//等待
WAITING,
//超时等待
TIMED_WAITING,
//终止
TERMINATED;
- sleep和wait的区别：
- sleep 来自 Thread 类，和 wait 来自 Object 类。
- 最主要是sleep方法没有释放锁，而wait方法释放了锁，使得其他线程可以使用同步控制块或者方法。
- wait，notify和 notifyAll 只能在同步控制方法或者同步控制块里面使用，而 sleep 可以在任何地方使用(使用范围)
- sleep ，wait 必须捕获异常，notify 和 notifyAll 不需要捕获异常
- 公平锁以及非公平锁
- 公平锁 定义：多个线程按照先到先得的策略获取锁。 优点：所有线程都有机会获得锁，不会饿死 缺点：由于所有线程都会经历阻塞态，因此唤醒阻塞线程的开销会很大。
- 非公平锁（默认） 定义：所有的线程拼运气，谁运气好，谁就获取到锁 优点：可以减少CPU唤醒线程的开销，整体的吞吐效率会高 缺点：可能会有线程长时间甚至永远获取不到锁，导致饿死。
- Lock的实现类：
- ReentrantLock（常用）：可重入锁
- ReentrantReadWriteLock.ReadLock：读锁，
- ReentrantReadWriteLock.WriteLock ：写锁
- lock和synchronized的区别：
- lock是一个接口，而synchronized是java的一个关键字；
- 异常是否释放锁：
synchronized在发生异常时候会自动释放占有的锁，因此不会出现死锁；而lock发生异常时候，不会主动释放占有的锁，必须手动unlock来释放锁，可能引起死锁的发生。 - 是否响应中断
lock等待锁过程中可以用interrupt来中断等待，而synchronized只能等待锁的释放，不能响应中断； - 是否能够获取获取锁的状态
Lock可以通过trylock来知道有没有获取锁，而synchronized不能； - Lock可以提高多个线程进行读操作的效率。（可以通过readwritelock实现读写分离）
- 在性能上来说，如果竞争资源不激烈，两者的性能是差不多的，而当竞争资源非常激烈时（即有大量线程同时竞争），此时Lock的性能要远远优于synchronized。所以说，在具体使用时要根据适当情况选择。
2.生产者消费者问题
- 老版的synchronized +wait+notify
- 新版的Lock+condition.await+condition.signall，用法与老版的基本没有什么差别。
- 优势：使用condition可以进行精准的唤醒线程操作：
public static void main(String[] args) {
Data3 data = new Data3();
new Thread(()->{
for (int i = 0; i {
for (int i = 0; i {
for (int i = 0; i
虚假唤醒：该问题常常出现在生产者消费者问题之中，出现这种现象的原因就是通过if去设定一个判断条件，当满足该条件的时候线程会进入wait状态，但是当有多个线程被唤醒后去执行操作时，只有一个线程进入了if语句执行了wait方法，而其他的线程没有执行到wait方法
-
解决：将if改成while即可！
-
八锁问题
-
现有两个普通同步方法a，b，此时通过两个线程A，B调用同一个对象，a方法先被调用，谁先执行？
- 答：先拿到锁的先执行，也就是A线程先执行
-
如果使线程Asleep一段时间，谁先执行?
-
答：还是A，因为A已经先拿到了锁，sleep并不会释放锁
- 新增普通方法c，B线程调用c方法，谁先执行？
- B线程执行，因为A线程拿到锁好会sleep一段时间，此时B线程直接调用c方法，不需要锁就能执行
- 现在还是两个同步方法a，b，此时A，B线程个调用一个对象，谁先执行？
- 谁的sleep的时间短，也就是先苏醒谁就先执行，因为不存在锁的竞争问题
- 现将普通的同步方法a，b使用static修饰，A，B线程调用一个对象，谁先执行？
- 还是先拿到锁的先执行，如果用static修饰同步方法的话，那么锁的对象就class模板而不是调用对象了
- 现在有一个普通的同步方法，一个静态的同步方法，A，B线程调用同一个对象，那么谁先执行？
- 谁的sleep的时间短，也就是先苏醒谁就先执行，因为使用static的同步方法锁的是class模板，而没有static修饰的方法锁的对象是对象，此时不存在锁的竞争
- 现在有一个普通的同步方法，一个静态的同步方法，A，B线程调用同两个对象，那么谁先执行？
- 同7，因为也不存在锁的竞争
- 现将普通的同步方法a，b使用static修饰，A，B线程调用两个对象，谁先执行？
- 谁先拿到锁谁就先执行
总结：
- 非静态方法的锁默认为this,静态方法的锁对应的Class实例
- 在面对锁的问题的时候只有两种情况，一种是存在锁的竞争，此时是先拿到锁的先执行，一种是不存在锁的竞争，此时是看哪个线程沉睡时间短谁就先执行。
4.集合类线程不安全解决
- 如何解决ArrayList线程不安全：
- 使用synchronized同步代码块
- 使用List list=new Vector()，其底层也是使用的synchronized关键字
- 使用Collections.synchronizedList()，参数传一个ArrayList即可
- 使用CopyOnWriteArrayList，当 List 需要被修改的时候，我并不修改原有内容，而是对原有数据进行一次复制，将修改的内容写入副本。写完之后，再将修改完的副本替换原来的数据，这样就可以保证写操作不会影响读操作了。其底层是使用了Lock进行了加锁的操作，而其数组同时用volatile关键字修饰。
- 如何解决HashSet线程不安全：
- 使用synchronized同步代码块
- 使用Collections.synchronizedSet()，参数传一个HashSet即可
- 使用CopyOnWriteArraySet
- 如何解决HashMap线程不安全：
- 使用synchronized同步代码块
- 使用Collections.synchronizedMap()，参数传一个HashMap即可
- 使用ConcurrentHashMap
5.Callable再理解
- 特点：
- 有返回值
- 可以抛出异常
- 方法不同，run（）/call（）
- 继承Callable接口的时候需要传入泛型，该泛型也是返回值的类型
- 通过callable创建线程的方式一：
- 实现Callable接口，通过FutureTask包装器来创建Thread线程；
package com.xiaoxuzhu;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
/**
* Description: 实现Callable接口，通过FutureTask包装器来创建Thread线程
* 跟Runnable比，不同点在于它是一个具有返回值的，且会抛出异常
* //用futureTask接收结果
*/
public class ThreadDemo3 implements Callable {
public static void main(String[] args) {
ThreadDemo3 threadDemo03 = new ThreadDemo3();
//1、用futureTask接收结果
FutureTask futureTask = new FutureTask<>(threadDemo03);
new Thread(futureTask).start();
//2、接收线程运算后的结果
try {
//futureTask.get();这个是堵塞性的等待
Integer sum = futureTask.get();
System.out.println("sum="+sum);
System.out.println("-------------------");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
@Override
public Integer call() throws Exception {
int sum = 0;
for (int i = 0; i
注：
+ 实现原理：Thread参数只能接收Runnable接口或者其实现类，而FutureTask<>正好是其实现类，而FutureTask<>可以接收一个callable类型的参数，使用可以通过FutureTask<>这个中间人来使Thread和callable搭上关系
+ futureTask.get();这个是堵塞性的等待，所以如果call方法里面是一个耗时多的部分那么结果就会返回的很慢，所以解决办法一般是把futureTask.get()放到最后或者使用异步通信处理
+ 当有两个线程调用同一个callable时，只会输出一次call方法里卖弄的内容，因为结果会有缓冲，效率高
6.三大常用辅助类
6.1、 CountDownLatch
- 功能：在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待
- 实例：
import java.util.concurrent.CountDownLatch;
public class model02 {
public static void main(String[] args) throws InterruptedException {
// 计数器3个。
CountDownLatch countDownLatch = new CountDownLatch(3);
for (int i = 0; i < 3; ++i) {
new Thread(new Worker(countDownLatch, i)).start();
}
// 等待三个线程都完成
countDownLatch.await();
System.out.println("3个线程全部执行完成");
}
// 搬运工人工作线程工作类。
static class Worker implements Runnable {
private final CountDownLatch countDown;
private final Integer id;
Worker(CountDownLatch countDown, Integer id) {
this.countDown = countDown;
this.id = id;
}
@Override
public void run() {
try {
doWork();
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("第" + id + "个线程执行完成工作");
countDown.countDown();
}
void doWork() {
System.out.println("第" + id + "个线程开始工作");
}
}
}
- 主要方法：
- countDown()：计数器减一
- await()：必须等待计数器归零才能执行后面的操作
- 注：在new CountDownLatch对象时必须传入一个数值当做计数器
6.2、CyclicBarrier
CyclicBarrier在用法上其实跟CountDownLatch十分相似，但是前者功能更加强大，它的屏障可以重置！！
实例：
public class CycleBarrierTest2 {
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
// 将线程A添加到线程池
executorService.submit(new Runnable() {
@Override
public void run() {
try {
System.out.println("线程A：" + Thread.currentThread().getName() + "执行第1步。");
cyclicBarrier.await();
System.out.println("线程A：" + Thread.currentThread().getName() + "执行第2步。");
cyclicBarrier.await();
System.out.println("线程A：" + Thread.currentThread().getName() + "执行第3步。");
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 将线程B添加到线程池
executorService.submit(new Runnable() {
@Override
public void run() {
try {
System.out.println("线程B：" + Thread.currentThread().getName() + "执行第1步。");
cyclicBarrier.await();
System.out.println("线程B：" + Thread.currentThread().getName() + "执行第2步。");
cyclicBarrier.await();
System.out.println("线程B：" + Thread.currentThread().getName() + "执行第3步。");
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 关闭线程池
executorService.shutdown();
}
}
注：CyclicBarrier的特性就在于它的计数器是可以重置的，这也就有了上图那种多屏障的情况，虽然第一次调用await使得计数器等于0屏障1失效，但是后续如果继续调用await，屏障还能继续使用，这也就是计数器重置的好处。而CountDownLatch第一次屏障失效后，就结束了。
总结：
- 首先new 一个CyclicBarrier对象，同时传入一个数值，表示数值，同时还可以传入一个Runnable接口类型的对象，表示当达到条件时执行该runnable的方法
- 原理：当一个线程开始执行后调用 cyclicBarrier.await()方法，然后该线程进入等待，同时 cyclicBarrier对象的数值加一，当加到传入的数值大小后，立即执行CyclicBarrier对象的方法，同时因为cyclicBarrier.await()等待的线程也会被唤醒继续执行
- 使用CyclicBarrier的好处是它的计数器是可以重置的，，当第一次达到屏障时，再次调用cyclicBarrier.await()方法就会出现计数
6.3、Semaphore
限流！！！
- 主要方法：
- Semaphore(int permits):构造方法，创建具有给定许可数的计数信号量并设置为非公平信号量。
- Semaphore(int permits,boolean fair):构造方法，当fair等于true时，创建具有给定许可数的计数信号量并设置为公平信号量。
- void acquire():从此信号量获取一个许可前线程将一直阻塞。相当于一辆车占了一个车位。
- void acquire(int n):从此信号量获取给定数目许可，在提供这些许可前一直将线程阻塞。比如n=2，就相当于一辆车占了两个车位。
- void release():释放一个许可，将其返回给信号量。就如同车开走返回一个车位。
- void release(int n):释放n个许可。
- int availablePermits()：当前可用的许可数。
- 实例：
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
public class model03 {
public static void main(String[] args) {
//限流，相当于只有三个停车位
Semaphore semaphore=new Semaphore(3);
for (int i = 0; i < 6; i++) {
final int temp=i;
new Thread(()->{
try {
//设置一个车位被占用
semaphore.acquire();
System.out.println(temp+"线程抢占到车位");
TimeUnit.SECONDS.sleep(1);//相当于sleep2s
System.out.println(temp+"线程离开车位");
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
semaphore.release();
}
},i+"").start();
}
}
}
注：通过该类可以达到一个限流的操作！！
7.读写锁
- 读写锁的创建：
ReadWriteLock readWriteLock=new ReentrantReadWriteLock();
注：ReentrantReadWriteLock是ReadWriteLock接口的唯一实现类
2. 读锁以及写锁：
– 读锁：又叫做共享锁，S锁
Lock readLock=readWriteLock.readLock();
readLock.lock();//加锁
readLock.unlock();//解锁
注：
+ 通过同一个readWriteLock对象获得的读锁一样！！
+ 加锁解锁方式与lock没有区别
– 写锁：有叫做独占锁，排他锁，X锁
Lock writeLock=readWriteLock.writeLock();
writeLock.lock();
writeLock.unlock();
注：与读锁一样
注意：读写锁（ReentrantReadWriteLock）就是读线程和读线程之间不互斥。读读不互斥，读写互斥，写写互斥
8.阻塞队列
拓展：队列家族
8.1、了解阻塞队列
3. 使用场景：多线程并发处理，线程池技术
8.2、阻塞队列核心方法
核心方法：：
- 抛出异常：
public static void main(String[] args) {
BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);//在new对象时需要指定队列大小
//add方法添加元素，返回boolean。当阻塞队列满时，再往阻塞队列add插入元素会抛 java.lang.IllegalStateException: Queue full
System.out.println(blockingQueue.add("a"));
System.out.println(blockingQueue.add("b"));
System.out.println(blockingQueue.add("c"));
// System.out.println(blockingQueue.add("d"));
//element方法返回队列的第一个元素，当阻塞队列为空时，抛出java.util.NoSuchElementException
System.out.println(blockingQueue.element());
//remove方法清空队列第一个元素，并返回该元素。当阻塞为空时，抛出java.util.NoSuchElementException
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
// System.out.println(blockingQueue.element());
System.out.println(blockingQueue.remove());
}
- add方法添加元素到阻塞队列。当阻塞队列满时，再往队列里add插入元素会抛出java.lang.IllegalStateException: Queue full
- remove方法清除阻塞队列的第一个元素。当阻塞队列空时，再从队列里移除元素会抛出java.util.NoSuchElementException
- element方法获取阻塞队列的第一个元素。当阻塞队列空时，再从队列里获取元素会抛出java.util.NoSuchElementException
- 返回特殊值：
BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
//offer方法添加元素，添加成功返回true，添加失败就返回false
System.out.println(blockingQueue.offer("a"));
System.out.println(blockingQueue.offer("b"));
System.out.println(blockingQueue.offer("c"));
System.out.println(blockingQueue.offer("d"));
//peek方法获取队列里的第一个元素，如果队列为空，则返回null
System.out.println(blockingQueue.peek());
//poll方法移除队列里的第一个元素，并返回该元素，如果队列为空，则返回null
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
- offer方法添加元素，添加成功返回true，添加失败就返回false
- peek方法获取队列里的第一个元素，如果队列为空，则返回null
- poll方法移除队列里的第一个元素，并返回该元素，如果队列为空，则返回null
- 一直阻塞：
//一直阻塞
BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
//put方法添加元素，当阻塞队列满时，再往队列里添加元素，队列会一直阻塞线程直到有其他线程take数据或中断退出
blockingQueue.put("a");
blockingQueue.put("b");
blockingQueue.put("c");
// blockingQueue.put("d");
//take方法移除元素，当阻塞队列空时，再从队列里移除元素，队列会一直阻塞线程直到队列有新的元素或者中断退出
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
// System.out.println(blockingQueue.take());
- put方法添加元素。当阻塞队列满时，再往队列里添加元素，队列会一直阻塞线程直到有其他线程take数据或响应中断退出
- take方法移除元素。当阻塞队列空时，再从队列里移除元素，队列会一直阻塞线程直到队列可用
- 超时退出：
//超时退出
BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
//offer方法添加元素，并返回boolean，如果队列为满，阻塞一定时间，然后退出
System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));
System.out.println(blockingQueue.offer("b", 2L, TimeUnit.SECONDS));
System.out.println(blockingQueue.offer("c", 2L, TimeUnit.SECONDS));
System.out.println(blockingQueue.offer("d", 2L, TimeUnit.SECONDS));
//poll获取队列里第一个元素，如果队列为空时，阻塞一定时间，然后退出
System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
- offer方法添加元素，并返回boolean，如果队列为满时，阻塞一定时间，然后退出，第一个参数为添加到队列的对象，第二个为时间的数值大小，第三个为时间单位，使用的是TimeUnit
- poll获取队列里第一个元素，如果队列为空时，阻塞一定时间，然后退出
8.3、同步队列SynchronousQueue
- 介绍：SynchronousQueue是这样一种阻塞队列，其中每个put必须等待一个take，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。不能在同步队列上进行peek，因为仅在试图要取得元素时，该元素才存在，除非另一个线程试图移除某个元素，否则也不能（使用任何方法）添加元素，也不能迭代队列，因为其中没有元素可用于迭代。队列的头是尝试添加到队列中的首个已排队线程元素，如果没有已排队线程，则不添加元素并且头为 null。
对于其他Collection方法（例如 contains），SynchronousQueue作为一个空集合，此队列不允许 null 元素。 - 重点：
- 每次存操作后必须去操作，反之同样
- 同步队列没有容量
- 同步队列无法使用检查方法查看队首元素
- 在new对象时不能传入队列大小
- 使用：同步队列是阻塞队列的实现类，所以它的使用与阻塞队列的方法相通！！
9.线程池
- 定义：线程池其实就是一种多线程处理形式，处理过程中可以将任务添加到队列中，然后在创建线程后自动启动这些任务。
- 好处：
- 降低系统资源消耗
- 提高系统响应速度
- 方便线程并发数的管控
- 通过线程池创建线程：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
//线程池实现
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
//创建线程池
ExecutorService executorService = Executors.newFixedThreadPool(10);
ThreadPool threadPool = new ThreadPool();
for(int i =0;i
注：
– submit方法需要Runnable或者Callable参数，execute需要Runnable参数！！
– 使用完线程池服务需要关闭线程池！！
9.1、三大方法（不建议）
主要是通过Executors工具类创建
- Executors.newSingleThreadExecutor()，只含有单个线程
- Executors.newFixedThreadPool(5)，创建指定线程个数的线程池
- Executors.newCachedThreadPool()，创建可伸缩的线程池
package com.qcby.lock;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author HuangHaiyang
* @date 2020/07/07
* @description: description
* @version: 1.0.0
*/
public class ThreadPoolTest {
public static void main(String[] args) {
//单个线程
ExecutorService threadPool = Executors.newSingleThreadExecutor();
// 创建一个固定的线程池的大小
ExecutorService threadPool1 = Executors.newFixedThreadPool(5);
//可伸缩的，遇强则强，遇弱则弱
ExecutorService threadPool2 = Executors.newCachedThreadPool();
try {
for (int i = 1; i {
// System.out.println(Thread.currentThread().getName()+"-------->OK");
// });
// threadPool1.execute(()->{
// System.out.println(Thread.currentThread().getName()+"-------->OK");
// });
threadPool2.execute(()->{
System.out.println(Thread.currentThread().getName()+"-------->OK");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
threadPool1.shutdown();
threadPool2.shutdown();
}
}
}
9.2、七大参数
通过探究三大方法的底层源码发现：三大方法寝室都是通过返回一个ThreadPoolExecutor来创建线程池的，而ThreadPoolExecutor就正好具有七个参数！！！
public ThreadPoolExecutor(int corePoolSize,//核心线程数
int maximumPoolSize,//最大线程数
long keepAliveTime,//超时时间大小
TimeUnit unit,//超时单位
BlockingQueue workQueue,//阻塞队列
ThreadFactory threadFactory,//线程工厂：创建线程的
RejectedExecutionHandler handler) {//拒绝策略
if (corePoolSize < 0 ||
maximumPoolSize
- corePoolSize：核心线程数，在处理的过来时最大的线程数
- int maximumPoolSize：最大运行的线程数，当核心线程数大小的数目处理不过来时就会开启
- long keepAliveTime：线程释放的超时时间大小
- TimeUnit unit，超时时间单位
- BlockingQueue workQueue：阻塞队列，设定阻塞的线程数大小
- ThreadFactory threadFactory：线程工厂，一般使用默认的即可，通过Executors.defaultThreadFactory()创建
- RejectedExecutionHandler handler：拒绝策略，当阻塞队列满且以及达到最大线程运行数时所采用的拒绝策略
手动创建一个线程池（推荐！！）：
import java.util.concurrent.*;
public class model04 {
public static void main(String[] args) {
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
2,//核心线程2个
5,//最大线程5个
3,//3秒没人就断开，只留核心线程
TimeUnit.SECONDS,//时间单位
new LinkedBlockingDeque<>(5),//阻塞队列，相当于候客区
Executors.defaultThreadFactory(),//默认线程工厂
new ThreadPoolExecutor.AbortPolicy()//默认拒绝策略，例如：银行满了，还有人进来，不处理这个人，抛出异常
);
try {
for (int i = 1; i {
System.out.println(Thread.currentThread().getName()+"-------->OK");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
poolExecutor.shutdown();
}
}
}
9.3、四种拒绝策略
- new ThreadPoolExecutor.AbortPolicy() // 银行满了，还有人进来，不处理这个人的，抛出异常，默认的
- new ThreadPoolExecutor.CallerRunsPolicy() //当线程满时，就会交给上层线程处理，也就是创建该线程的线程，相当于父线程
- new ThreadPoolExecutor.DiscardPolicy() //队列满了，丢掉任务，不会抛出异常！
- new ThreadPoolExecutor.DiscardOldestPolicy() //队列满了，尝试去和最早的竞争，但是竞争失败也会被抛弃，也不会抛出异常！
9.4、最大线程数设定方式
- CPU 密集型：也就是根据当前电脑的CPU核数去设定最大线程数，以此来达到使CPU的最大限度的使用，一般可以通过Runtime.getRuntime().availableProcessors()方法获得CPU核数
- IO 密集型：也就是通过判断程序中十分耗IO资源的线程数，然后大于这个数即可
10.四大函数式接口
1.消费型接口：Consumer : void accpt(T t)
import java.util.function.Consumer;
public class Test {
public static void main(String[] args) {
//只有一个参数，没有返回值，有一个泛型，参数类型为传入的泛型类型
Consumer cusumer = new Consumer() {
@Override
public void accept(String s) {
System.out.println("输出s==>".concat(s));
}
};
cusumer.accept("yinkai");
}
}
2.供给型接口：Supplier: T get( )
import java.util.function.Supplier;
public class Test {
public static void main(String[] args) {
//没有参数，只有返回值，有一个泛型，返回值的类型为泛型的类型
Supplier supplier = new Supplier() {
@Override
public String get() {
return "hello world";
}
};
String s = supplier.get();
System.out.println(s);
}
}
3.函数型接口：Function
/**
public interface Function {
R apply(T t);
}
传入两个泛型，一个是参数类型，一个是返回类型，并且有一个参数
*/
import java.util.function.Function;
public class Test {
public static void main(String[] args) {
Function function = new Function() {
@Override
public Integer apply(String s) {
return s.length();
}
};
Integer size = function.apply("yinkai");
System.out.println(size);
}
}
4.断言型接口：Predicate : boolean test( T t)
import java.util.function.Predicate;
public class Test {
public static void main(String[] args) {
//有一个参数，一个泛型，参数类型为传入的泛型，返回值固定是boolean类型
Predicate predicate = new Predicate() {
@Override
public boolean test(String o) {
return o.isEmpty();
}
};
boolean test = predicate.test("");
System.out.println(test);
}
}
11.Stream流式计算
- 简介：大数据：存储 + 计算；集合、MySQL 本质就是存储东西的；计算都应该交给流来操作！
- 作用：Stream是对集合功能的增强，它提供了各种非常便利、高效的聚合操作，可以大批量数据操作，同时再结合Lambda表达式，就可以极大的提高编程效率。
- 操作：
- 中间操作，返回Stream本身，这样就可以将多个操作依次串起来例如，
map、flatMap、filter、distinct、sorted、peek、limit、skip、parallel、sequential、unordered，这些操作之后可以跟其他的操作 - 最终操作，返回一特定类型的计算结果例如，
forEach、forEachOrdered、toArray、reduce、collect、min、max、count、anyMatch、allMatch、noneMatch、findFirst、findAny、iterator，这些操作之后返回的都不是流本身了 - 实例：
package com.kuang.stream;
import java.util.Arrays; import java.util.List;
/**
* 题目要求：一分钟内完成此题，只能用一行代码实现！ * 现在有5个用户！筛选：
* 1、ID 必须是偶数
* 2、年龄必须大于23岁
* 3、用户名转为大写字母
* 4、用户名字母倒着排序
* 5、只输出一个用户！
*/
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 list = Arrays.asList(u1, u2, u3, u4, u5);
// 计算交给Stream流
// lambda表达式、链式编程、函数式接口、Stream流式计算
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);
}
}
- 常用方法：
iterator：返回迭代器对象forEach：将Stream中的每个元素，都执行一遍交给一个函数处理，由于是消费型函数，所以必须有一个参数
users.stream().forEach((user)->{
System.out.println(user.getAge()+1);
});
count：统计流中的元素数，并返回结果
System.out.println(users.stream().count());
注：无法再在后面跟forEach输出
– max：返回流中基于comparator所指定的比较规则，比较出的最大值
users.stream().max((u1,u2)->{
if(u1.getAge()
注：由于参数是一个比较器，所以需要传入两个参数，并且需要一个int类型的返回值
– toArray：使用调用流中的元素，生成数组返回
for (Object o : users.stream().toArray()) {
System.out.println(o);
}
filter：过滤,由于断言型函数，所以需要传入一个参数，返回类型boolean，但该方法过滤后返回的是list的对象
users.stream()
.filter((user)->{return user.getAge()>22;})
.forEach(System.out::println);
distinct：去重
users.stream()
.filter((user)->{return user.getAge()>21;})
.map((user)->{return user.getAge();})
.distinct()
.forEach(System.out::println);
map：映射,比如可以将list中的user实体映射为user的年龄,由于是函数型接口，所以需要传入一个参数
users.stream()
.filter((user)->{return user.getAge()>21;})
.map((user)->{return user.getAge();})
.forEach(System.out::println);
collect(Collectors.toList())：返回一个 Collector ，它将输入元素 List到一个新的 Listlimit：limit返回包含前n个元素的流，当集合大小小于n时，则返回实际长度的流
users.stream()
.limit(5)
.forEach(System.out::println);
sorted：sorted要求待比较的元素必须实现comparator接口
users.stream().sorted((u1,u2)->{
if(u1.getAge()
注：如果直接是基本数据类型的话那么可以不用实现排序规则，可以直接使用默认的排序规则
– 调用串行Stream的 parallel()方法，可以将其转换 并行Stream 注：串行是有序的执行，并行是同时执行
– skip： 元素跳过，抛弃前指定个元素后，使用剩下的元素组成新的Stream对象返回
users.stream()
.skip(2)
.forEach(System.out::println);
- peek：生成一个包含原Stream的所有元素的新Stream，同时会提供一个消费函数即引用的方法，当Stream每个元素被消费的时候都会先执行新Stream给定的方法
users.stream()
.peek(user->{user.setAge(user.getAge()+1);})
.forEach(System.out::println);
}
- flatMap()：元素一对多转换，对Stream对象中的所有元素进行操作，每个元素会有一个或多个结果，然后将所有的元素组合成一个统一的Stream并返回
Arrays.stream(strs)
.map(str -> str.split(""))
.flatMap(Arrays::stream)// 扁平化为Stream
.distinct()
.forEach(System.out::println);
12.ForkJoin
- 介绍：ForkJoin是一个分而治之的任务框架，如一个任务需要多线程执行，分割成很多块计算的时候，可以采用这种方法。
- 动态规范：和分而治之不同的是，每个小任务之间互相联系。
- 工作密取：分而治之分割了每个任务之后，某个线程提前完成了任务，就会去其他线程偷取任务来完成，加快执行效率。同时，第一个分配的线程是从队列中的头部拿任务，当完成任务的线程去其他队列拿任务的时候是从尾部拿任务，所以这样就避免了竞争。
- 使用：
- 首先通过继承ForkJion的两个子类来生成任务类：
- RecursiveAction：用于没有返回结果的任务
- RecursiveTask：用于有返回结果的任务
//有返回值的任务
private static class SumTask extends RecursiveTask {//该泛型为返回值的类型
//要累加的源数组
private int[] src;
//开始角标
private int startIndex;
//结束角标
private int endIndex;
public SumTask(int[] src, int startIndex, int endIndex) {
this.src = src;
this.startIndex = startIndex;
this.endIndex = endIndex;
}
//实现具体的累加逻辑和任务分割逻辑
@Override
protected Integer compute() {
//不满足阈值的时候，这里面的逻辑也是当满足阈值的时候，递归执行的逻辑
if (endIndex - startIndex < POINT) {
int count = 0;
for (int i = startIndex; i
主要方法：
* compute()：继承抽象类后必须实现的方法
* fork()：通过该方法将任务压入到线程队列中以便执行
* join()：得到任务的结果
2. 接着通过ForkJoinPool创建池，然后将任务交给该对象执行：
private static void testForkJoin() {
//创建ForkJoinPool池
ForkJoinPool forkJoinPool = new ForkJoinPool();
long startTime = System.currentTimeMillis();
SumTask task = new SumTask(array, 0, array.length - 1);
//forkJoinPool.execute(task);提交任务，但是没有返回值
ForkJionTask submit=forkJoinPool.submit(task);//提交任务有返回值
//submit.get();通过该方法也可以得到结果
long endTime = System.currentTimeMillis();
System.out.println("采用forkjoin执行结果是：" + task.join() + "---------用时：" + (endTime - startTime));
}
主要方法：
+ new ForkJoinPool()：通过该方法创建一个池
+ forkJoinPool.submit(task)：通过该方法去执行任务，而任务对象的类必须实现了ForkJion的接口或者ForkJionTask继承其实现类
+ task.join()：通过该方法得到结果
+ submit.get()：通过该方法也可以得到结果
3. 注意：如果是单核、单CPU，不建议使用该框架，会带来额外的性能开销，反而比单线程的执行效率低。当然不是因为并行的任务会进行频繁的线程切换，因为ForkJoin框架在进行线程池初始化的时候默认线程数量为Runtime.getRuntime().availableProcessors()，单CPU单核的情况下只会产生一个线程，并不会造成线程切换，而是会增加ForkJoin框架的一些队列、池化的开销。
13.异步回调Future
主要使用的类：CompletableFuture，是Future接口的实现类
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
public class FutureDemo {
public static void main(String[] args) throws Exception {
CompletableFuture completableFuture = CompletableFuture.runAsync(() -> {
try {
TimeUnit.SECONDS.sleep(2);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " 执行");
});
System.out.println("11111111111");
completableFuture.get();
CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(()->{
int a=10/0;
return 1024L;
});
Long result = completableFuture2.whenComplete((t,u)->{//参数t为返回值，u为错误信息，如果没有出错则错误信息为空，反之返回值为空
System.out.println("t:"+t);
System.out.println("u:"+u);
}).exceptionally((e)->{
System.out.println(e.getMessage());
return 1023L;
}).get();
System.out.println(result);
}
}
主要方法：
- runAsync(Runnable runnable)：没有返回值
- supplyAsync(Supplier supplier)：有返回值
- whenComplete：在有返回值时，使用该方法可以获取到返回值或者错误信息
- exceptionally：当执行发生错误的时候就执行该方法里面的内容
14.JMM
2. 八种操作：JMM规定了所有的变量都存储在主内存中，每条线程还有自己的工作内存，工作内存中保留了线程使用到的变量的主内存的副本。
JMM定义了 8 个操作来完成主内存和工作内存之间的交互操作。JVM 实现时必须保证下面介绍的每种操作都是 原子的。
– lock (锁定) – 作用于主内存的变量，它把一个变量标识为一条线程独占的状态。【加琐时的操作】
– unlock (解锁) – 作用于主内存的变量，它把一个处于锁定状态的变量释放出来，释放后的变量才可以被其他线程锁定。
– read (读取) – 作用于主内存的变量，它把一个变量的值从主内存传输到线程的工作内存中，以便随后的 load 动作使用。
– write (写入) – 作用于主内存的变量，它把 store 操作从工作内存中得到的变量的值放入主内存的变量中。
– load (载入) – 作用于工作内存的变量，它把 read 操作从主内存中得到的变量值放入工作内存的变量副本中。
– use (使用) – 作用于工作内存的变量，它把工作内存中一个变量的值传递给执行引擎，每当虚拟机遇到一个需要使用到变量的值得字节码指令时就会执行这个操作【运算】。
– assign (赋值) – 作用于工作内存的变量，它把一个从执行引擎接收到的值赋给工作内存的变量，每当虚拟机遇到一个给变量赋值的字节码指令时执行这个操作【赋值】。
– store (存储) – 作用于工作内存的变量，它把工作内存中一个变量的值传送到主内存中，以便随后 write 操作使用。
3. 该内存模型带来的问题即解决：
– 问题：当A线程读取了主内存中的某个值后，在它还没有将新的值刷回主内存之前，此时有一个B线程对主内存的值进行了更改，而此时这种更改对于A来说是不可见的
– 解决：Volatile
15.Volatile
- 保证可见性：即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。
- 不保证原子性：volatile无法保证对变量的任何操作都是原子性的。比如让一个用volatile修饰的变量自增，然后让10个线程去调用它，每个线程执行100次，理论上最终结果应该为1000，但是结果却不会是1000，而是少于1000。
- 怎么保证原子性（除了使用lock和synchronized）：使用原子类
import java.util.concurrent.atomic.AtomicInteger;
/**
* AtomicInteger demo
*/
public class AtomicIntegerDemo {
private static AtomicInteger sum = new AtomicInteger(0);
public static void increase(){
sum.incrementAndGet();//值加一操作
}
public static void main(String[] args) throws InterruptedException{
for (int i=0;i{
for (int j=0;j
- 禁止指令重排：volatile关键字能禁止指令重排序，所以volatile能在一定程度上保证有序性。
16.原子类
参考博客：https://www.jianshu.com/p/28beef89dcd4
- 定义：Java中提供了一些原子类，原子类包装了一个变量，并且提供了一系列对变量进行原子性操作的方法。我们在多线程环境下对这些原子类进行操作时，不需要加锁，大大简化了并发编程的开发。
- 底层实现：大部分底层使用了CAS锁（CompareAndSet自旋锁），如AtomicInteger、AtomicLong等；也有使用了分段锁+CAS锁的原子类，如LongAdder等。
- 常用原子类： 1.AtomicInteger与AtomicLong
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author IT00ZYQ
* @date 2021/5/24 15:33
**/
public class T13_AtomicInteger {
private static AtomicInteger atomicInteger = new AtomicInteger();
private static AtomicLong atomicLong = new AtomicLong();
private static Integer integer = 0;
private static Long lon = 0L;
public static void main(String[] args) {
// 创建10个线程，分别对atomicInteger、atomicLong、integer、lon进行1000次增加1的操作
// 如果操作是原子性的，那么正确结果 = 10 * 1000 = 10000
Thread[] threads = new Thread[10];
for (int i = 0; i < 10; i++) {
threads[i] = new Thread(() -> {
for (int j = 1; j
- LongAdder：LongAdder的底层实现使用了分段锁，每个段使用的锁是CAS锁，所以LongAdder的底层实现是分段锁+CAS锁。
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
/**
* @author IT00ZYQ
* @date 2021/5/24 15:33
**/
public class T13_AtomicInteger {
private static AtomicInteger atomicInteger = new AtomicInteger();
private static AtomicLong atomicLong = new AtomicLong();
private static LongAdder longAdder = new LongAdder();
private static Integer integer = 0;
private static Long lon = 0L;
public static void main(String[] args) {
// 创建10个线程，分别对atomicInteger、atomicLong、integer、lon进行1000次增加1的操作
// 如果操作是原子性的，那么正确结果 = 10 * 1000 = 10000
Thread[] threads = new Thread[10];
for (int i = 0; i < 10; i++) {
threads[i] = new Thread(() -> {
for (int j = 1; j
- Unsafe：该类是CAS的核心类，由于java无法直接访问计算机的底层，所以就有了Unsafe类，该类的方法都是加上了native关键字的，表示可以调用c++的去操作底层，利用它来直接操作内存，因此效率也很高。
17.CAS
- 介绍：CAS是compare and swap的缩写，即我们所说的比较交换。cas是一种基于锁的操作，而且是乐观锁。
- 执行函数：CAS(V,A,B)
- V 内存位置
- A 预期原值
- B 新值
- 执行原理：如果内存位置（V）的值与预期原值（A）相匹配，那么处理器会自动将该位置值更新为新值 （B）。否则，处理器不做任何操作。它都会在 CAS 指令之前返回该位置的值（V）。（在 CAS 的一些特殊情况下将仅返回 CAS 是否成功，而不提取当前值。）CAS 有效地说明了”我认为位置 V 应该包含值 A；如果包含该值，则将 B 放到这个位置；否则，不要更改该位置，只告诉我这个位置现在的值即可。”
- ABA问题：因为CAS需要在操作值的时候检查下值有没有发生变化，如果没有发生变化则更新，但是如果一个值原来是A，变成了B，又变成了A，那么使用CAS进行检查时会发现它的值没有发生变化，但是实际上却变化了。
- 解决：ABA问题的解决思路就是使用版本号。在变量前面追加上版本号，每次变量更新的时候把版本号加一，那么A－B－A 就会变成1A-2B－3A。可以使用AtomicStampReference 原子类
- AtomicStampReference：在cas的基础上增加了一个标记stamp，使用这个标记可以用来觉察数据是否发生变化，给数据带上了一种实效性的检验。它有以下几个参数：
//参数代表的含义分别是 期望值，写入的新值，期望标记，新标记值
public boolean compareAndSet(V expected,V newReference,int expectedStamp,int newStamp);
public V getRerference();
public int getStamp();
public void set(V newReference,int newStamp);
- 实例：
public class Demo24 {
//initialRef 期望值 ， initialStamp 版本号时间戳
static AtomicStampedReference atomicStampedReference = new AtomicStampedReference(1, 1);
public static void main(String[] args) {
new Thread(()->{
int stamp = atomicStampedReference.getStamp();//获取当前最新版本号
System.out.println("a1="+stamp);
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(atomicStampedReference.compareAndSet(1, 2,
atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1));
System.out.println("a2=" + atomicStampedReference.getStamp());
System.out.println(atomicStampedReference.compareAndSet(2, 1,
atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1));
System.out.println("a3=" + atomicStampedReference.getStamp());
},"a").start();
//乐观锁原理相同
new Thread(()->{
int stamp = atomicStampedReference.getStamp();
System.out.println("b1="+stamp);
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(atomicStampedReference.compareAndSet(1,6, stamp, stamp + 1));
System.out.println("b2=" + atomicStampedReference.getStamp());
},"b").start();
}
}
总结：CAS： 比较当前工作内存中的值，若果这个值是期望中的，那么执行该操作 ，如果不是会一直循环因为底层是do while会一直循环
缺点： 1.循环耗时 2.一次性只能保证一个共享变量的共享性 3.ABA问题
大坑：！！！
总结：发现都是用==来判断的而并不是equals来判断，但是 ==是比较对象地址的而Integer类型的范围-128~127，就会在堆里面重新创建一个对象而不使用原本的对象！！！
18.各种锁
- 公平锁和非公平锁：
公平锁:非常公平，不能够插队，必须先来后到!FIFO
非公平锁:非常不公平，可以插队（默认都是非公平)
- 可重入锁：又叫递归锁
- 什么是可重入锁：通常情况下，锁可以用来控制多线程的访问行为。那对于同一个线程，如果连续两次对同一把锁进行lock，会怎么样了？对于一般的锁来说，这个线程就会被永远卡死在那边，如
voidhandle(){
lock();
lock();//和上一个lock()操作同一个锁对象，那么这里就永远等待了
unlock();
unlock();
}
重入锁使得同一个线程可以对同一把锁，在不释放的前提下，反复加锁，而不会导致线程卡死。而唯一需要注意的就是加锁次数和解锁次数必须一致！！
2. 可重入锁种类：
+ 隐式锁：即synchronized关键字使用的锁，默认是可重入锁
public class Demo25 {
public static void main(String[] args) {
Deap deap = new Deap();
new Thread(()->{
deap.sms();
},"A").start();
new Thread(()->{
deap.sms();
},"B").start();
}
}
class Deap {
public synchronized void sms() {
System.out.println(Thread.currentThread().getName() + "sms");
call();
}
public synchronized void call() {
System.out.println(Thread.currentThread().getName() + "call");
}
}
+ 显式锁：即Lock也有ReentrantLock这样的可重入锁。
public class Demo25 {
public static void main(String[] args) {
Deap deap = new Deap();
new Thread(()->{
deap.sms();
},"A").start();
new Thread(()->{
deap.sms();
},"B").start();
}
}
class Deap {
Lock lock = new ReentrantLock();
public void sms() {
lock.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();
lock.lock();
try {
System.out.println(Thread.currentThread().getName() + "call");
} catch (Exception e) {
e.printStackTrace();
}finally {
lock.unlock();
lock.unlock();
}
}
}
+ 总结：
* 对于同一个线程，重入锁允许你反复获得通一把锁，但是，申请和释放锁的次数必须一致。
- 默认情况下，重入锁是非公平的，公平的重入锁性能差于非公平锁
-
重入锁的内部实现是基于CAS操作的。
-
自旋锁：
- 什么是自旋锁：简单来说就是通过while循环的方式，如果不满足条件就一直循环，顾名思义叫自旋锁，大部分原子类的底层都是自旋锁实现的，如：
public final int getAndSetInt(Object var1, long var2, int var4) {
int var5;
do {
var5 = this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var4));//自旋
return var5;
}
- 自旋锁：它并不会放弃CPU 时间片，而是通过自旋等待锁的释放，也就是说，它会不停地再次地尝试获取锁，如果失败就再次尝试，直到成功为止
- 通过在自旋锁自定义一个简单锁：
public class Demo26 {
//Thread null
AtomicReference 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);
}
}
- 自旋锁的优缺点：
- 优点：自旋锁用循环去不停地尝试获取锁，让线程始终处于 Runnable 状态，节省了线程状态切换带来的开销
- 缺点：虽然避免了线程切换的开销，但是它在避免线程切换开销的同时也带来了新的开销，因为它需要不停得去尝试获取锁。如果这把锁一直不能被释放，那么这种尝试只是无用的尝试，会白白浪费处理器资源。也就是说，虽然一开始自旋锁的开销低于线程切换，但是随着时间的增加，这种开销也是水涨船高，后期甚至会超过线程切换的开销，得不偿失。
- 死锁排查：
- 死锁排查：
1、使用jps -l定位进程号
例如：jps -l
2、使用jstack 进程号 找到死锁问题
例如：jstack 110
Original: https://www.cnblogs.com/xiaoye-Blog/p/16630365.html
Author: 小也取不到名字
Title: JUC并发编程进阶！！
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/590012/
转载文章受原作者版权保护。转载请注明原作者出处！