java 并发编程学习笔记（三）之 线程安全性
【摘要】 线程安全性
（1）java.util.concurrent.atomic 包
AtomicInteger 、 AtomicLong 、LongAdder的使用
@Slf4j@ThreadSafepublic class AtomicExample2 { //请求次数 pri...
线程安全性
（1）java.util.concurrent.atomic 包
- AtomicInteger 、 AtomicLong 、LongAdder的使用
-
@Slf4j
-
@ThreadSafe
-
public class AtomicExample2 {
-
-
//请求次数
-
private static int clientTotal = 5000;
-
-
//允许同时运行的线程数
-
private static int threadTotal = 200;
-
-
//计数
-
//public static AtomicInteger count = new AtomicInteger(0);
-
//public static AtomicLong count = new AtomicLong(0);
-
/**
-
* 一个或多个变量一起保持初始为零long和。 当跨线程争用更新（方法add(long) ）时，
-
* 该变量集可以动态增长以减少争用。 方法sum() （或等效地， longValue() ）返回保持总和的整个变量组合的当前总和。
-
* 当多个线程更新用于诸如收集统计信息，而不是细粒度同步控制的常用总和时，此类通常优于AtomicLong 。
-
* 在低更新争议下，这两类具有相似的特征。 但是，在高度争议的情况下，这一类的预期吞吐量明显高于牺牲更高的空间消耗。
-
*
-
* LongAdders可以使用ConcurrentHashMap来维护可扩展的频率映射（直方图或多集的形式）。
-
* 例如，要将一个计数添加到ConcurrentHashMap<String,LongAdder> freqs ，如果尚未存在，
-
* 则可以使用freqs.computeIfAbsent(key, k -> new LongAdder()).increment();
-
*/
-
public static LongAdder count = new LongAdder();
-
public static void main(String[] args) {
-
ExecutorService executorService = Executors.newCachedThreadPool();
-
final Semaphore semaphore = new Semaphore(threadTotal);
-
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
-
for (int i = 0; i < clientTotal; i++) {
-
executorService.execute(() -> {
-
try {
-
semaphore.acquire();
-
add();
-
semaphore.release();
-
} catch (InterruptedException e) {
-
log.error("exception", e);
-
}
-
countDownLatch.countDown();
-
});
-
-
}
-
try {
-
countDownLatch.await();
-
} catch (InterruptedException e) {
-
e.printStackTrace();
-
}
-
executorService.shutdown();
-
log.info("count:{}", count);
-
}
-
private static void add() {
-
count.increment();
-
}
-
}
通过查看AtomicInteger源码，可以看到AtomicInteger以原子方式更新数值，确保了线程安全性。
-
/**
-
* Atomically increments by one the current value.
-
*
-
* @return the updated value
-
*/
-
public final int incrementAndGet() {
-
return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
-
}
-
-
-
-
-
public final int getAndAddInt(Object var1, long var2, int var4) {
-
int var5;
-
do {
-
var5 = this.getIntVolatile(var1, var2);
-
} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
-
-
return var5;
-
}
- AtomicReference 、AtomicIntegerFieldUpdater、AtomicBoolean的使用
-
@Slf4j
-
@ThreadSafe
-
public class AtomicExample3 {
-
-
@Data
-
static class Student {
-
private int age=0 ;
-
public Student(int age) {
-
this.age = age;
-
}
-
}
-
-
//请求次数
-
private static AtomicReference<Integer> count =new AtomicReference<Integer>(0);
-
private static AtomicReference<Student> studentAtomic =new AtomicReference<Student>(new Student(0));
-
-
public static void main(String[] args){
-
-
//原子更新integer
-
count.compareAndSet(0,2);
-
count.compareAndSet(0,1);
-
count.compareAndSet(1,3);
-
count.compareAndSet(2,4);
-
count.compareAndSet(3,5);
-
log.info("count:{}",count.get());
-
-
//原子更新学生对象的age
-
studentAtomic.compareAndSet(studentAtomic.get(),new Student(10));
-
log.info("count:{}",studentAtomic.get().getAge());
-
}
-
}
-
@Slf4j
-
@ThreadSafe
-
public class AtomicExample4 {
-
-
/**
-
* AtomicIntegerFieldUpdater
-
*
-
* 基于反射的实用程序，可以对指定类的指定volatile int字段进行原子更新。 该类设计用于原子数据结构，其中同一节点的多个字段独立受原子更新的影响。
-
* 请注意， compareAndSet中的compareAndSet方法的保证弱于其他原子类。 因为这个类不能确保该字段的所有用途都适用于原子访问的目的，所以它可以
-
* 保证原子性仅在相同更新set上的compareAndSet和set其他调用。
-
*
-
* 类型为T参数的对象参数不是传递给newUpdater(java.lang.Class<U>, java.lang.String)的类的实例将导致抛出ClassCastException 。
-
*/
-
private static AtomicIntegerFieldUpdater<AtomicExample4> updater = AtomicIntegerFieldUpdater.newUpdater(AtomicExample4.class, "count");
-
-
@Getter
-
private volatile int count = 100;
-
-
public static void main(String[] args) {
-
AtomicExample4 atomicExample4 = new AtomicExample4();
-
if (updater.compareAndSet(atomicExample4, 100, 120)) {
-
log.info("update success 1,{}", atomicExample4.getCount());
-
}
-
if (updater.compareAndSet(atomicExample4, 100, 120)) {
-
log.info("update success 2,{}", atomicExample4.getCount());
-
} else {
-
log.info("update failed,{}", atomicExample4.getCount());
-
}
-
}
-
}
-
@Slf4j
-
@ThreadSafe
-
public class AtomicExample5 {
-
-
//请求次数
-
private static int clientTotal = 5000;
-
-
//允许同时运行的线程数
-
private static int threadTotal = 200;
-
-
//boolean标识
-
public static AtomicBoolean flag = new AtomicBoolean(false);
-
-
public static void main(String[] args) {
-
ExecutorService executorService = Executors.newCachedThreadPool();
-
final Semaphore semaphore = new Semaphore(threadTotal);
-
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
-
for (int i = 0; i < clientTotal; i++) {
-
executorService.execute(() -> {
-
try {
-
semaphore.acquire();
-
test();
-
semaphore.release();
-
} catch (InterruptedException e) {
-
log.error("exception", e);
-
}
-
countDownLatch.countDown();
-
});
-
}
-
try {
-
countDownLatch.await();
-
} catch (InterruptedException e) {
-
e.printStackTrace();
-
}
-
executorService.shutdown();
-
log.info("count:{}", flag.get());
-
}
-
-
private static void test() {
-
if (flag.compareAndSet(false, true)) {
-
log.info("多个线程同时操作我，我也只会执行一次");
-
}
-
}
-
}
（2）原子性 - 锁
synchronized ： 依赖jvm
Lock ：依赖特殊的cpu指令，代码实现，ReentrantLock 等
-
@Slf4j
-
public class SynchronizedExample1 {
-
-
//修饰一个代码块,作用范围：整个代码块，只作用于调用于该方法的对象，不同的对象之间互不影响
-
public void test1(String objectName) {
-
synchronized (this) {
-
for (int i = 0; i < 10; i++) {
-
log.info("test1-" + objectName + "-{}", i);
-
}
-
}
-
}
-
-
//修饰一个方法，作用范围：整个方法，只作用于调用于该方法的对象，不同的对象之间互不影响
-
public synchronized void test2(String objectName) {
-
for (int i = 0; i < 10; i++) {
-
log.info("test2-" + objectName + "-{}", i);
-
}
-
}
-
-
-
//修饰静态方法，作用范围：整个方法，作用于所有对象，不同的对象也会相互影响，进行锁竞争
-
public static synchronized void test3(String objectName) {
-
for (int i = 0; i < 10; i++) {
-
log.info("test2-" + objectName + "-{}", i);
-
}
-
}
-
-
//修饰静态方法，作用范围：整个方法，作用于所有对象，不同的对象也会相互影响，进行锁竞争
-
public static synchronized void test4(String objectName) {
-
synchronized (SynchronizedExample1.class) {
-
for (int i = 0; i < 10; i++) {
-
log.info("test2-" + objectName + "-{}", i);
-
}
-
}
-
}
-
-
-
public static void main(String[] args) {
-
//声明一个对象
-
SynchronizedExample1 synchronizedExample1 = new SynchronizedExample1();
-
ExecutorService service = Executors.newCachedThreadPool();
-
// service.execute(() -> {
-
// // synchronizedExample1.test1("Example1");
-
// synchronizedExample1.test2("Example1");
-
// });
-
// service.execute(()->{
-
// // synchronizedExample1.test1("Example1");
-
// synchronizedExample1.test2("Example1");
-
// });
-
-
//声明二个对象
-
SynchronizedExample1 synchronizedExample2 = new SynchronizedExample1();
-
// service.execute(() -> {
-
// // synchronizedExample1.test1("Example1");
-
// synchronizedExample1.test2("Example1");
-
// });
-
// service.execute(()->{
-
// // synchronizedExample1.test1("Example1");
-
// synchronizedExample2.test2("Example2");
-
// });
-
-
-
// service.execute(() -> {
-
// // synchronizedExample1.test1("Example1");
-
// synchronizedExample1.test3("Example1");
-
// });
-
// service.execute(()->{
-
// // synchronizedExample1.test1("Example1");
-
// synchronizedExample2.test3("Example2");
-
// });
-
-
service.execute(() -> {
-
// synchronizedExample1.test1("Example1");
-
synchronizedExample1.test4("Example1");
-
});
-
service.execute(() -> {
-
// synchronizedExample1.test1("Example1");
-
synchronizedExample2.test4("Example2");
-
});
-
}
-
}
- 可见性
volatile
-
@Slf4j
-
@NotThreadSafe
-
public class ConcurrencyTest {
-
-
//请求次数
-
private static int clientTotal = 5000;
-
-
//允许同时运行的线程数
-
private static int threadTotal = 200;
-
-
//计数
-
public static volatile int count = 0;
-
-
public static void main(String[] args) {
-
ExecutorService executorService = Executors.newCachedThreadPool();
-
final Semaphore semaphore = new Semaphore(threadTotal);
-
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
-
for (int i = 0; i < clientTotal; i++) {
-
executorService.execute(() -> {
-
try {
-
semaphore.acquire();
-
add();
-
semaphore.release();
-
} catch (InterruptedException e) {
-
log.error("exception", e);
-
}
-
countDownLatch.countDown();
-
});
-
-
}
-
try {
-
countDownLatch.await();
-
} catch (InterruptedException e) {
-
e.printStackTrace();
-
}
-
executorService.shutdown();
-
log.info("count:{}", count);
-
}
-
private static void add() {
-
count++;
-
/**
-
* count++进行了 三步操作
-
* 读取 count
-
* count + =1
-
* 写入 count
-
* 当我们没有用volatile 修饰时，无法确保每个线程读取到的值是最新的，就会出问题。
-
*
-
* 当我们用了volatile 修饰之后，虽然可以确保每个线程读取的值是最新的，但是不同线程进行+1 操作时还是会出问题。
-
* 所以 volatile 不具备原子性，因此volatile 特别适合用于状态标记变量，
-
* 当其中一个线程 修改了boolean值之后，可以在另外一个线程中，读取到最新的 boolean变量值
-
*/
-
}
例如：
（3）happens-before原则
程序次序规则：一个线程内，按照代码顺序，书写在前面的操作，先行发生于书写在后面的操作
锁定规则：一个unlock操作先行发生于后面对同一个锁的lock操作
volatile变量：对于一个变量的写操作先行发生于后面对这个变量的读操作
传递规则：如果操作A先行发生与操作B，而操作B又先行发生于操作C，那么操作A先行发生于操作C。
线程启动规则: Thread对象的start()方法先行发生于此线程的每一个动作
线程中断规则：对线程interrupt()方法的调用先行发生于被中断线程的代码检测到中断事件的发生
线程终结规则：线程中所有的操作都先行发生于线程的终止检测，我们可以通过Thread.join()方法结束，
Thread.isAlive(）的返回值手段检测到线程已经终止执行
对象终结规则：一个对象的初始化完成先行发生于他的finalize()方法的开始
文章来源: blog.csdn.net，作者：血煞风雨城2018，版权归原作者所有，如需转载，请联系作者。
原文链接：blog.csdn.net/qq_31905135/article/details/84189111
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）