网站加入百度广告联盟/2021小学生新闻摘抄

一、并发安全、不安全描述

安全：多个线程操作同一个资源，最后的执行结果与单线程执行结果一致，则说明是线程安全的

不安全：多个线程操作同一个资源，最后执行结果不确定的，则说明不是线程安全的

这里我觉得还是解释一下并发与并行的一点区别比好(并非绝对概念)，并发通常是多个线程去竞争相同资源，而并行通常是多个线程之间是协作关系，例如，在秒杀场景下，多个用户(线程)共同争抢某个资源，这个是并发。例如，多个线程统计一个几千万数据的文件，这个时候线程之间是协作关系，每个线程各自统计分配的一段数据，最后汇总给主线程。

二、常见并发模拟工具以及代码模拟并发(工具使用之后再单独学习)

a) Postman ：http 请求模拟工具

b) AB (Apache Bench) :Apache 附带的模拟工具，主要用来测试网站性能

c) JMeter ： Apache 组织开发的压力测试工具

d) 使用 CountDownLatch、Semaphore 进行并发模拟 (在笔记一中已经提到)

三、代码模拟并发

a) CountDownLatch 介绍：该类为一个计数器，字面意思就是向下减的一个闭锁类

上图解释：

     TA 为主线程，主线程初始化 CountDownLatch 计数器为3，T1~3 为子线程，主线程调用CountDownLatch的 await 后就开始阻塞，直到T1~3 调用 CountDownLatch 的 countDown() 方法将计数器减为0，然后主线程继续运行。

b) Semaphore 介绍：

     字面意思是信号量，它可以控制同一时间内有多少个线程可以执行，比如我们常见的马路，有4车道，8车道，可以把这里的车道比作线程，4车道相当于4个线程同时执行，8车道想相当于8个线程执行。semaphore 就是好比是这个车道，可以指定有多个车道，从对信号量的功能描述，可以想到在实际开发中可以用来限制同一时间请求接口的次数，通常semaphore 会与线程池配合使用。

c) 并发模拟代码(Not thread safe)

/*** 并发模拟* * @author Aaron**/
@NotThreadSafe
@Slf4j
public class ConcurrencyTest1 {// 模拟 1000个用户请求private final static int TotalClient = 1000;// 限制同一时间只能有10个线程执行private final static int TotalThread = 10;// 计数器private static int count = 0;public static void main(String[] args) throws InterruptedException {ExecutorService es = Executors.newCachedThreadPool();// 设置信号量，允许同时最多执行的线程数final Semaphore sp = new Semaphore(TotalThread);final CountDownLatch cdl = new CountDownLatch(TotalClient);for (int i = 0; i < TotalClient; i++) {es.execute(new Runnable() {@Overridepublic void run() {try {sp.acquire();add();sp.release();} catch (InterruptedException e) {log.error("A", e);}cdl.countDown();}});}// 中断主线层代码，直至countdownlatch 的计数器变为0cdl.await();es.shutdown();log.info(String.valueOf(count));}@NotThreadSafepublic static void add() {count++;}}

/*** 并发模拟* * @author Aaron**/
@ThreadSafe
@Recommend
@Slf4j
public class ConcurrencyTest2 {// 模拟 1000个用户请求private final static int TotalClient = 1000;// 限制同一时间只能有10个线程执行private final static int TotalThread = 10;// 使用原子类private final static AtomicInteger count = new AtomicInteger(0);public static void main(String[] args) throws InterruptedException {ExecutorService es = Executors.newCachedThreadPool();// 设置信号量，允许同时最多执行的线程数final Semaphore sp = new Semaphore(TotalThread);final CountDownLatch cdl = new CountDownLatch(TotalClient);for (int i = 0; i < TotalClient; i++) {es.execute(new Runnable() {@Overridepublic void run() {try {sp.acquire();add();sp.release();} catch (InterruptedException e) {log.error("A", e);}cdl.countDown();}});}// 中断主线层代码，直至countdownlatch 的计数器变为0cdl.await();es.shutdown();log.info(String.valueOf(count.get()));}@ThreadSafepublic static void add() {count.incrementAndGet();}}

四、类的线程安全性定义

      当多个线程同时访问一个类时，不管运行时环境采用何种方式调用或者这些线程如何交替执行， 并且在主调代码中不需要做额外的同步或协同操作，这个类始终表现出正确的行为，那么这个类就是线程安全的。

五、线程安全的主要体现点

a) 原子性：

    原子性可以解释为互斥性访问，既同一时刻，只能有一个线程进行操作

b) 可见性：

   某个线程对主内存的修改，其它线程必须能及时观察到

c) 有序性：

   某个线程观察其它线程中的指令执行顺序，由于指令重排序的存在，通常观察到的是杂乱无序的

