并发编程三大特性
原子性
线程的原子性
证明线程的原子性
package com.juc.c_001_sync_basics;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class T00_00_IPlusPlus {
private static long n = 0L;
public static void main(String[] args) throws Exception {
//Lock lock = new ReentrantLock();
Thread[] threads = new Thread[100];
CountDownLatch latch = new CountDownLatch(threads.length);
for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(() -> {
for (int j = 0; j < 10000; j++) {
synchronized (T00_00_IPlusPlus.class) {//当成一个整体不可打断 本身保持了可见性 原子性 但是不保证有序性 单线程保证
//最终一致性和锁没有关系
//lock.lock();
n++;
//lock.unlock();
}
}
latch.countDown();
});
}
for (Thread t : threads) {
t.start();
}
latch.await();
System.out.println(n);
}
}
//多线程访问同一个数据会产生竞争 可能会产生数据不一致性
//上锁 锁定某个对象 保证原子性 上锁的本质是把并发编程序列化
通过使用synchronized关键字或者ReentrantLock锁，让一百个线程在执行n++的时候保证n不会被其他线程拿去累加使用。
synchronized关键字锁的必须是所有线程共同拥有的一个东西才可以保证数据安全，保证线程的序列化执行。
package com.juc.c_001_sync_basics;
import com.util.SleepHelper;
public class T00_02_SingleLockVSMultiLock {
private static Object o1 = new Object();
private static Object o2 = new Object();
private static Object o3 = new Object();
public static void main(String[] args) {
Runnable r1 = () -> {
synchronized (o1) {
System.out.println(Thread.currentThread().getName() + " start!");
SleepHelper.sleepSeconds(2);
System.out.println(Thread.currentThread().getName() + " end!");
}
};
Runnable r2 = () -> {
synchronized (o2) {
System.out.println(Thread.currentThread().getName() + " start!");
SleepHelper.sleepSeconds(2);
System.out.println(Thread.currentThread().getName() + " end!");
}
};
Runnable r3 = () -> {
synchronized (o3) {
System.out.println(Thread.currentThread().getName() + " start!");
SleepHelper.sleepSeconds(2);
System.out.println(Thread.currentThread().getName() + " end!");
}
};
new Thread(r1).start();
new Thread(r2).start();
new Thread(r3).start();
}
}
//上锁上的同1把锁才能保证原子性 才能保证同一时刻只有一个线程执行
三个线程的锁锁的不是同一个对象，所以多线程会同时执行还是2s。
package com.juc.c_001_sync_basics;
import com.mashibing.util.SleepHelper;
//上锁的本质是把并发编程序列化 上锁之后效率变低
public class T00_01_WhatIsLock {
private static Object o = new Object();
public static void main(String[] args) {
Runnable r = () -> {
synchronized (o) {//如果临界区执行时间长，语句多，叫做 锁的粒度比较粗，反之，就是锁的粒度比较细
System.out.println(Thread.currentThread().getName() + " start!");
SleepHelper.sleepSeconds(2);
System.out.println(Thread.currentThread().getName() + " end!");
}
};
for (int i = 0; i < 3; i++) {
new Thread(r).start();
}
}
}
synchronized 关键字锁的是同一个东西，所以每一个线程执行完下一个线程才会执行。
synchronized 本身也保证可见性的。锁定的代码块的数据在执行完之后也会同步到主存，从而让下一个线程用的时候会获到最新的。
一些基本概念
- race condition => 竞争条件 ， 指的是多个线程访问共享数据的时候产生竞争
- 数据的不一致（unconsistency)，并发访问之下产生的不期望出现的结果
- 如何保障数据一致呢？–> 线程同步（线程执行的顺序安排好），
- monitor （管程） —> 锁
- critical section -> 临界区
如果临界区执行时间长，语句多，叫做 锁的粒度比较粗，反之，就是锁的粒度比较细
具体： 保障操作的原子性（Atomicity)
悲观的认为这个操作会被别的线程打断（悲观锁）synchronized（上一个小程序）
乐观的认为这个做不会被别的线程打断（乐观锁 自旋锁 无锁）cas操作 CAS = Compare And Set/Swap/Exchange
Java中的8大原子操作：
- lock：主内存，标识变量为线程独占
- unlock：主内存，解锁线程独占变量
- read：主内存，读取内存到线程缓存（工作内存）
- load：工作内存，read后的值放入线程本地变量副本
- use：工作内存，传值给执行引擎
- assign：工作内存，执行引擎结果赋值给线程本地变量
- store：工作内存，存值到主内存给write备用
- write：主内存，写变量值
CAS
CAS 全称是 compare and swap，是一种用于在多线程环境下实现同步功能的机制。CAS 操作包含三个操作数 – 内存位置、预期数值和新值。CAS 的实现逻辑是将内存位置处的数值与预期数值想比较，若相等，则将内存位置处的值替换为新值。若不相等，则不做任何操作.
ABA问题解决
加版本号
CAS底层原子性保证
CAS底层还是一把锁，锁的缓存行或者总线。
乐观锁操作，回写的时候比较一下。
/**
* 解决同样的问题的更高效的方法，使用AtomXXX类
* AtomXXX类本身方法都是原子性的，但不能保证多个方法连续调用是原子性的
*
*/
package com.juc.c_018_00_AtomicXXX;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
public class T01_AtomicInteger {
/*volatile*/ //int count1 = 0;
AtomicInteger count = new AtomicInteger(0);
/* synchronized */void m() {
for (int i = 0; i < 10000; i++)
//if count1.get() < 1000
count.incrementAndGet(); //count1++
}
public static void main(String[] args) {
T01_AtomicInteger t = new T01_AtomicInteger();
List<Thread> threads = new ArrayList<Thread>();
for (int i = 0; i < 100; i++) {
threads.add(new Thread(t::m, "thread-" + i));
}
threads.forEach((o) -> o.start());
threads.forEach((o) -> {
try {
o.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println(t.count);
}
}
AtomicInteger:
public final int incrementAndGet() {
for (;;) {
int current = get();
int next = current + 1;
if (compareAndSet(current, next))
return next;
}
}
public final boolean compareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
Unsafe:
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
运用：
package com.jol;
import sun.misc.Unsafe;
import java.lang.reflect.Field;
public class T02_TestUnsafe {
int i = 0;
private static T02_TestUnsafe t = new T02_TestUnsafe();
public static void main(String[] args) throws Exception {
//Unsafe unsafe = Unsafe.getUnsafe();
Field unsafeField = Unsafe.class.getDeclaredFields()[0];
unsafeField.setAccessible(true);
Unsafe unsafe = (Unsafe) unsafeField.get(null);
Field f = T02_TestUnsafe.class.getDeclaredField("i");
long offset = unsafe.objectFieldOffset(f);
System.out.println(offset);
boolean success = unsafe.compareAndSwapInt(t, offset, 0, 1);
System.out.println(success);
System.out.println(t.i);
//unsafe.compareAndSwapInt()
}
}
jdk8u: unsafe.cpp:
cmpxchg = compare and exchange set swap
UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
UnsafeWrapper("Unsafe_CompareAndSwapInt");
oop p = JNIHandles::resolve(obj);
jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);
return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END
jdk8u: atomic_linux_x86.inline.hpp 93行
is_MP = Multi Processors
inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {
int mp = os::is_MP();//是不是多核处理器
__asm__ volatile (LOCK_IF_MP(%4) "cmpxchgl %1,(%3)"
: "=a" (exchange_value)
: "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
: "cc", "memory");
return exchange_value;
}
jdk8u: os.hpp is_MP()
static inline bool is_MP() {
// During bootstrap if _processor_count is not yet initialized
// we claim to be MP as that is safest. If any platform has a
// stub generator that might be triggered in this phase and for
// which being declared MP when in fact not, is a problem - then
// the bootstrap routine for the stub generator needs to check
// the processor count directly and leave the bootstrap routine
// in place until called after initialization has ocurred.
return (_processor_count != 1) || AssumeMP;
}
jdk8u: atomic_linux_x86.inline.hpp
#define LOCK_IF_MP(mp) "cmp $0, " #mp "; je 1f; lock; 1: "
最终实现：
cmpxchg = cas修改变量值
lock cmpxchg 指令
硬件：
lock指令在执行的时候视情况采用缓存锁或者总线锁
