Day22

Day22

一,生产者消费者模型

1.1,单个生产者单个消费者

public class Test01 {/*** 知识点：生产者消费者模型 - 单个生产者单个消费者* * 分析：* 		产品类 - Phone：属性(brand,price)* 		生产者线程 - Producer* 		消费者线程 - Consumer* 最终的目的：生产一个、消费一个* * 步骤：* 		1.多个线程(生产者线程、消费者线程)操作同一个资源(产品类的对象)* 		2.多个产品之间来回切换(华为 <--> 小米)* 			null -- 0.0* 			华为 -- 0.0* 			小米 -- 3999* 			华为 -- 1999* 			脏数据的解决思路：加锁* 		3.生产一个消费一个* 			*/public static void main(String[] args) {//brand - null//price - 0.0Phone phone = new Phone();Producer p = new Producer(phone);Consumer c = new Consumer(phone);p.start();c.start();}
}

//产品类
public class Phone {private String brand;private double price;private boolean store;public Phone() {}public Phone(String brand, double price) {this.brand = brand;this.price = price;}public String getBrand() {return brand;}public void setBrand(String brand) {this.brand = brand;}public double getPrice() {return price;}public void setPrice(double price) {this.price = price;}public boolean isStore() {return store;}public void setStore(boolean store) {this.store = store;}@Overridepublic String toString() {return "Phone [brand=" + brand + ", price=" + price + "]";}
}

//生产者线程
public class Producer extends Thread{private Phone phone;public Producer(Phone phone) {this.phone = phone;}@Overridepublic void run() {boolean flag = true;while(true){synchronized(phone){if(phone.isStore()){//有库存//等待//1.将当前线程记录在对象监视器中(记录当前线程进入到阻塞状态)//2.释放锁资源(解锁)//3.当前线程进入到阻塞状态try {phone.wait();} catch (InterruptedException e) {e.printStackTrace();}}if(flag){phone.setBrand("华为");phone.setPrice(3999);}else{phone.setBrand("小米");phone.setPrice(1999);}flag = !flag;phone.setStore(true);//唤醒：唤醒的是对象监视器中随机一个等待的线程，唤醒的线程进入到就绪态phone.notify();}}}
}

//消费者线程
public class Consumer extends Thread{private Phone phone;public Consumer(Phone phone) {this.phone = phone;}@Overridepublic void run() {while(true){synchronized(phone){if(!phone.isStore()){//没有库存try {//等待：//1.将当前线程记录在对象监视器中(记录当前线程进入到阻塞状态)//2.释放锁资源(解锁)//3.当前线程进入到阻塞状态phone.wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(phone.getBrand() + " -- " + phone.getPrice());phone.setStore(false);//唤醒：唤醒的是对象监视器中随机一个等待的线程，唤醒的线程进入到就绪态phone.notify();}}}
}

1.2,多个生产者多个消费者

public class Test01 {/*** 知识点：生产者消费者模型 - 多个生产者多个消费者* * 分析：* 		产品类 - Phone：属性(brand,price)* 		生产者线程 - Producer* 		消费者线程 - Consumer* 最终的目的：生产一个、消费一个* * 步骤：* 		1.多个线程(生产者线程、消费者线程)操作同一个资源(产品类的对象)* 		2.多个产品之间来回切换(华为 <--> 小米)* 			null -- 0.0* 			华为 -- 0.0* 			小米 -- 3999* 			华为 -- 1999* 			脏数据的解决思路：加锁* 		3.生产一个消费一个* 			*/public static void main(String[] args) {//brand - null//price - 0.0Phone phone = new Phone();Producer p1 = new Producer(phone);Producer p2 = new Producer(phone);Consumer c1 = new Consumer(phone);Consumer c2 = new Consumer(phone);p1.start();p2.start();c1.start();c2.start();}
}

//生产者线程
public class Producer extends Thread{private Phone phone;public Producer(Phone phone) {this.phone = phone;}@Overridepublic void run() {boolean flag = true;while(true){synchronized(phone){while(phone.isStore()){//有库存//等待//1.将当前线程记录在对象监视器中(记录当前线程进入到阻塞状态)//2.释放锁资源(解锁)//3.当前线程进入到阻塞状态try {phone.wait();} catch (InterruptedException e) {e.printStackTrace();}}if(flag){phone.setBrand("华为");phone.setPrice(3999);}else{phone.setBrand("小米");phone.setPrice(1999);}flag = !flag;phone.setStore(true);//唤醒：唤醒对象监视器中所有的线程phone.notifyAll();}}}
}

//产品类
public class Phone {private String brand;private double price;private boolean store;public Phone() {}public Phone(String brand, double price) {this.brand = brand;this.price = price;}public String getBrand() {return brand;}public void setBrand(String brand) {this.brand = brand;}public double getPrice() {return price;}public void setPrice(double price) {this.price = price;}public boolean isStore() {return store;}public void setStore(boolean store) {this.store = store;}@Overridepublic String toString() {return "Phone [brand=" + brand + ", price=" + price + "]";}
}

//消费者线程
public class Consumer extends Thread{private Phone phone;public Consumer(Phone phone) {this.phone = phone;}@Overridepublic void run() {while(true){synchronized(phone){while(!phone.isStore()){//没有库存try {//等待：//1.将当前线程记录在对象监视器中(记录当前线程进入到阻塞状态)//2.释放锁资源(解锁)//3.当前线程进入到阻塞状态phone.wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(phone.getBrand() + " -- " + phone.getPrice());phone.setStore(false);//唤醒：唤醒对象监视器中所有的线程phone.notifyAll();}}}
}

区别就是只有一个生产者和消费者的时候不需要一直去判断有没有库存,因为生产了才能消费,消费了才生产

二,仓储模型

2.1一个生产者一个消费者

public class Test01 {/*** 知识点：仓储模型 - 一个生产者一个消费者* * 分析：* 		产品类 - Cake（brand、price、datetime）* 		仓库类 - Store（maxCapacity、curCapacity、list）* 		生产者线程 - Producer* 		消费者线程 - Consumer* 		注意：先生产的先卖出 -- 队列模式* * 经验：对象监视器如何选择？* 		锁对象* */public static void main(String[] args) {Store store = new Store();Producer p = new Producer(store);Consumer c = new Consumer(store);p.start();c.start();}
}

import java.util.LinkedList;
//仓库类
public class Store {private static final int DEFAULT_INIT_CAPACITY = 20;private int maxCapacity;private int curCapacity;private LinkedList<Cake> list;public Store() {maxCapacity = DEFAULT_INIT_CAPACITY;list = new LinkedList<>();}public Store(int maxCapacity) {this.maxCapacity = maxCapacity;list = new LinkedList<>();}//入库public synchronized void push(Cake cake){if(curCapacity >= maxCapacity){try {this.wait();} catch (InterruptedException e) {e.printStackTrace();}}list.add(cake);curCapacity++;System.out.println("入库，当前库存为：" + curCapacity);this.notify();}//出库public synchronized Cake pop(){if(curCapacity <= 0){try {this.wait();} catch (InterruptedException e) {e.printStackTrace();}}Cake cake = list.removeFirst();curCapacity--;System.out.println("出库，当前的库存为：" + curCapacity + "，卖出的产品是：" + cake);this.notify();return cake;}}

//生产与存储对象类
public class Cake {private String brand;private double price;private String datetime;public Cake() {}public Cake(String brand, double price, String datetime) {this.brand = brand;this.price = price;this.datetime = datetime;}public String getBrand() {return brand;}public void setBrand(String brand) {this.brand = brand;}public double getPrice() {return price;}public void setPrice(double price) {this.price = price;}public String getDatetime() {return datetime;}public void setDatetime(String datetime) {this.datetime = datetime;}@Overridepublic String toString() {return "Cake [brand=" + brand + ", price=" + price + ", datetime=" + datetime + "]";}
}

import java.time.LocalDateTime;
//生产者类
public class Producer extends Thread{private Store store;public Producer(Store store) {this.store = store;}@Overridepublic void run() {while(true){Cake cake = new Cake("桃李蛋糕", 2.5, LocalDateTime.now().toString());store.push(cake);}}
}

//消费者类
public class Consumer extends Thread{private Store store;public Consumer(Store store) {this.store = store;}@Overridepublic void run() {while(true){store.pop();}}
}

2.2,多个生产者多个消费者

只需把仓库类的出库入库的判断改为while循环即测试类多几个生产,消费线程,其他同上

import java.util.LinkedList;public class Store {private static final int DEFAULT_INIT_CAPACITY = 20;private int maxCapacity;private int curCapacity;private LinkedList<Cake> list;public Store() {maxCapacity = DEFAULT_INIT_CAPACITY;list = new LinkedList<>();}public Store(int maxCapacity) {this.maxCapacity = maxCapacity;list = new LinkedList<>();}//入库public synchronized void push(Cake cake){while(curCapacity >= maxCapacity){try {this.wait();} catch (InterruptedException e) {e.printStackTrace();}}list.add(cake);curCapacity++;System.out.println("入库，当前库存为：" + curCapacity);this.notifyAll();}//出库public synchronized Cake pop(){while(curCapacity <= 0){try {this.wait();} catch (InterruptedException e) {e.printStackTrace();}}Cake cake = list.removeFirst();curCapacity--;System.out.println("出库，当前的库存为：" + curCapacity + "，卖出的产品是：" + cake);this.notifyAll();return cake;}
}

//测试类
public class Test01 {/*** 知识点：仓储模型 - 多个生产者多个消费者* * 分析：* 		产品类 - Cake（brand、price、datetime）* 		仓库类 - Store（maxCapacity、curCapacity、list）* 		生产者线程 - Producer* 		消费者线程 - Consumer* 		注意：先生产的先卖出 -- 队列模式* * 经验：对象监视器如何选择？* 		锁对象* */public static void main(String[] args) {Store store = new Store();Producer p1 = new Producer(store);Producer p2 = new Producer(store);Consumer c1 = new Consumer(store);Consumer c2 = new Consumer(store);p1.start();p2.start();c1.start();c2.start();}
}

三,线程池

//任务类
public class Task implements Runnable{private int num;public Task(int num) {this.num = num;}@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + "去执行任务" + num);}}

1,创建单个线程的线程池

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class Test01 {/*** 知识点：使用JDK1.8自带的线程池*/public static void main(String[] args) {//创建单个线程的线程池ExecutorService pool = Executors.newSingleThreadExecutor();for (int i = 1; i <= 100; i++) {Task task = new Task(i);//将任务提交给线程池pool.execute(task);}//关闭线程池pool.shutdown();}
}

2,创建指定线程个数的线程池

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class Test02 {/*** 知识点：使用JDK1.8自带的线程池*/public static void main(String[] args) {//创建指定线程个数的线程池ExecutorService pool = Executors.newFixedThreadPool(3);for (int i = 1; i <= 100; i++) {Task task = new Task(i);//将任务提交给线程池pool.execute(task);}//关闭线程池pool.shutdown();}
}

3,创建可缓存的线程池

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class Test03 {/*** 知识点：使用JDK1.8自带的线程池*/public static void main(String[] args) {//创建可缓存的线程池(原本线程池中没有线程，有任务就创建线程，60秒工作的线程任务是闲置线程，就回收)ExecutorService pool = Executors.newCachedThreadPool();for (int i = 1; i <= 100; i++) {Task task = new Task(i);//将任务提交给线程池pool.execute(task);}//关闭线程池pool.shutdown();}
}

4,创建延迟任务的线程池

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;public class Test04 {/*** 知识点：使用JDK1.8自带的线程池*/public static void main(String[] args) throws InterruptedException {//创建延迟任务的线程池ScheduledExecutorService pool = Executors.newScheduledThreadPool(3);for (int i = 1; i <= 100; i++) {Task task = new Task(i);//提交任务，设置延迟时间pool.schedule(task, 5, TimeUnit.SECONDS);}//关闭线程池pool.shutdown();}
}

四,自定义线程

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class Test01 {/*** 知识点：自定义线程池*/public static void main(String[] args) {//自定义线程池ThreadPoolExecutor pool = new ThreadPoolExecutor(5, //核心线程数20, //最大线程数60, //闲置时间TimeUnit.SECONDS, //时间单位new ArrayBlockingQueue<>(30), //任务队列 - 有界队列new ThreadFactory() {//自定义线程工程int num = 1;@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r);t.setName("最牛逼的线程" + num);t.setPriority(Thread.MAX_PRIORITY);num++;return t;}}, new RejectedExecutionHandler() {//自定义拒绝策略@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.out.println("傻逼，你提交的任务太快了~~~");}});for (int i = 1; i <= 100; i++) {Task task = new Task(i);//将任务提交给线程池pool.execute(task);}//闲置时间到后，会销毁核心线程的方法//注意：如果连核心线程都销毁了，何谈线程的复用率？pool.allowCoreThreadTimeOut(true);pool.shutdown();}
}

public class Task implements Runnable{private int num;public Task(int num) {this.num = num;}@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + "去执行任务" + num);}
}

.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;public class Test01 {/*** 知识点：自定义线程池*/public static void main(String[] args) {MyThreadPool pool = new MyThreadPool(5, 20, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(30));for (int i = 1; i <= 100; i++) {Task task = new Task(i);//将任务提交给线程池pool.execute(task);}pool.shutdown();}
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class MyThreadPool extends ThreadPoolExecutor{private static final MyThreadFactory myThreadFactory = new MyThreadFactory();private static final MyRejectedExecutionHandler myRejectedExecutionHandler = new MyRejectedExecutionHandler();public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, myThreadFactory, myRejectedExecutionHandler);}public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);}//自定义拒绝策略private static class MyRejectedExecutionHandler implements RejectedExecutionHandler{@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.out.println("哎呀，你提交的任务太快了~~~");}}//自定义线程工程private static class MyThreadFactory implements ThreadFactory{int num = 1;@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r);t.setName("最厉害的线程" + num);t.setPriority(Thread.MAX_PRIORITY);num++;return t;}}}

五,死锁

小结：

• 	1.死锁不会报错，程序会一直抢资源
  

• 	2.尽可能不要锁嵌套
  

public class Test01 {/*** 知识点：死锁* */public static void main(String[] args) {new Thread(new Runnable() {@Overridepublic void run() {synchronized (KuaiZi.a) {try {Thread.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}synchronized (KuaiZi.b) {System.out.println("哲学家1吃饭饭");}}}}).start();new Thread(new Runnable() {@Overridepublic void run() {synchronized (KuaiZi.b) {try {Thread.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}synchronized (KuaiZi.a) {System.out.println("哲学家2吃饭饭");}}}}).start();}
}class KuaiZi{public static Object a = new Object();public static Object b = new Object();
}

六,带返回值得任务类(补Day21)

//注意MyThreadPool是我上面自写的线程池
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
//导入你自己路径上的
import com.qf.thread_pool_03.MyThreadPool;public class Test01 {/*** 需求：计算任务，一个包含了2万个整数的数组，分拆了多个线程来进行并行计算，最后汇总出计算的结果。* * 使用带有返回值的任务类去解决该需求
s	 */public static void main(String[] args) throws InterruptedException, ExecutionException {//创建数组int[] arr = new int[20000];//初始化数组中的元素for (int i = 0; i < arr.length; i++) {arr[i] = i+1;}//创建线程池MyThreadPool pool = new MyThreadPool(4, 4, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(30));//创建任务Task task1 = new Task(arr, 0, 5000);Task task2 = new Task(arr, 5000, 10000);Task task3 = new Task(arr, 10000, 15000);Task task4 = new Task(arr, 15000, 20000);//提交任务，返回Future，任务的返回值就在Future里Future<Integer> future1 = pool.submit(task1);Future<Integer> future2 = pool.submit(task2);Future<Integer> future3 = pool.submit(task3);Future<Integer> future4 = pool.submit(task4);//合并计算System.out.println(future1.get() + future2.get() + future3.get() + future4.get());pool.shutdown();}
}

import java.util.concurrent.Callable;//带有返回值的任务类
public class Task implements Callable<Integer>{private int[] arr;private int startIndex;private int endIndex;public Task(int[] arr, int startIndex, int endIndex) {this.arr = arr;this.startIndex = startIndex;this.endIndex = endIndex;}@Overridepublic Integer call() throws Exception {int sum = 0;for (int i = startIndex; i < endIndex; i++) {sum += arr[i];}return sum;}
}