Java并发编程

秒速五厘米 2022-01-25 20:37 288阅读 0赞

**目录**

线程安全性

原子性

CAS

synchronized

可见性

Java内存模型（JMM）

volatile

synchronized

有序性

happens-before

volatile-禁止指令重排序

synchronized

AQS常用类

CountDownLatch

Semaphore

CyclicBarrier

ReentrantLock

线程池

ThreadPoolExecutor核心参数

拒绝策略

常用方法

常用线程池

配置线程池

--------------------

# 线程安全性 #

## 原子性 ##

一个操作是不可被中断的。

### CAS ###

compare and swap 判断当前线程工作内存的值和主内存的值是否相同，相同就改变数值，如果不同就讲工作内存的值刷新回主内存，不需要加锁，根据下面的源码也可以得出这个结论。

java.util.concurrent.atomic包下的类

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70][]

是线程安全的，i++在多线程情况下是线程不安全的，但是AtomicInteger 是线程安全的。

AtomicInteger

cas(compareAndSwap)

>     public static  AtomicInteger atomicInteger = new AtomicInteger(0);
>     
>      public static void main(String[] args) {
>          atomicInteger.incrementAndGet(); // 类似于i++每次加1

原理？

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70 1][]

因为以上的源代码，在每次执行加1操作时都要比较工作内存和共享内存的一致性，不同就不加1（在i++）的操作中从而在多线程中保证安全。

AtomicBoolean

cas(compareAndSet)

>     public static AtomicBoolean aBoolean = new AtomicBoolean();
>     
>      public static void main(String[] args) {
>          aBoolean.compareAndSet(false,false);
>      }

![20190528091943277.png][]

AtomicLong 和LongAdder（新增的）类似 。

CAS的缺点

1.  循环时间开销很大
2.  只能保证一个共享变量的原子操作

### synchronized ###

（实例锁）对象锁

>     // 修饰一个代码块
>     public void test1(int j) {
>             synchronized (this) { 
>             }
>         }
>     
>         // 修饰一个方法
>         public synchronized void test2(int j) {
>         }

当只有一个线程对象时，同时调用两个方法得等一个先执行完另一个才能执行。

public static void main(String[] args) {
            SynchronizedExample1 example1 = new SynchronizedExample1();
            SynchronizedExample1 example2 = new SynchronizedExample1();
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(() -> {
                example1.test2(1);
            });
            executorService.execute(() -> {
                example2.test2(2);
            });
        }

当有两个线程同时调用上面的方法时会出现交替输出的现象，因为是对象锁，每个线程都是一个对象，都会获得锁，两个线程互不干扰。

![image.png][]

Class对象（全局锁）（类锁）

当用static修饰时说明是类锁，整个类的所有对象共有一把锁。

// 修饰一个类
        public static void test1(int j) {
            synchronized (SynchronizedExample2.class) {
                for (int i = 0; i < 10; i++) {
                    log.info("test1 {} - {}", j, i);
                }
            }
        }
    
        // 修饰一个静态方法
        public static synchronized void test2(int j) {
            for (int i = 0; i < 10; i++) {
                log.info("test2 {} - {}", j, i);
            }
        }

public static void main(String[] args) {
            SynchronizedExample2 example1 = new SynchronizedExample2();
            SynchronizedExample2 example2 = new SynchronizedExample2();
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(() -> {
                example1.test1(1);
            });
            executorService.execute(() -> {
                example2.test1(2);
            });
        }

尽管上面是两个对象，但是由于此时共有一把锁，所以是example1的方法先执行完，example2的方法再执行。

![image.png][image.png 1]

## 可见性 ##

一个线程对主内存修改可以被其他线程知道。

什么是共享变量？一个变量可以被多个线程同时访问就叫做共享变量通常是全局变量（成员变量、实例变量）属于类的。

导致共享变量在线程间不可见的原因有哪些？

1.由于线程交叉执行，线程A读取到值2进行加1刷新回主内存同时线程B也读取到之前的值2然后加1，然后刷新回主内存。

2.共享变量值修改后没有及时刷新回主内存。

### Java内存模型（JMM） ###

JMM通过在变量读取前从主内存读取最新的值，变量修改后将最新的值刷新回主内存。这种依赖主内存作为传递媒介的方式实现可见性。

![image.png][image.png 2]

### volatile ###

关键字实现可见性，保证读取到的值一直是最新的。但是volatile操作不具备原子性。结合java内存模型来说。

线程A从共享内存中读取到最新的值进行修改，同时线程B由于可见性也从共享内存中读取到最新的值，然后执行操作进行+1将最新的值刷新回主内存，此时在主内存中存放的值是2，因为两个线程刷新回主内存的值都是2 很明显就不是最新的了。

### synchronized ###

同一个时间只有一个线程，加锁前从主内存同步最新的值，解锁前将最新的值刷新回主内存。（加锁和解锁是同一把锁）

## 有序性 ##

### happens-before ###

保证写在前面的先执行，写在后面的后执行。要实现这个得让上下代码有依赖关系

int a = 1;  
int b = a+1;

### volatile-禁止指令重排序 ###

每次读写操作后都有一个屏障从而禁止了重排序。

### synchronized ###

每次只有一个线程执行，根据happes-before原则，一个线程内是有序的。