六、CAS 原理 (以 AtomicInteger 为参考)

由于学习发现我的 eclipse 看不到 unsafe 源码，其它源码可以看到，所以特意安装了反编译插件（Decompiler）

地址：https://www.cnblogs.com/godtrue/p/5499785.html

a ) CAS 是 unsafe 中的 compareAndSwapInt 方法的缩写

b) 原理，在 AtomicInteger 的 incrementAndGet 方法里调用了 unsafe 中的 getAndAddInt 方法，在该方法中，核心的方法是 compareAdnSwapInt 方法，核心原理是通过对象以及值的内存地址取出当前值，然后再进行比较，如果比较是值发生了更改则重新取出最新的值再继续比较，直到比较成功，然后更新值。

// 标记为 native 的方法，说明不是用java实现的，通常是由C、C++ 等等实现的
// 第一个参数是当前对象，第二个参数是值所对应的内存地址
public native int getIntVolatile(Object arg0, long arg1);// 也是标记为 native 的方法，也是核心方法
// 第一个参数是当前对象，第二个参数是值所对应的内存地址，第三个参数为内存值，第四个参数为准备更新的值
public final native boolean compareAndSwapInt(Object arg0, long arg1, int arg3, int arg4);// AtomicInteger 中调用的是该方法
// 第一个参数是当前对象，第二个参数是值的内存地址，第三个参数为增加量
public final int getAndAddInt(Object arg0, long arg1, int arg3) {int arg4;do {// 取出当前内存值(预期值)arg4 = this.getIntVolatile(arg0, arg1);// 比较当前内存值是否与预期值相等，如果不相等则继续比较，如果相等则返回当前的内存值// 如果预期值 与 arg1 指向的值一样，则更新为 arg4 + arg3 // 如果不一样则继续循环，直到完成更新} while (!this.compareAndSwapInt(arg0, arg1, arg4, arg4 + arg3));return arg4;
}

c) CAS 缺点

    分析CAS源码后可以发现，如果大量线程进行CAS操作，那么竞争就会很激烈，导致一部分线程由于总是比较失败而长时间停留在循环体中，可能会有瞬间或一段时间的CPU过载，影响系统性能。

d) (JDK1.8 新增 ) LongAdder 与 DoubleAdder 处理思想

     对于普通类型的 long 或 double 类型的变量，JVM 允许将64位的读操作或写操作拆分成两个32位的读写操作，该处理方式的主要思想是将热点数据分离，将内部 Value 分离成一个Cell 数组，当多个线程访问时通过HASH等算法将线程映射到其中一个Cell 上进行操作，最终的计算结果则是Cell 数组的求和值，当低并发的时候，算法会直接更新变量的值，在高并发的时候通过分散操作Cell 提高性能，当然缺点也是有的，当并发更改以及调用sum操作时，sum统计的值可能不准确，以下是原话。

    /*** Returns the current sum.  The returned value is <em>NOT</em> an* atomic snapshot; invocation in the absence of concurrent (意思是在非并发的情况下使用)* updates returns an accurate result, but concurrent updates that* occur while the sum is being calculated might not be* incorporated.** @return the sum*/public long sum() {Cell[] as = cells; Cell a;long sum = base;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;}}return sum;}

七、java.util.concurrent.atomic 包

八、AtomicReference 与 AtomicIntegerFieldUpdater

a) AtomicReference 基本使用

该类提供一个泛型参数，用于对多种对象的原子操作(注意是对象的操作，如果传入原子类，则是对这个原子类本身的原子操作，并非是原子类中数据的原子操作)，以下为简单的示例

@ThreadSafe
@Slf4j
public class AtomicReferenceTest {private static AtomicReference<Integer> ar = new AtomicReference<Integer>(0);public static void main(String[] args) {// 比较更新方法，如果是值是0，则更新为1log.info("{} -> {} - {}", 0, 1, ar.compareAndSet(0, 1));// 获取原先的值，并设置为指定的新值log.info("{} -> {}", ar.getAndSet(3), ar.get());// 以下是源码实现，核心还是使用的Unsafe类的方法// /**// * Atomically sets the value to the given updated value// * if the current value {@code ==} the expected value.// * @param expect the expected value// * @param update the new value// * @return {@code true} if successful. False return indicates that// * the actual value was not equal to the expected value.// */// public final boolean compareAndSet(V expect, V update) {// native 原子方法// return unsafe.compareAndSwapObject(this, valueOffset, expect,// update);// }}
}

b) AtomicIntegerFieldUpdater
基本使用

该类提供一个泛型参数，用于对对象内部的成员变量进行原子操作，以下为简单的示例

