线程和进程
一个进程往往可以包含多个线程,至少包含一个!
Java默认有几个线程?
2 个: mian、GC
对于Java而言:Thread、Runnable、Callable 三种实现线程的方式。
JAVA不能开启线程,是调用本地方法,查看start方法可以看到底层是C++来开启线程的
并发和并行
- 并发(多线程操作同一个资源):CPU 一核 ,模拟出来多条线程,天下武功,唯快不破,快速交替。
- 并行(同时多条线程同时执行):CPU 多核 ,多个线程可以同时执行; 线程池。
并发编程的本质:充分利用CPU的资源
线程的6种状态
public enum State {
// 新生
NEW,
// 运行
RUNNABLE,
// 阻塞
BLOCKED,
// 等待,死死地等
WAITING,
// 超时等待
TIMED_WAITING,
// 终止
TERMINATED;
}
wait和sleep的区别
- 来自不同的类 wait => Object sleep => Thread
- 关于锁的释放 wait 会释放锁,sleep 睡觉了,抱着锁睡觉,不会释放!
- 范围不同:wait必须在同步代码块,sleep可以任意地方。
synchronized和lock
synchronized方式上锁
/**
* 真正的多线程开发,公司中的开发,降低耦合性
* 线程就是一个单独的资源类,没有任何附属的操作!
* 1、 属性、方法
*/
public class SaleTicketDemo01 {
public static void main(String[] args) {
// 并发:多线程操作同一个资源类, 把资源类丢入线程
Ticket ticket = new Ticket();
// @FunctionalInterface 函数式接口,jdk1.8 lambda表达式 (参数)->{ 代码 }
new Thread(()->{
for (int i = 1; i < 40 ; i++) {
ticket.sale();
}
},"A").start();
new Thread(()->{
for (int i = 1; i < 40 ; i++) {
ticket.sale();
}
},"B").start();
new Thread(()->{
for (int i = 1; i < 40 ; i++) {
ticket.sale();
}
},"C").start();
}
}
// 资源类 OOP
class Ticket {
// 属性、方法
private int number = 30;
// 卖票的方式
// synchronized 本质: 队列,锁
public synchronized void sale(){
if (number>0){
System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余:"+number);
}
}
}
lock方式上锁
public class SaleTicketDemo02 {
public static void main(String[] args) {
// 并发:多线程操作同一个资源类, 把资源类丢入线程
Ticket2 ticket = new Ticket2();
// @FunctionalInterface 函数式接口,jdk1.8 lambda表达式 (参数)->{ 代码 }
new Thread(()->{for (int i = 1; i < 40 ; i++) ticket.sale();},"A").start();
new Thread(()->{for (int i = 1; i < 40 ; i++) ticket.sale();},"B").start();
new Thread(()->{for (int i = 1; i < 40 ; i++) ticket.sale();},"C").start();
}
}
// Lock三部曲
// 1、 new ReentrantLock();
// 2、 lock.lock(); // 加锁
// 3、 finally=> lock.unlock(); // 解锁
class Ticket2 {
// 属性、方法
private int number = 30;
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(); // 解锁
}
}
}
两者的区别
- Synchronized 内置的Java关键字, Lock 是一个Java类
- Synchronized 无法判断获取锁的状态,Lock 可以判断是否获取到了锁
- Synchronized 会自动释放锁,lock 必须要手动释放锁!如果不释放锁,死锁
- Synchronized 线程 1(获得锁,阻塞)、线程2(等待,傻傻的等);Lock锁就不一定会等待下去;
- Synchronized 可重入锁,不可以中断的,非公平;Lock ,可重入锁,可以 判断锁,非公平(可以 自己设置);
- Synchronized 适合锁少量的代码同步问题,Lock 适合锁大量的同步代码
生产者消费者模式两种实现
await 和 notify的实现方式
public class B {
public static void main(String[] args) {
Data2 data = new Data2();
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.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"B").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();
}
}
},"D").start();
}
}
// 判断等待,业务,通知
class Data2{ // 数字 资源类
private int number = 0;
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
//condition.await(); // 等待
//condition.signalAll(); // 唤醒全部
//+1
public void increment() throws InterruptedException {
lock.lock();
try {
// 业务代码
while (number!=0){ //0
// 等待
condition.await();
}
number++;
System.out.println(Thread.currentThread().getName()+"=>"+number);
// 通知其他线程,我+1完毕了
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
//-1
public synchronized void decrement() throws InterruptedException {
lock.lock();
try {
while (number==0){ // 1
// 等待
condition.await();
}
number--;
System.out.println(Thread.currentThread().getName()+"=>"+number);
// 通知其他线程,我-1完毕了
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
用JUC实现
public class C {
public static void main(String[] args) {
Data3 data = new Data3();
new Thread(()->{
for (int i = 0; i <10 ; i++) {
data.printA();
}
},"A").start();
new Thread(()->{
for (int i = 0; i <10 ; i++) {
data.printB();
}
},"B").start();
new Thread(()->{
for (int i = 0; i <10 ; i++) {
data.printC();
}
},"C").start();
}
}
class Data3{ // 资源类 Lock
//几个线程就用几个Condition 精准的通知和唤醒线程
private Lock lock = new ReentrantLock();
private Condition condition1 = lock.newCondition();
private Condition condition2 = lock.newCondition();
private Condition condition3 = lock.newCondition();
private int number = 1; // 1A 2B 3C
public void printA(){
lock.lock();
try {
// 业务,判断-> 执行-> 通知
while (number!=1){
// 等待
condition1.await();
}
System.out.println(Thread.currentThread().getName()+"=>AAAAAAA");
// 唤醒,唤醒指定的人,B
number = 2;
condition2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printB(){
lock.lock();
try {
// 业务,判断-> 执行-> 通知
while (number!=2){
condition2.await();
}
System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
// 唤醒,唤醒指定的人,c
number = 3;
condition3.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printC(){
lock.lock();
try {
// 业务,判断-> 执行-> 通知
// 业务,判断-> 执行-> 通知
while (number!=3){
condition3.await();
}
System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
// 唤醒,唤醒指定的人,c
number = 1;
condition1.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
使用JUC的集合
原生的List,Map,Set的实现类方法在多线程会出现不同步的问题,所以有以下解决的方案:
List
- List
list = new Vector<>(); 其实Vector底层也就是在add方法上加了synchronized。 -
List
list = Collections.synchronizedList(new ArrayList<> ()); -
List
list = new CopyOnWriteArrayList<>(); (推荐使用这个,效率高):底层使用了可重入锁Lock,效率更高。
Set
- Set
set = Collections.synchronizedSet(new HashSet<>()); 使用这个锁的就是整个对象。 - Set
set = new CopyOnWriteArraySet<>();(推荐使用这个,效率高):底层使用了可重入锁Lock,效率更高。
Map
- Map
map= Collections.synchronizedMap(new HashMap<>()); - Map
map = new ConcurrentHashMap<>();(推荐使用这个~)
使用Callable
- 执行相同的任务有存在缓存,效率高。
- 结果可能需要等待,会阻塞!
public class CallableTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// new Thread(new Runnable()).start();
// new Thread(new FutureTask<V>()).start();
// new Thread(new FutureTask<V>( Callable )).start();
new Thread().start(); // 怎么启动Callable
MyThread thread = new MyThread();
FutureTask futureTask = new FutureTask(thread); // 适配类
new Thread(futureTask,"A").start();
new Thread(futureTask,"B").start(); // 结果会被缓存,效率高
Integer o = (Integer) futureTask.get(); //这个get 方法可能会产生阻塞!把他放到最后
// 或者使用异步通信来处理!
System.out.println(o);
}
}
class MyThread implements Callable<Integer> {
@Override
public Integer call() {
System.out.println("call()"); // 会打印几个call
// 耗时的操作
return 1024;
}
}
GUC种的三大辅助类
CountDownLatch(减法计数器)
countDownLatch.countDown();
// 数量-1countDownLatch.await();
// 等待计数器归零,然后再向下执行- 每次有线程调用 countDown() 数量-1,假设计数器变为0,
countDownLatch.await()
就会被唤醒,继续执行
// 计数器
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
// 总数是6,必须要执行任务的时候,再使用!
CountDownLatch countDownLatch = new CountDownLatch(6);
for (int i = 1; i <=6 ; i++) {
new Thread(()->{
System.out.println(Thread.currentThread().getName()+" Go out");
countDownLatch.countDown(); // 数量-1
},String.valueOf(i)).start();
}
countDownLatch.await(); // 等待计数器归零,然后再向下执行
System.out.println("Close Door");
}
}
CyclicBarrier (加法计数器)
只有等待线程数达到特定值才会结束。
public class CyclicBarrierDemo {
public static void main(String[] args) {
/**
* 集齐7颗龙珠召唤神龙
*/
// 召唤龙珠的线程
CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
System.out.println("召唤神龙成功!");
});
for (int i = 1; i <=7 ; i++) {
final int temp = 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(信号量)
规定有几个线程就最多只能有几个线程执行。
- semaphore.acquire();
获得,假设如果已经满了,等待,等待被释放为止
- semaphore.release();
释放,会将当前的信号量释放 + 1,然后唤醒等待的线程!
作用: 多个共享资源互斥的使用!并发限流,控制大的线程数!
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(); // release() 释放
}
},String.valueOf(i)).start();
}
}
}
读写锁
ReadLock
读锁,多线程可同时占有执行。
writeLock()
写锁,多线程中只能一个线程占有执行。
/**
* 独占锁(写锁) 一次只能被一个线程占有
* 共享锁(读锁) 多个线程可以同时占有
* ReadWriteLock
* 读-读 可以共存!
* 读-写 不能共存!
* 写-写 不能共存!
*/
public class ReadWriteLockDemo {
public static void main(String[] args) {
MyCache myCache = new MyCache();
// 写入
for (int i = 1; i <= 5 ; i++) {
final int temp = i;
new Thread(()->{
myCache.put(temp+"",temp+"");
},String.valueOf(i)).start();
}
// 读取
for (int i = 1; i <= 5 ; i++) {
final int temp = i;
new Thread(()->{
myCache.get(temp+"");
},String.valueOf(i)).start();
}
}
}
// 加锁的
class MyCacheLock{
private volatile Map<String,Object> map = new HashMap<>();
// 读写锁: 更加细粒度的控制
private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private Lock lock = new ReentrantLock();
// 存,写入的时候,只希望同时只有一个线程写
public void put(String key,Object value){
readWriteLock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName()+"写入"+key);
map.put(key,value);
System.out.println(Thread.currentThread().getName()+"写入OK");
} catch (Exception e) {
e.printStackTrace();
} finally {
readWriteLock.writeLock().unlock();
}
}
// 取,读,所有人都可以读!
public void get(String key){
readWriteLock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName()+"读取"+key);
Object o = map.get(key);
System.out.println(Thread.currentThread().getName()+"读取OK");
} catch (Exception e) {
e.printStackTrace();
} finally {
readWriteLock.readLock().unlock();
}
}
}
/**
* 自定义缓存
*/
class MyCache{
private volatile Map<String,Object> map = new HashMap<>();
// 存,写
public void put(String key,Object value){
System.out.println(Thread.currentThread().getName()+"写入"+key);
map.put(key,value);
System.out.println(Thread.currentThread().getName()+"写入OK");
}
// 取,读
public void get(String key){
System.out.println(Thread.currentThread().getName()+"读取"+key);
Object o = map.get(key);
System.out.println(Thread.currentThread().getName()+"读取OK");
}
}
阻塞队列
- 抛出异常,add==>remove
public static void test1(){ // 队列的大小 ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3); System.out.println(blockingQueue.add("a")); System.out.println(blockingQueue.add("b")); System.out.println(blockingQueue.add("c")); // IllegalStateException: Queue full 抛出异常! // System.out.println(blockingQueue.add("d")); System.out.println("=-==========="); System.out.println(blockingQueue.element()); // 查看队首元素是谁 System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); // java.util.NoSuchElementException 抛出异常! // System.out.println(blockingQueue.remove()); }
- 有返回值,无抛出异常 offer==>poll peel(获取队首元素)
public static void test2(){
// 队列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.offer("a"));
System.out.println(blockingQueue.offer("b"));
System.out.println(blockingQueue.offer("c"));
System.out.println(blockingQueue.peek());
// System.out.println(blockingQueue.offer("d")); // false 不抛出异常!
System.out.println("============================");
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll()); // null 不抛出异常!
}
3. 死死的阻塞等待 put==>take
public static void test3() throws InterruptedException {
// 队列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
// 一直阻塞
blockingQueue.put("a");
blockingQueue.put("b");
blockingQueue.put("c");
// blockingQueue.put("d"); // 队列没有位置了,一直阻塞
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take()); // 没有这个元素,一直阻塞
}
- 等待,阻塞(设置等待超时)依旧是offer==>poll
public static void test4() throws InterruptedException {
// 队列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
blockingQueue.offer("a");
blockingQueue.offer("b");
blockingQueue.offer("c");
// blockingQueue.offer("d",2,TimeUnit.SECONDS); // 等待超过2秒就退出
System.out.println("===============");
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
blockingQueue.poll(2,TimeUnit.SECONDS); // 等待超过2秒就退出
}
同步队列
- 和其他的BlockingQueue 不一样, SynchronousQueue 不存储元素
- put了一个元素,必须从里面先take取出来,否则不能在put进去值!意思就是队列只能存一个值,取一个值。
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new SynchronousQueue<>(); // 同步队列
new Thread(()->{
try {
System.out.println(Thread.currentThread().getName()+" put 1");
blockingQueue.put("1");
System.out.println(Thread.currentThread().getName()+" put 2");
blockingQueue.put("2");
System.out.println(Thread.currentThread().getName()+" put 3");
blockingQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
},"T1").start();
new Thread(()->{
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
},"T2").start();
}
线程池
线程池的好处: 1. 降低资源的消耗 2. 提高响应的速度 3. 方便管理。
线程复用、可以控制最大并发数、管理线程
三大创建方法
ExecutorService threadPool = Executors.newSingleThreadExecutor();
// 单个线 程ExecutorService threadPool = Executors.newFixedThreadPool(10);
// 创建一 个固定的线程池的大小ExecutorService threadPool = Executors.newCachedThreadPool();
// 由执行的线程数决定大小。
具体过程:创建线程池对象--->执行---->关闭线程池
try {
for (int i = 1; i <= 10; i++) {
// 使用了线程池之后,使用线程池来创建线程
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 线程池用完,程序结束,关闭线程池
threadPool.shutdown();
}
七大参数
一般开发中使用线程池最好不使用Executors去创建对象。如阿里巴巴开发手册所说的:OOM(内存溢出)
所以应该使用ThreadPoolExecutor来创建线程池
查看源码得到以下:
- 当执行线程数大于核心线程锁时,其它休眠的线程位置将会启用,最多不超过最大线程数。
- 当超过最大线程锁的线程任务会在阻塞队列中进行等待。
- 当此时并发执行的线程任务超过了最大线程数+阻塞队列长度时,拒绝策略起效。
拒绝策略有四种
new ThreadPoolExecutor.AbortPolicy()
// 同时并发的线程数大于最大线程数+阻塞队列长度时,不处理超过部分的线程任务,直接抛出异常。new ThreadPoolExecutor.DiscardPolicy()
//同时并发的线程数大于最大线程数+阻塞队列长度时,直接丢掉超过部分的线程任务,不抛出异常。new ThreadPoolExecutor.CallerRunsPolicy()
// 同时并发的线程数大于最大线程数+阻塞队列长度时,超过部分的线程任务回归main主线程执行。new ThreadPoolExecutor.DiscardOldestPolicy()
//阻塞队列满了,尝试去和最先执行的线程发起竞争,也不会抛出异常!
public ThreadPoolExecutor(int corePoolSize, //核心线程数
int maximumPoolSize, //最大线程数。最多几个线程并发。
long keepAliveTime, //当线程无任务时,几秒后结束该线程
TimeUnit unit,//线程结束的时间单位
BlockingQueue<Runnable> workQueue,//阻塞队列,限制等候线程数
ThreadFactory threadFactory,//线程工厂
RejectedExecutionHandler handler)//拒绝策略
例子如下:
ExecutorService threadPool = new ThreadPoolExecutor(
2,
Runtime.getRuntime().availableProcessors(),
3,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(3),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.DiscardOldestPolicy()); //队列满了,尝试去和最早的竞争,也不会抛出异常!
如何选择线程池的大小
-
IO 密集型(多数据多写操作) :判断你程序中十分耗IO的线程,有n个,线程池就设置(2n)个,根据他人经验,应该时cpu核数/0.1 也就是(10n)个。
-
CPU 密集型 :一般电脑几核,线程池就设置几个线程,可以保持cpu的效率最高!可以利用
Runtime.getRuntime().availableProcessors()
获取当前主机的核数。