# AQS常用类 #

AQS(AbstractQueuedSynchronizer)

## CountDownLatch ##

等待子线程执行完之后再执行主线程，构造函数里面是一共要执行的线程数量

核心方法 countDownLatch.countDown(); countDownLatch.await();

>     public static void main(String[] args) throws Exception {
>     
>            CountDownLatch countDownLatch = new CountDownLatch(5);
>            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 300, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>());
>     
>             for (int i = 0; i < 5; i++) {
>                 Thread thread = new Thread(new Runnable() {
>                     @Override
>                     public void run() {
>                         // 业务
>                         
>                         countDownLatch.countDown();
>                     }
>                 });
>     
>                 threadPoolExecutor.execute(thread);
>     
>             }
>             countDownLatch.await();  // 执行主线程操作
>             // 以下是主线程操作
>             threadPoolExecutor.shutdown(); // 线程池使用完之后一定要关闭
>         }

## Semaphore ##

控制可以并行执行的线程数，核心方法acquire(),release()；

>     public static void main(String[] args) throws Exception {
>     
>             Semaphore semaphore = new Semaphore(2);  // 并行2个线程
>             ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 300, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>());
>     
>             for (int i = 0; i < 5; i++) {
>                 Thread thread = new Thread(new Runnable() {
>                     @Override
>                     public void run() {
>                         try {
>                             semaphore.acquire();   // 同一时间就有2个线程获得许可
>                         } catch (InterruptedException e) {
>                             e.printStackTrace();
>                         }
>                         // 业务
>                         semaphore.release();
>                     }
>                 });
>     
>                 threadPoolExecutor.execute(thread);
>     
>             }
>             threadPoolExecutor.shutdown();   // 线程池使用完之后一定要关闭
>           
>         }

## CyclicBarrier ##

一个线程执行完阻塞，当所有线程都执行完之后再共同执行接下来的操作

>     public static void main(String[] args) throws Exception {
>     
>             CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
>             ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 300, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>());
>     
>             for (int i = 0; i < 5; i++) {
>                 Thread thread = new Thread(new Runnable() {
>                     @Override
>                     public void run() {
>                         try {
>                             // 一个线程执行完之后阻塞,数量加1
>                             cyclicBarrier.await();
>                             // 当所有线程都执行完之后再往下执行,当参与者数量为5时往下执行
>                             System.out.println("continued");
>                         } catch (InterruptedException e) {
>                             e.printStackTrace();
>                         } catch (BrokenBarrierException e) {
>                             e.printStackTrace();
>                         }
>                         
>                     }
>                 });
>     
>                 threadPoolExecutor.execute(thread);
>     
>             }
>             threadPoolExecutor.shutdown();
>     
>         }

## ReentrantLock ##

> lock(); unlock();
> 
> 等待可中断，当一个持有锁的线程长期不释放锁的时候，等待锁的线程可以放弃等待。
> 
> 公平锁，默认的构造函数为非公平锁，通过参数true设置
> 
> 分组唤醒需要唤醒的线程 condition

# 线程池 #

## ThreadPoolExecutor核心参数 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70 3][]

## 拒绝策略 ##

当当前线程数>最大线程数时会执行

1.丢弃任务不抛异常 DiscardPolicy

2.丢弃任务并抛出异常 AbortPolicy（默认）

3.丢弃最前面的任务，执行后面的任务 DiscardOldestPolicy

4.由调用的线程执行该任务 CallerRunsPolicy

## 常用方法 ##

> execute() 向线程池提交一个任务
> 
> submit() 能返回任务执行的结果，execute + future
> 
> shutdown() 关闭线程执行完当前任务
> 
> shutdownNow() 关闭线程不执行完当前任务

## 常用线程池 ##

>     1.ExecutorService executorService = Executors.newCachedThreadPool();  核心线程数为0，来了线程就执行，当线程空闲时间超过60s，自动销毁
> 
> 2.ExecutorService executorService = Executors.newFixedThreadPool(int nThreads);       固定长度的线程池，无超时时间，不会自动销毁
> 
> 3.ExecutorService executorService = Executors.newSingleThreadExecutor(); 创建干戈工作线程来执行任务
> 
> 4.ExecutorService executorService = Executors.newScheduledThreadPool(int corePoolSize); 创建固定长度的线程池，可以以延迟或定时的方式来执行任务

## 配置线程池 ##

CPU密集型 ，CPU的数量+1个线程

IO密集型，CPU数量\*2

实际情况还是需要根据压测来算

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70]: /images/20220126/b15efa7d1c1f4152add85eae69e847cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70 1]: /images/20220126/3aea83f92bcc4186866c2796762cc65e.png
[20190528091943277.png]: /images/20220126/bc369e4d35fb4dacb9cf8f94e3870b85.png
[image.png]: /images/20220126/7928cbdb73424103b2670c4a8150a6e1.png
[image.png 1]: /images/20220126/cb62b7393c8748048b3fde87f1819974.png
[image.png 2]: /images/20220126/12146f00a6f1485b8bf01d343b807282.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70 2]: /images/20220126/97def3c1a0fa47dda4bc19db2469fe39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pbmdtZW5nYmFieQ_size_16_color_FFFFFF_t_70 3]: /images/20220126/d42741702fdf466dbaab5309c03a8325.png