目录
一、原子操作类
????????java.util.concurrent.atomic包下提供了一系列的原子操作类，主要包含：原子更新基本类型、原子更新数组、原子更新引用、原子更新类的属性。这些类都具有线程安全性，其底层实现基本上使用Unsafe实现的包装类，完成自旋和CAS操作。
1. 类型总结
| 作用 | 实现类 | 特点 |
| 原子更新 基本类型 | AtomicBoolean AtomicInteger AtomicLong | 1. 只能对一个变量进行操作； 2.?AtomicBoolean转换成int型。 |
| 原子更新 数组元素 | AtomicIntegerArray (引用类型数) | 1. 根据索引更新元素。 |
| 原子更新 引用类型 | AtomicReference (更新引用类型) (更新引用类型的属性) (更新有标记位的引用类型) | 1. 一次更新多个变量； 2. 底层实现AtomicInteger一致。 |
| 原子更新 类的属性 | AtomicIntegerFieldUpdater (更新带版本号的引用类型) | 1. 更新类的字段； 2. 使用步骤： ? ? ?step1: 更新器设置类和字段； ? ? ?step2: 该字段必须public volatile修饰。 |
2.?AtomicInteger原理
????????AtomicInteger原理
3. 代码实例
package com.atomic;
import org.junit.jupiter.api.Test;
import java.util.concurrent.atomic.*;
/**
* @description java原子类操作
* @author TCM
* @version 1.0
* @date 2022/4/15 10:59
**/
public class AtomicTest {
AtomicInteger atomicInteger = new AtomicInteger(1);
AtomicBoolean atomicBoolean = new AtomicBoolean();
// 测试原子更新基本类型
@Test
public void atomicIntegerTest(){
// 先获取，再自增+1
System.out.println("atomicInteger.getAndIncrement(): " + atomicInteger.getAndIncrement());
// 懒设置，其他线程一小段时间内可能读到旧值
atomicInteger.lazySet(4);
System.out.println("atomicInteger.get(): " + atomicInteger.get());
System.out.println("atomicBoolean.get()1: " + atomicBoolean.get());
System.out.println("atomicBoolean.getAndSet(): " + atomicBoolean.getAndSet(true));
System.out.println("atomicBoolean.get()2: " + atomicBoolean.get());
}
int[] arr = new int[]{1,2,3,5,6};
AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(arr);
// 测试原子更新数组元素
@Test
public void atomicIntegerArrayTest(){
System.out.println("atomicIntegerArray.get()1: " + atomicIntegerArray.get(0));
// 先获取，再自增
atomicIntegerArray.getAndAdd(0,10);
System.out.println("atomicIntegerArray.get()2: " + atomicIntegerArray.get(0));
}
AtomicReference<User> atomicReference = new AtomicReference<>();
// 测试原子更新引用类型
@Test
public void atomicReferenceTest(){
// 没有设置的提前，获取结果为null
System.out.println("atomicReference.get()1: " + (atomicReference.get() == null ? "":atomicReference.get().getName()));
User user = new User("张三", 25);
// 设置
atomicReference.set(user);
System.out.println("atomicReference.get()2: " + atomicReference.get().getName());
User updateUser = new User("李四", 35);
// 更新
atomicReference.getAndSet(updateUser);
System.out.println("atomicReference.get()3: " + atomicReference.get().getName());
}
private static class User {
private String name;
private int age;
public User(String name, int age){
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public int getAge() {
return age;
}
}
AtomicIntegerFieldUpdater<Person> atomicIntegerFieldUpdater =
AtomicIntegerFieldUpdater.newUpdater(Person.class, "age");
/**
* 更新类的指定字段
* step1：更新器设置类和字段，如：AtomicIntegerFieldUpdater.newUpdater(Person.class, "age")
* step2：该字段必须用public volatile修饰
*/
@Test
public void atomicIntegerFieldUpdaterTest(){
Person person = new Person("张三", 25);
atomicIntegerFieldUpdater.set(person, 26);
System.out.println("atomicIntegerFieldUpdater.get()1: " + atomicIntegerFieldUpdater.get(person));
}
private static class Person {
private String name;
public volatile int age;
public Person(String name, int age){
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public int getAge() {
return age;
}
}
}
二、并发工具类
1. 总结
| 类型 | 实现类 | 特点 |
| 并发流程控制 | CountDownLatch (等待多线程完成) | 1. 作用：允许单或多个线程等待其他线程完成后执行； 2. 过程：step1：调用countDown()，则计数器减1 ? ? ? ? ? ? ? ?step2：计数器=0时，各个线程从await()返回 3. 只有计数器=0时，调用await()才不会阻塞线程； 4.?调用countDown() happens-before?调用await()。 |
| CyclicBarrier (同步屏障) | 1. 作用：可循环使用的线程屏障，即：等待线程都调用await()后，则各个线程才从await()返回； 2.?CountDownLatch只能执行一次，而可以执行多次（即：调用reset()）。 | |
| Semaphore (信号量) | 1. 作用：控制公共资源的访问量。 | |
| 线程间数据传输 | Exchanger (线程间数据交换) | 1. 作用：线程之间交换数据； 2. 交换时机：都调用exchange()才能交换数据。 |
2. 代码实例
package com.atomic;
import org.junit.jupiter.api.Test;
import java.util.concurrent.*;
/**
* @description 并发工具类测试
* @author TCM
* @version 1.0
* @date 2022/4/15 18:03
**/
public class AtomicUtilTest {
CountDownLatch countDownLatch = new CountDownLatch(2);
// 测试CountDownLatch（等待多线程完成）
@Test
public void countDownLatchTest() throws InterruptedException {
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(countDownLatch.getCount());
System.out.println(1);
// 调用一次countDown()，计数器-1
countDownLatch.countDown();
System.out.println(2);
countDownLatch.countDown();
System.out.println("============");
System.out.println(countDownLatch.getCount());
countDownLatch.countDown();
System.out.println(countDownLatch.getCount());
}
}).start();
// 直到计数器=0时，当前线程才从await()返回
countDownLatch.await();
System.out.println(3);
}
// 定义线程屏障数量，及屏蔽过后优先执行的线程
CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new A());
// 测试CyclicBarrier（可循环使用的线程屏障）
@Test
public void cyclicBarrierTest(){
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
System.out.println("屏障之前进行");
cyclicBarrier.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(1);
}
});
thread.start();
// thread.interrupt();
try {
cyclicBarrier.await();
} catch (Exception e) {
System.out.println("isBroken: " + cyclicBarrier.isBroken());
// e.printStackTrace();
}
System.out.println(2);
}
public class A implements Runnable {
@Override
public void run() {
System.out.println("屏障开启后优先执行的线程");
}
}
Semaphore semaphore = new Semaphore(2);
final int THREAD_COUNT = 30;
ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
// 测试Semaphore（信号量，及：控制公共资源访问量）
@Test
public void semaphoreTest(){
for (int i = 0; i < THREAD_COUNT; i++) {
executorService.execute(() -> {
try {
// 当前线程进入到资源
semaphore.acquire();
System.out.println("save data: " + Thread.currentThread());
} catch (Exception e) {
e.printStackTrace();
} finally {
// 当前线程释放资源
semaphore.release();
// 获取所有等待线程的数量
System.out.println("semaphore.getQueueLength(): " + semaphore.getQueueLength());
}
});
}
executorService.shutdown();
System.out.println("main thread end！");
}
Exchanger<String> exchanger = new Exchanger();
ExecutorService executorService2 = Executors.newFixedThreadPool(2);
// 测试Exchanger（线程间数据传输）
@Test
public void exchangerTest(){
executorService2.execute(() -> {
try {
String a = "A";
exchanger.exchange(a);
} catch (Exception e) {
e.printStackTrace();
}
});
executorService2.execute(() -> {
try {
String b = "B";
// 执行交换数据，必须等待另一个线程执行exchange()
String a = exchanger.exchange(b);
System.out.println("是否一致：" + b.equals(a) + ",a=" + a + ",b=" + b);
} catch (Exception e) {
e.printStackTrace();
}
});
System.out.println("main thread end！");
}
}
三、参考资料
AtomicInteger原理_爱我所爱0505的博客-CSDN博客_atomicinteger原理
CountDownLatch详解_西瓜游侠的博客-CSDN博客_countdownlatch