@Slf4j
public class AtomicIntegerFieldUpdaterTest {private static AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a = AtomicIntegerFieldUpdater.newUpdater(AtomicIntegerFieldUpdaterTest.class, "value");// 变量必须是 int 基本类型，不能是对象类型// 变量必须有 volatile 关键字修饰// 以下是源码中的判断// if (field.getType() != int.class)// throw new IllegalArgumentException("Must be integer type");//// if (!Modifier.isVolatile(modifiers))// throw new IllegalArgumentException("Must be volatile type");@Getter// 未初始化默认是0private volatile int value;public static void main(String[] args) {

         AtomicIntegerFieldUpdaterTest aifu = new AtomicIntegerFieldUpdaterTest();// 比较&设置a.compareAndSet(aifu, 0, 2);log.info("{}", aifu.getValue());
}

a) ABA问题解释：

     当多个线程操作一个资源时，T1取出值A，T2线程也取出值A，这个时候T2执行过程中将A变为B又变回A，然后T1继续执行发现与自己的值相同，然后进行了更新操作，操作虽然成功，但是这个过程却是有隐患的，比如对一个单项链表操作，T1 取出栈顶A与下一个栈B，想用CAS替换栈顶A为B，在T1执行CAS操作之前，这时候T2取出A和B，然后push了A、C、D，这个时候B属于独立的链表，处于游离状态(当前有两个链表 A->C->D->NULL ,还有一个游离的 B->NULL)，然后T1开始执行CAS操作，发现ACD栈顶还是A，然后开始处理，由于之前已经取出B，当时的B->NULL 这样的，最后结果是把T2的 CD 给丢了。

b) 为了解决ABA问题，有了 AtomicStampedReference 这个类

    该类的核心思想是，每次操作都有个一 version 来记录，例如 T2 取出A时版本是 1，更新为B后版本号变为2，再更新为A时版本号变为3，此时由于有版本号控制，T1 再来更新A时发现自己的版本号1 与 3 不一致，最后CAS操作失败。

示例：

@Slf4j
public class ABATest {// 普通原子类private static AtomicInteger atomicInt = new AtomicInteger(100);// 有版本号的实现(参数是初始值与初始版本号)private static AtomicStampedReference<Integer> atomicStampedRef = new AtomicStampedReference<Integer>(100, 0);public static void main(String[] args) throws InterruptedException {// 模拟 B->AThread intT1 = new Thread(new Runnable() {@Overridepublic void run() {// A->BatomicInt.compareAndSet(100, 101);// B->AatomicInt.compareAndSet(101, 100);}});// 模拟 A->BThread intT2 = new Thread(new Runnable() {@Overridepublic void run() {try {// 线程休眠，给 T1 执行TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// A->Bboolean c3 = atomicInt.compareAndSet(100, 101);log.info("一般 CAS={}", c3);// 操作成功}});intT1.start();intT2.start();intT1.join();intT2.join();Thread refT1 = new Thread(new Runnable() {@Overridepublic void run() {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// A-B 版本号 1atomicStampedRef.compareAndSet(100, 101, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);// B-A 版本号 2atomicStampedRef.compareAndSet(101, 100, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);}});Thread refT2 = new Thread(new Runnable() {@Overridepublic void run() {// T2取出版本号int stamp = atomicStampedRef.getStamp();log.info("有版本号，线程休眠之前:stamp={}", stamp);try {// 休眠2秒TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}// T1 的版本号log.info("有版本号，线程休眠之后:stamp={}", atomicStampedRef.getStamp());// A->B ，T1 将版本号增加到了2，然后执行时由于T2// 持有的版本号还是之前的0，与当前的版本号2不一致，最中CAS操作失败boolean c3 = atomicStampedRef.compareAndSet(100, 101, stamp, stamp + 1);log.info("有版本号 CAS={}", c3);}});refT1.start();refT2.start();}

十、 AtomicBoolean 示例

可以使用该类来保证某项操作只执行一次

@Slf4j
public class AtomicBooleanTest {private static AtomicBoolean ab = new AtomicBoolean(true);public static void main(String[] args) throws InterruptedException {ExecutorService es = Executors.newCachedThreadPool();int c = 1000;int s = 100;final Semaphore sh = new Semaphore(s);final CountDownLatch cdl = new CountDownLatch(c);for (int i = 0; i < c; i++) {es.execute(new Runnable() {@Overridepublic void run() {try {sh.acquire();init();sh.release();} catch (InterruptedException e) {log.error("Error", e);}cdl.countDown();}});}cdl.await();es.shutdown();}private static void init() {// 个人理解这么写会有效率问题，因为每次都要进行CAS比较,应该加一层 if 判断,如 init2if (ab.compareAndSet(true, false)) {log.info(".......init.....OK");}}private static void init2() {if (ab.get()) {if (ab.compareAndSet(true, false)) {log.info(".......init.....OK");}}}
}