JUC
- 环境准备
- 线程和进程
- Lock锁（接口）
- 生产者和消费者的问题
- condition实现精准通知唤醒
- 8锁现象彻底理解锁.
- CopyOnWriteArrayList
- CopyOnWriteArraySet
- ConcurrentHashMap
- 走进Callable
- 常用辅助类(必回)
- ReadWritelock
- 阻塞队列BlockingQueue
- 同步队列SychronizedQueue
- 线程池(重点)
- 四大函数是接口（重点）
- Stream流式计算
- ForkJoin
- 异步回调
- JMM
- Volatile
- 指令重排
- 彻底玩转单例模式
- 深入理解CAS
- 原子引用解决ABA问题
环境准备
新建maven项目
-
使用jdk8，项目中要用到新特性
-
查看官方文档学习
-
导入lombok
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.8</version>
</dependency>
线程和进程
什么是juc
Runnable没有返回值，企业中用Callable
java默认有两个线程：main ，GC
线程：开了一个进程TYpora，写字，自动保存（线程负责）
java真的可以开启线程吗？不可以
private native void start0（）；
//本地方法，调用底层c++，java运行在虚拟机之上，无法直接操作硬件，由c++开启多线程
8核心
并发编程的本质：充分利用cpu资源
package com.fang.demo0;
public class Test1 {
public static void main(String[] args) {
//获取cpu的核数
//cpu密集型，io密集型
System.out.println(Runtime.getRuntime().availableProcessors());
}
}
查看源码Thread.State
线程的状态：6个
public enum State {
//就绪
NEW,
//运行
RUNNABLE,
//阻塞
BLOCKED,
//等待
WAITING,
//超时等待
TIMED_WAITING,
//终止
TERMINATED;
}
wait与sleep的区别
- 来自不同的类
wait=》Object
sleep=》Thread - wait释放锁，sleep抱着锁睡觉
- wait必须在同步代码快中，sleep可以在任何地方睡觉
- wait不需要捕获异常，sleep需要捕获异常(可能发生超时等待)
lock锁（重点）
传统的Synchronized锁
不加Synchronized
package com.fang.demo0;
/**
* 真正的多线程开发
* 线程就是一个资源类，没有任何附属的操作
*/
public class SaleTicket {
public static void main(String[] args) {
Ticket ticket = new Ticket();
//Runnable接口为函数式接口
new Thread(()->{
for (int i = 0; i < 40; i++) {
ticket.sale();
}
},"a").start();
new Thread(()->{
for (int i = 0; i < 40; i++) {
ticket.sale();
}
},"b").start();
new Thread(()->{
for (int i = 0; i < 40; i++) {
ticket.sale();
}
},"c").start();
}
}
//资源类oop编程
class Ticket {
//属性，方法
private int number = 50;
//买票的方式
public void sale() {
if (number>0) {
System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余："+number);
}
}
}
买票结果混乱
加了锁后正常
public synchronized void sale() {
if (number>0) {
System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余："+number);
}
}
Lock锁（接口）
默认非公平锁：十分不公平可以插队
公平锁：先来后到
加锁
class Ticket {
//属性，方法
private int number = 50;
Lock lock = new ReentrantLock();
//买票的方式
public void sale() {
lock.lock();//加锁
try {
//业务代码
if (number>0) {
System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余："+number);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();//解锁
}
}
}
Sychronized和lock的区别
1.Sychronized 内置的java关键字,Lock锁是一个java类
2.Sychronized 无法判断获取锁的状态,Lock锁可以判断是否获取到了锁.
3.Sychronized 会自动释放锁lock必须手动释放锁,如果不释放锁,死锁
4.Sychronized 线程一(获得锁,阻塞),线程二(等待,傻傻的等),Lock锁就不一定会等待下去.
5.Sychronized 可重入锁,不可以中断,非公平;Lock,可重入锁,可以中断锁,非公平(可以自己设置)
6.Sychronized 适合锁少量的代码的同步问题,Lock适合锁大量的代码同步问题.
锁是什么,如何判断锁的是谁
生产者和消费者的问题
面试:单例模式,排序算法,生产者消费者,死锁
Sychronized版
package com.fang.product;
/**
* 线程间的通信问题:生产者和消费者的问题! 等待唤醒 通知唤醒
* 线程交替执行 A B同时操作一个变量
* A num+1
* B num-1
*/
public class A {
public static void main(String[] args) {
Data data = new Data();
new Thread(()->{
for (int i = 0; i < 10; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"A").start();
// new Thread(()->{
// for (int i = 0; i < 10; i++) {
// try {
// data.increment();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }
// },"C").start();
new Thread(()->{
for (int i = 0; i < 10; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"B").start();
}
}
//等待 业务 通知
class Data{//数字 资源类
private int num = 0;
public synchronized void increment() throws InterruptedException {
if (num != 0){
//等待
this.wait();
}
num++;
System.out.println(Thread.currentThread().getName()+">="+num);
//通知其他线程,我加一完毕了
this.notifyAll();
}
public synchronized void decrement() throws InterruptedException {
if (num == 0){
this.wait();
}
num--;
System.out.println(Thread.currentThread().getName()+">="+num);
//通知其他线程,我减一完毕
this.notifyAll();
}
}
加入C线程,执行结果
存在问题虚假唤醒
将if改成while防止虚假唤醒.
guc版生产者消费者问题
代码实现
package com.fang.product;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class B {
public static void main(String[] args) {
Data2 data2 = new Data2();
new Thread(()->{
for (int i = 0; i < 10; i++) {
data2.increment();
}
},"A").start();
new Thread(()->{
for (int i = 0; i < 10; i++) {
data2.increment();
}
},"C").start();
new Thread(()->{
for (int i = 0; i < 10; i++) {
data2.decrement();
}
},"B").start();
}
}
class Data2{//数字 资源类
private int num = 0;
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
// condition.await()等待 condition.signalAll()唤醒全部
public void increment() {
lock.lock();
try {
//业务代码
while (num != 0){
condition.await();
}
num++;
System.out.println(Thread.currentThread().getName()+">="+num);
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void decrement() {
lock.lock();
try {
//业务代码
while (num == 0){
condition.await();
}
num--;
System.out.println(Thread.currentThread().getName()+">="+num);
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
线程状态随机
想要的执行循序A->B->C
condition实现精准通知唤醒
package com.fang.product;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* A->B->C
*/
public class C {
public static void main(String[] args) {
Data3 data3 = new Data3();
new Thread(()->{
for (int i = 0; i < 10; i++) {
data3.printA();
}
},"A").start();
new Thread(()->{
for (int i = 0; i < 10; i++) {
data3.printB();
}
},"B").start();
new Thread(()->{
for (int i = 0; i < 10; i++) {
data3.printC();
}
},"C").start();
}
}
class Data3{
private Lock lock = new ReentrantLock();
private Condition condition1 = lock.newCondition();
private Condition condition2 = lock.newCondition();
private Condition condition3 = lock.newCondition();
private int num =1; //1A,2B,3C
public void printA() {
lock.lock();
try {
//业务,判断,执行,通知
while (num != 1){
condition1.await();
}
System.out.println("aaaaaaaaaaaa");
//唤醒指定的人,B
num =2;
condition2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printB() {
lock.lock();
try {
while (num != 2){
condition2.await();
}
System.out.println("bbbbbbbbbbbb");
//唤醒指定的人,B
num =3;
condition3.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printC() {
lock.lock();
try {
while (num != 3){
condition3.await();
}
System.out.println("ccccccccccc");
//唤醒指定的人,B
num =1;
condition1.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
8锁现象彻底理解锁.
什么是锁,锁到底锁的是谁
package com.fang.lock8;
import java.sql.Time;
import java.util.concurrent.TimeUnit;
/**
* 8锁,就是关于锁的8个问题
* 1.标准情况下是先发短信还是打电话
* 2.发短信方法延迟4秒
*/
public class Test1 {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(()->{
phone.sendSms();
},"A").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(()->{
phone.call();
},"B").start();
}
}
class Phone{
//synchronized锁的对象是方法的调用者
//两个方法用的是同一个锁,谁先拿到谁先执行
public synchronized void sendSms() {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("sendSms");
}
public synchronized void call() {
System.out.println("call");
}
}
package com.fang.lock8;
import java.sql.Time;
import java.util.concurrent.TimeUnit;
/**
* 3.增加一个普通方法,是先执行发短信还是hello（1秒钟输出hello，4秒后输出发短信）
* 4.两个对象，两个同步方法，先打电话，再发短信（两个不同的对象，两把锁）
*/
public class Test2 {
public static void main(String[] args) {
Phone2 phone = new Phone2();
Phone2 phone2 = new Phone2();
new Thread(()->{
phone.sendSms();
},"A").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(()->{
phone2.call();
},"B").start();
}
}
class Phone2{
//synchronized锁的对象是方法的调用者
//两个方法用的是同一个锁,谁先拿到谁先执行
public synchronized void sendSms() {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("sendSms");
}
public synchronized void call() {
System.out.println("call");
}
//这里没有锁,不是同步方法,不受锁的影响
public void hello() {
System.out.println("hello");
}
}
package com.fang.lock8;
import java.sql.Time;
import java.util.concurrent.TimeUnit;
/**
* 5.增加两个静态同步方法，只有一个对象（先发短信，再打电话）
* 6.两个对象，增加两个静态同步方法（先发短信，再打电话）
*/
public class Test3 {
public static void main(String[] args) {
//两个对象的class模板只有一个，static，锁的是class
Phone3 phone = new Phone3();
Phone3 phone2 = new Phone3();
new Thread(()->{
phone.sendSms();
},"A").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(()->{
phone2.call();
},"B").start();
}
}
class Phone3{
//synchronized锁的对象是方法的调用者
//static 静态方法 类一加载就有了！class模板,锁的是class对象Class<Phone3> phone3Class = Phone3.class;
//两个方法用的是同一个锁
public static synchronized void sendSms() {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("sendSms");
}
public static synchronized void call() {
System.out.println("call");
}
}
package com.fang.lock8;
import java.sql.Time;
import java.util.concurrent.TimeUnit;
/**
* 7.一个静态同步方法，一个普通的同步方法，一个对象（先打电话）
* 8.两个对象（先打电话）
*/
public class Test4 {
public static void main(String[] args) {
//两个对象的class模板只有一个，static，锁的是class
Phone4 phone = new Phone4();
Phone4 phone2 = new Phone4();
new Thread(()->{
phone.sendSms();
},"A").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(()->{
phone2.call();
},"B").start();
}
}
class Phone4{
//锁的是class类模板
//static 静态方法 类一加载就有了！class模板,锁的是class对象Class<Phone3> phone3Class = Phone3.class;
//两个方法用的是同一个锁
public static synchronized void sendSms() {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("sendSms");
}
//普通的同步方法，锁的是调用者
public synchronized void call() {
System.out.println("call");
}
}
CopyOnWriteArrayList
单线程的情况下，线程安全
package com.fang.unsafe;
import java.util.Arrays;
import java.util.List;
public class ListTest {
public static void main(String[] args) {
List<String> list = Arrays.asList("1","2","3");
list.forEach(System.out::println);
}
}
出现并发修改异常
package com.fang.unsafe;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
//java.util.ConcurrentModificationException 并发修改异常
public class ListTest {
public static void main(String[] args) {
//并发下ArrayList不安全
List<String> list = new ArrayList<>();
for (int i = 0; i < 10; i++) {
new Thread(()->{
list.add(UUID.randomUUID().toString().substring(0,5));
System.out.println(list);
},String.valueOf(i)).start();
}
}
}
解决办法：1.List list = new Vector<>();
2.集合工具类的使用List list = Collections.synchronizedList(new ArrayList<>());
3.List list = new CopyOnWriteArrayList<>();
CopyOnWrite：写入时复制，COW 计算机程序设计领域的优化策略。
多个线程调用的时候，list读取时固定，写入时覆盖
在写入时避免覆盖，造成数据问题
读写分离 mycat
CopyOnWriteArrayList比Vector好，
Vector这个用的是Sychronized，效率比CopyOnWriteArrayList低。
CopyOnWriteArraySet
package com.fang.unsafe;
import java.util.HashSet;
import java.util.UUID;
public class SetTest {
public static void main(String[] args) {
HashSet<Object> hashSet = new HashSet<>();
for (int i = 0; i < 10; i++) {
new Thread(()->{
hashSet.add(UUID.randomUUID().toString().substring(0,5));
System.out.println(hashSet);
},String.valueOf(i)).start();
}
}
}
执行结果
解决办法
Set set = new CopyOnWriteArraySet<>();
hashSet底层是什么
底层是HashMap，set的add的方法本质就是map，key是无法重复的。
ConcurrentHashMap
解决办法
Map<String, String> map = new ConcurrentHashMap<>();
走进Callable
1.可以有返回值,可以抛出异常,方法不同,run(),/call()
代码测试
package callable;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
public class CallableTest {
public static void main(String[] args) {
new Thread().start();//怎么启动callable
MyThread thread = new MyThread();
FutureTask futureTask = new FutureTask(thread); //适配类
new Thread(futureTask,"A").start();
new Thread(futureTask,"B").start();//结果会被缓存,效率高
try {
String s = (String) futureTask.get();//callable的返回值,这个get方法可能会产生阻塞,把他放在最后
//或者使用异步通信来处理
System.out.println(s);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
class MyThread implements Callable<String> {
@Override
public String call() throws Exception {
System.out.println("call");//会打印1个call
return "1234";
}
}
常用辅助类(必回)
CountDownLatch
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
//总数是6
CountDownLatch countDownLatch = new CountDownLatch(6);
for (int i = 0; i < 6; i++) {
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"go out");
countDownLatch.countDown();
},String.valueOf(i)).start();
}
//等待计数器归零才会向下执行
countDownLatch.await();
countDownLatch.countDown();//-1
}
}
6个线程都走完才会向下执行
CyclicBarrier
加法计数器
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
public static void main(String[] args) {
//集齐七颗龙珠，召唤神龙
//召唤龙珠
CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
System.out.println("召唤神龙成功");
});
for (int i = 0; i <= 7; i++) {
final int temp = i;
//lambda不能直接拿到for循环中的i
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"收集"+temp);
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}
Semaphore
抢车位
package add;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
public class SemaphoreDemo {
public static void main(String[] args) {
Semaphore semaphore = new Semaphore(3);
for (int i = 1; i <= 6; i++) {
new Thread(()->{
//acquire()
try {
semaphore.acquire();//获得,如果满了,会等待被释放为止
System.out.println(Thread.currentThread().getName()+"抢到车位");
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName()+"离开车位");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
semaphore.release();//释放
}
},String.valueOf(i)).start();
}
}
}
ReadWritelock
实现类:ReetrantReadWritelock
读可以被多个线程同时读,写的时候只能有一个线程去写
package rw;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
//ReadWriteLock
public class ReadWriteLockDemo {
// 自定义缓存
public static void main(String[] args) {
MyCatchLock myCatch = new MyCatchLock();
//写入
for (int i = 0; i <= 5; i++) {
final int temp = i;
new Thread(()->{
myCatch.put(temp+"",temp+"");
},String.valueOf(i)).start();
}
//读取
for (int i = 0; i <=5; i++) {
final int temp = i;
new Thread(()->{
myCatch.get(temp+"");
},String.valueOf(i)).start();
}
}
}
class MyCatch {
private volatile Map<String,Object> map = new HashMap<>(0);
//存
public void put(String key,Object value) {
System.out.println(Thread.currentThread().getName()+"写入"+value);
map.put(key, value);
System.out.println(Thread.currentThread().getName()+"写入成功");
}
//取
public void get(String key) {
System.out.println(Thread.currentThread().getName()+"读取"+key);
Object o = map.get(key);
System.out.println(Thread.currentThread().getName()+"读取成功");
}
}
//加锁的
class MyCatchLock {
private volatile Map<String,Object> map = new HashMap<>(0);
//读写锁更加细粒度的控制
private ReadWriteLock lock = new ReentrantReadWriteLock();
private Lock lock1 = new ReentrantLock();
//存,写的时候,只希望同时有一个线程写
public void put(String key,Object value) {
lock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName()+"写入"+value);
map.put(key, value);
System.out.println(Thread.currentThread().getName()+"写入成功");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.writeLock().unlock();
}
}
//取,读,所有的人都可以读
public void get(String key) {
lock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName()+"读取"+key);
Object o = map.get(key);
System.out.println(Thread.currentThread().getName()+"读取成功");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.readLock().unlock();
}
}
}
独占锁(写锁) 一次只能被一个线程占有
共享锁(读锁) 可以同时被多个线程占有
阻塞队列BlockingQueue
写入:如果队列满了,就必须阻塞等待
取:如果队列是空的就必须阻塞等待生产
学会使用队列:四组api
1.抛出异常
2.不会抛出异常
3.阻塞等待
4.超时等待
| 方式 | 抛出异常 | 有返回值,不抛出异常 | 阻塞等待 | 超时等待 |
|---|---|---|---|---|
| 添加 | add | offer | put | offer(,) |
| 移除 | remove | poll | take | poll(,) |
| 判断队列首 | element | peek | - | - |
抛出异常
public class Test {
public static void main(String[] args) {
//List,Set的父类Collection
//BlockQueue不是新的东西继承自 Collection
// 什么情况下会使用阻塞队列：多线程并发处理，线程池
test1();
}
public static void test1 () {
//队列的大小
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(arrayBlockingQueue.add("a"));
System.out.println(arrayBlockingQueue.add("b"));
System.out.println(arrayBlockingQueue.add("c"));
System.out.println(arrayBlockingQueue.element());//查看队首元素
//ava.lang.IllegalStateException
System.out.println(arrayBlockingQueue.add("d"));
System.out.println("=============");
//队列移除顺序
System.out.println(arrayBlockingQueue.remove());
System.out.println(arrayBlockingQueue.remove());
System.out.println(arrayBlockingQueue.remove());
//java.util.NoSuchElementException
System.out.println(arrayBlockingQueue.remove());
}
}
不抛出异常
public static void test2 () {
//队列的大小
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(arrayBlockingQueue.offer("a"));
System.out.println(arrayBlockingQueue.offer("b"));
System.out.println(arrayBlockingQueue.offer("c"));
//ava.lang.IllegalStateException
System.out.println(arrayBlockingQueue.offer("d"));
System.out.println(arrayBlockingQueue.peek());
System.out.println("=============");
//队列移除顺序
System.out.println(arrayBlockingQueue.poll());
System.out.println(arrayBlockingQueue.poll());
System.out.println(arrayBlockingQueue.poll());
System.out.println(arrayBlockingQueue.poll());
}
}
等待阻塞(一直阻塞)
public static void test3 () throws InterruptedException {
//队列的大小
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(3);
arrayBlockingQueue.put("a");
arrayBlockingQueue.put("b");
arrayBlockingQueue.put("c");
// arrayBlockingQueue.put("d");队列没有位置,一直阻塞
System.out.println("=============");
//队列移除顺序
System.out.println(arrayBlockingQueue.take());
System.out.println(arrayBlockingQueue.take());
System.out.println(arrayBlockingQueue.take());
System.out.println(arrayBlockingQueue.take());//没有这个元素,一直阻塞
}
等待阻塞(等待超时)
public static void test4 () throws InterruptedException {
//队列的大小
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(arrayBlockingQueue.offer("a"));
System.out.println(arrayBlockingQueue.offer("b"));
System.out.println(arrayBlockingQueue.offer("c"));
//等待超过两秒退出
arrayBlockingQueue.offer("d", 2,TimeUnit.SECONDS);
System.out.println(arrayBlockingQueue.poll());
System.out.println(arrayBlockingQueue.poll());
//等待超过两秒九退出
System.out.println(arrayBlockingQueue.poll(2,TimeUnit.SECONDS));
同步队列SychronizedQueue
没有容量,进去一个元素,必须等待取出来之后,才能往里面再放一个元素
put,take
public static void main(String[] args) {
SynchronousQueue<Object> synchronousQueue = new SynchronousQueue<>();
new Thread(()->{
try {
System.out.println(Thread.currentThread().getName()+"put 1");
synchronousQueue.put("1");
System.out.println(Thread.currentThread().getName()+"put 2");
synchronousQueue.put("2");
System.out.println(Thread.currentThread().getName()+"put 3");
synchronousQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
},"t1").start();
new Thread(()->{
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+synchronousQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+synchronousQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+synchronousQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
},"t2").start();
}
线程池(重点)
池化技术及线程池的使用
程序的运行,本质:占用系统的资源!优化资源的使用
线程池,连接池,内存池,对象池
池化技术:事先准备好一些资源,有人要用就来拿,用完之后归还
线程池的好处
1.降低资源的消耗
2.提高响应速度
3.方便管理
线程可以复用,可以控制最大并发量,管理线程
线程池:三大方法,7大参数,4种拒绝策略
线程的三大方法
线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。 说明：Executors各个方法的弊端：
1）newFixedThreadPool和newSingleThreadExecutor:
主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。
2）newCachedThreadPool和newScheduledThreadPool:
主要问题是线程数最大数是Integer.MAX_VALUE(约为21亿)，可能会创建数量非常多的线程，甚至OOM。
public static void main(String[] args) {
//Executors工具类,三大方法
ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程
for (int i = 0; i < 10; i++) {
//使用了线程池之后,使用线程池来创建线程
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName()+"ok");
});
}
//线程池用完,程序结束,关闭线程池
try {
threadPool.shutdown();
} catch (Exception e) {
e.printStackTrace();
} finally {
}
}
//Executors工具类,三大方法
// ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程
// ExecutorService threadPool = Executors.newFixedThreadPool(5);//创建一个固定大小得线程池
ExecutorService threadPool = Executors.newCachedThreadPool();//可伸缩，线程数可变
7大参数及自定义线程池
源码分析
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
}
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
}
三大方法底层都调用得是ThreadPoolExecutor
//七个参数
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;
}
手动创建一个线程池
四种拒绝策略
public static void main(String[] args) {
//Executors工具类,三大方法
// ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程
// ExecutorService threadPool = Executors.newFixedThreadPool(5);//创建一个固定大小得线程池
// ExecutorService threadPool = Executors.newCachedThreadPool();//可伸缩，线程数可变
//自定义线程池，工作
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2,
5,
3,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(3),
Executors.defaultThreadFactory(),
// new ThreadPoolExecutor.AbortPolicy()//银行满了还有人进来，不处理这个人，抛出异常
// new ThreadPoolExecutor.DiscardPolicy()//队列满了不会抛出异常，丢掉任务
// new ThreadPoolExecutor.CallerRunsPolicy()//哪里来的去哪里
new ThreadPoolExecutor.DiscardOldestPolicy()//队列满了，尝试去和最早得竞争，也不会抛出异常
);
//最大承载：队列+max值
for (int i = 0; i < 8; i++) {
//使用了线程池之后,使用线程池来创建线程
threadPoolExecutor.execute(()->{
System.out.println(Thread.currentThread().getName()+"ok");
});
}
//线程池用完,程序结束,关闭线程池
try {
threadPoolExecutor.shutdown();
} catch (Exception e) {
e.printStackTrace();
} finally {
}
}
最大线程池应该如何定义
1.cpu密集行，12条线程同时执行，几核心就是几，可以保证cpu的效率最高
2.io密集型>判断你的程序中十分耗io的线程
程序 15个大型任务 io十分暂用资源
System.out.println(Runtime.getRuntime().availableProcessors());//获得cpu的核心数
四大函数是接口（重点）
新生代程序员：lambda表达式，链式编程，函数式接口，Steram流式计算
函数式接口：只有一个方法的接口//简化编程模型，在新版的框架中大量的应用
//foreach（）参数消费者类型的函数式接口
代码测试：
函数型接口
传入T,返回R
public static void main(String[] args) {
// Function function = new Function<String,String>() {
// @Override
// public String apply(String str) {
// return str;
// }
// };
Function function = (str)->{return str;};
System.out.println(function.apply("abc"));
}
断定型接口
返回boolean值
public static void main(String[] args) {
//判断字符串是否为空
Predicate<String> predicate = new Predicate<String>() {
@Override
public boolean test(String s) {
return s.isEmpty();
}
};
Predicate<String> predicate = (str)- >{return str.isEmpty();
System.out.println(predicate.test(""));
}
消费型接口和供给型接口
只有输入没有返回值
只有返回没有输入
Stream流式计算
存储+计算
存储:mysql,集合
计算都要交给流计算你
public class Test {
public static void main(String[] args) {
User user1 = new User(1,21,"张三");
User user2 = new User(2,23,"李四");
User user3 = new User(3,29,"王五");
User user4 = new User(4,18,"赵六");
//集合存储
List<User> userList = Arrays.asList(user1, user2, user3, user4);
//计算交给流
userList.stream().filter(user -> {return user.getId()%2==0;})
.filter(user -> {return user.getAge()>20;})
.map(user -> {return user.getName().toUpperCase(Locale.ROOT);})
.sorted((u1,u2)->{return u2.compareTo(u1); })
// .limit(1)//分页
.forEach(System.out::println);
}
}
ForkJoin
什么是ForkJoin
ForkJoin在JDk1.7,并行执行任务!提高效率,数据量大!
大数据:Map Reduce把大任务拆分为小任务.
特点:
特点工作窃取
ForkJoin的操作
异步回调
Future的设计初衷:
public static void main(String[] args) throws ExecutionException, InterruptedException {
// //发起一个请求,没有返回值得异步回调
// CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{
// try {
// TimeUnit.SECONDS.sleep(2);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// System.out.println(Thread.currentThread().getName()+"run");
// });
// System.out.println("1111");
// //获取阻塞执行结果
// completableFuture.get();
//有返回值的异步回调
//ajax,成功和失败回调
//返回的是错误信息
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(()->{
System.out.println("completableFuture"+Thread.currentThread().getName());
int i = 10/0;
return 1024;
});
System.out.println(completableFuture.whenComplete((t, u) -> {
System.out.println(t);//正常的返回结果
System.out.println(u);//错误信息java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
}).exceptionally((e) -> {
System.out.println(e.getMessage());//java.lang.ArithmeticException: / by zero
return 233;
}).get());
}
JMM
volatile是java虚拟机提供的轻量级的同步机制
1.保证可见性
2.不保证原子性
3.由于内存屏障,禁止指令重排
什么是JMM
JMM:java的内存模型,不存在的东西,概念,约定
关于JMM的一些同步的约定:
1.线程解锁前,必须把共享变量立刻刷回主存
2.线程枷锁前,必须读取主存中的最新值到工作的内存中
3.加锁和解锁是同一把锁
线程:工作内存 ,主内存
八种操作
???
private static int num = 0;
public static void main(String[] args) throws InterruptedException {//main线程
new Thread(()->{//线程1
while (num == 0) {
}
}).start();
TimeUnit.SECONDS.sleep(1);
num = 1;
System.out.println(num);
//程序一直在执行,线程1不知道主存中的值发生了变化
}
Volatile
可见性
//加了volatile可以保证可见性,不加进入死循环
private volatile static int num = 0;
public static void main(String[] args) throws InterruptedException {//main线程
new Thread(()->{//线程1
while (num == 0) {
}
}).start();
TimeUnit.SECONDS.sleep(1);
num = 1;
System.out.println(num);
//程序一直在执行,线程1不知道主存中的值发生了变化
}
不保证原子性
原子性:不可分割
线程a在执行任务的时候,不能被打扰,也不能被分割,要么同时成功,要么同时失败
private volatile static int num = 0;
public static void add() {
num++;
}
public static void main(String[] args) {
//理论上num结果应该为20000,加volatile还是不能加到2万,加Synchronized可以
for (int i = 0; i < 20; i++) {
new Thread(()->{
for (int j = 0; j < 1000; j++) {
add();
}
}).start();
}
while (Thread.activeCount()>2) {
Thread.yield();
}
System.out.println(Thread.currentThread().getName()+num);
}
如果不加Synchronizd和Lock怎么保证原子性
num++;//不是原子性操作
使用原子类解决问题
private volatile static AtomicInteger num = new AtomicInteger();
public static void add() {
// num++;
num.getAndIncrement();//AtomicInteger+1方法CAS效率高
}
public void main(String[] args) {
//理论上num结果应该为20000,加volatile还是不能加到2万,加Synchronized可以
for (int i = 0; i < 20; i++) {
new Thread(()->{
for (int j = 0; j < 1000; j++) {
add();
}
}).start();
}
while (Thread.activeCount()>2) {
Thread.yield();
}
System.out.println(Thread.currentThread().getName()+num);
}
这些类的底层都和操作系统挂钩,直接在内存中修改值,Unsafe类是一个很特殊的存在
指令重排
?你写的程序,计算机并不是按照你写的那样去执行
源代码->编译器优化->指令并行可能重排->内存系统可能重排->执行
int x=1;
int y=1;
x=x+5;
y=x+x;
我们期望的是1234,但是可能是21344,1324
不可能是4123,==处理器在执行指定重排的时候,考虑数据之间的依赖性
可能造成影响的结果x,y,a,b默认是0
volitale可以避免指令重排:
内存屏障.cpu指令.作用
1.保证特定的操作执行循序
2.可以保证某些变量的内存可见性(利用这些特性,保证valitale实现了可见性)
彻底玩转单例模式
饿汉式
package 单例模式;
//饿汉式单例
public class Hungry {
//一上来就创建对象，可能会浪费空间
private byte[] data1 = new byte[1024*1024];
private byte[] data2 = new byte[1024*1024];
private byte[] data3 = new byte[1024*1024];
private byte[] data4 = new byte[1024*1024];
private Hungry() {
}
private final static Hungry HUNGRY= new Hungry();
public static Hungry getInstance() {
return HUNGRY;
}
}
DCL懒汉式
package 单例模式;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
//懒汉式单例模式
public class LazyMan {
private static boolean qinjaing = false;
private LazyMan() {
System.out.println(Thread.currentThread().getName() + "ok");
synchronized (LazyMan.class) {
// if (lazyMan != null) {
// throw new RuntimeException("不要试图用反射破坏异常");
}
if (qinjaing != false) {
qinjaing = true;
} else {
throw new RuntimeException("不要试图用反射破坏异常");
}
}
//单线程下确实单例ok
private volatile static LazyMan lazyMan;
//双重检测锁模式 懒汉式单例模式 DCL懒汉式
public static LazyMan getInstance() {
if (lazyMan == null) {
synchronized (LazyMan.class) {
if (lazyMan == null) {
lazyMan = new LazyMan();//不是原子性操作，
//1.分配内存空间
//2.执行构造方法，初始化对象
//3.把这个对象指向这个空间
//真实步骤可能执行132.此时lazyman还没被完成构造
}
}
}
return lazyMan;
}
// //多线程并发
// public static void main(String[] args) {
// for (int i = 0; i < 10; i++) {
// new Thread(()->{
// LazyMan.getInstance();
// }).start();
// }
// }
// //反射破解使其不安全,破坏单例
// public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
// LazyMan instance = LazyMan.getInstance();
// //获得无参构造器
// Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null);
// declaredConstructor.setAccessible(true);
// LazyMan lazyMan = declaredConstructor.newInstance();
// //单例模式.LazyMan@15fbaa4
// //单例模式.LazyMan@1ee12a7
// System.out.println(instance);
// System.out.println(lazyMan);
// }
//两个对象都使用反射再次破坏单例模式
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, NoSuchFieldException {
//通过反射破坏标志位qinjiang
Field qinjaing = LazyMan.class.getDeclaredField("qinjaing");
qinjaing.setAccessible(true);
//获得无参构造器
Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null);
declaredConstructor.setAccessible(true);
LazyMan lazyMan = declaredConstructor.newInstance();
//
qinjaing.set(lazyMan,false);
LazyMan instance = declaredConstructor.newInstance();
//单例模式.LazyMan@15fbaa4
//单例模式.LazyMan@1ee12a7
System.out.println(instance);
System.out.println(lazyMan);
}
}
静态内部类
package 单例模式;
//静态内部类
public class Holder {
private Holder() {
}
public static Holder getInstance() {
return InnerClass.HOLDER;
}
public static class InnerClass {
private static final Holder HOLDER = new Holder();
}
}
单例不安全，因为有反射，所以使用枚举
package 单例模式;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
//enum是什么？本身也是一个class类
public enum EnumSingle {
INSTANCE;
public EnumSingle getInstance() {
return INSTANCE;
}
}
class Test{
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
EnumSingle enumSingle1 = EnumSingle.INSTANCE;
//反射不能破坏枚举
// EnumSingle enumSingle2 = EnumSingle.INSTANCE;
Constructor<EnumSingle> declaredConstructor = EnumSingle.class.getDeclaredConstructor(String.class,int.class);
declaredConstructor.setAccessible(true);
EnumSingle enumSingle2 = declaredConstructor.newInstance();
System.out.println(enumSingle1);
System.out.println(enumSingle2);
//Cannot reflectively create enum objects
// at java.lang.reflect.Constructor.newInstance(Constructor.java:417)
}
}
深入理解CAS
什么是CAS
操作系统，计算机网络
Unsafe类
package cas;
import java.util.concurrent.atomic.AtomicInteger;
public class CASDemo {
//CAS compareAndSet:比较并交换
public static void main(String[] args) {
AtomicInteger atomicInteger = new AtomicInteger(2020);
//public final boolean compareAndSet(int expect, int update)
//如果我期望的值达到了就更新,CAS是Cpu的并发原理
// 如果不是就一直循环，底层是自旋锁。
System.out.println(atomicInteger.compareAndSet(2020, 2021));
System.out.println(atomicInteger.get());
System.out.println(atomicInteger.compareAndSet(2020, 2021));
System.out.println(atomicInteger.get());
// true
// 2021
// false
// 2021
}
}
CAS缺点
1.循环会耗时
2.一次性只能保证一个共享变量的原子性
3.引发ABA问题