0
点赞
收藏
分享

微信扫一扫

《Java 并发编程》共享模型之无锁

何晓杰Dev 2022-05-01 阅读 68

《Java 并发编程》共享模型之无锁

🚀1. 无锁解决线程安全问题

有如下需求,保证 account.withdraw 取款方法的线程安全

interface Account {
	Integer getBalance();

	void withdraw(Integer amount);

	/**
	 * 方法内会启动 1000 个线程,每个线程做 -10 元 的操作     * 如果初始余额为 10000 那么正确的结果应当是 0
	 */
	static void demo(Account account) {
		List<Thread> ts = new ArrayList<>();
		long start = System.nanoTime();
		for (int i = 0; i < 1000; i++) {
			ts.add(new Thread(() -> {
				account.withdraw(10);
			}));
		}
		ts.forEach(Thread::start);
		ts.forEach(t -> {
			try {
				t.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});
		long end = System.nanoTime();
		System.out.println(account.getBalance() + " cost: " + (end - start) / 1000_000 + " ms");
	}
}

//线程不安全的做法
class AccountUnsafe implements Account {
	private Integer balance;

	public AccountUnsafe(Integer balance) {
		this.balance = balance;
	}


	@Override
	public Integer getBalance() {
		return this.balance;
	}

	@Override
	public synchronized void withdraw(Integer amount) {
		balance -= amount;
	}

	public static void main(String[] args) {
		Account.demo(new AccountUnsafe(10000));
	}
}

某一次的运行结果

200, cost:178ms

使用原子整数 AtomicInteger 来保证线程安全的做法

public class AccountCas implements Account{
    //使用原子整数
    private AtomicInteger balance;

    public AccountCas(int balance) {
        this.balance = new AtomicInteger(balance);
    }

    @Override
    public Integer getBalance() {
        //得到原子整数的值
        return balance.get();
    }

    @Override
    public void withdraw(Integer amount) {
        while(true) {
            //获得修改前的值
            int prev = balance.get();
            //获得修改后的值
            int next = prev-amount;
            //比较并设值
            if(balance.compareAndSet(prev, next)) {
                break;
            }
        }
    }
    public static void main(String[] args) {
        Account.demo(new AccountCas(10000));
    }
}

某一次的运行结果

0,cost:94ms

🚀2. CAS 与 volatile

上一章节中看到的 AtomicInteger 的解决方法,内部并没有用锁来保护共享变量的线程安全。那么它是如何实现的呢?

回顾一下上一章节中的 withdraw 方法

public void withdraw(Integer amount) {
		//需要不断尝试,直到成功为止
       while(true) {
           //比如拿到了旧值1000
           int prev = balance.get();
           //在这个基础上 1000-10 = 990
           int next = prev-amount;
           /*
           compareAndSet正是做这个检查,在set前,先比较prev与当前值
           不一致了,next作废,返回false表示失败
           比如,别的线程已经做了减法,当前值为990
           那么本线程的这次990就作废了,进入while下次循环重试
           一致,以next设置为新值,返回true表示成功
           */
           if(balance.compareAndSet(prev, next)) {
               break;
           }
       }
}

其中的关键是比较并设置值 (CompareAndSet),简称就是 CAS (也有 Compare And Swap 的说法),它必须是原子操作。
在这里插入图片描述
工作流程

  • 当一个线程要去修改 Account 对象中的值时,先获取值 pre(调用 get 方法),然后再将其设置为新的值 next(调用 cas 方法)。在调用 cas 方法时,会将 pre 与 Account 中的余额进行比较。
    (1)如果两者相等,就说明该值还未被其他线程修改,此时便可以进行修改操作。
    (2)如果两者不相等,就不设置值,重新获取值 pre(调用 get 方法),然后再将其设置为新的值 next(调用 cas 方法),直到修改成功为止。

注意

  • 其实 CAS 的底层是 lock cmpxchg 指令(X86 架构),在单核 CPU 和多核 CPU 下都能够保证【比较-交换】的原子性。
  • 在多核状态下,某个核执行到带 lock 的指令时,CPU 会让总线锁住,当这个核把此指令执行完毕,再开启总线。这个过程中不会被线程的调度机制所打断,保证了多个线程对内存操作的准确性,是原子的。

volatile
获取共享变量时,为了保证该变量的可见性,需要使用 volatile 修饰。它可以用来修饰成员变量和静态成员变量,可以避免线程从自己的工作缓存中查找变量的值,必须到主存中获取它的值,线程操作 volatile 变量都是直接操作主存。即一个线程对 volatile 变量的修改,对另一个线程可见。

注意

  • volatile 仅仅保证了共享变量的可见性,让其它线程能够看到新值,但不能解决指令交错问题(不能保证原子性)

CAS 必须借助 volatile 才能读取到共享变量的新值来实现【比较并交换】的效果

通常情况下,使用无锁解决线程安全问题比有锁效率高,原因如下

  • 无锁情况下,即使重试失败,线程始终在高速运行,没有停歇,而使用 synchronized 锁会让线程在没有获得锁的时候,发生上下文切换,进入阻塞。
  • 但无锁情况下,因为线程要保持运行,需要额外 CPU 的支持,如果没有额外的 CPU 支持,还是会导致上下文切换

CAS 的特点

结合 CAS 和 volatile 可以实现无锁并发,适用于线程数少、多核 CPU 的场景下

  • CAS 是基于乐观锁的思想:乐观的估计,不怕别的线程来修改共享变量,就算改了也没关系,我吃亏点再重试呗
  • synchronized 是基于悲观锁的思想:悲观的估计,得防着其它线程来修改共享变量,我上了锁你们都别想改,我改完了解开锁,你们才有机会
  • CAS 体现的是无锁并发、无阻塞并发,请仔细体会这两句话的意思
    因为没有使用 synchronized,所以线程不会陷入阻塞,这是效率提升的因素之一
    但如果竞争激烈,可以想到重试必然频繁发生,反而效率会受影响

🚀3. 原子整数

J.U.C 并发包提供了

  • AtomicBoolean
  • AtomicInteger
  • AtomicLong

以 AtomicInteger 为例

AtomicInteger i = new AtomicInteger(0);
// 获取并自增(i = 0, 结果 i = 1, 返回 0),类似于 i++
System.out.println(i.getAndIncrement());
// 自增并获取(i = 1, 结果 i = 2, 返回 2),类似于 ++i
System.out.println(i.incrementAndGet());
// 自减并获取(i = 2, 结果 i = 1, 返回 1),类似于 --i
System.out.println(i.decrementAndGet());
// 获取并自减(i = 1, 结果 i = 0, 返回 1),类似于 i--
System.out.println(i.getAndDecrement());
// 获取并加值(i = 0, 结果 i = 5, 返回 0)
System.out.println(i.getAndAdd(5));
// 加值并获取(i = 5, 结果 i = 0, 返回 0)
System.out.println(i.addAndGet(-5));
// 获取并更新(i = 0, p 为 i 的当前值, 结果 i = -2, 返回 0)
// 其中函数中的操作能保证原子,但函数需要无副作用
System.out.println(i.getAndUpdate(p -> p - 2));
// 更新并获取(i = -2, p 为 i 的当前值, 结果 i = 0, 返回 0)
// 其中函数中的操作能保证原子,但函数需要无副作用
System.out.println(i.updateAndGet(p -> p + 2));
// 获取并计算(i = 0, p 为 i 的当前值, x 为参数1, 结果 i = 10, 返回 0)
// 其中函数中的操作能保证原子,但函数需要无副作用
// getAndUpdate 如果在 lambda 中引用了外部的局部变量,要保证该局部变量是 final 的
// getAndAccumulate 可以通过 参数1 来引用外部的局部变量,但因为其不在 lambda 中因此不必是 final
System.out.println(i.getAndAccumulate(10, (p, x) -> p + x));
// 计算并获取(i = 10, p 为 i 的当前值, x 为参数1, 结果 i = 0, 返回 0)
// 其中函数中的操作能保证原子,但函数需要无副作用
System.out.println(i.accumulateAndGet(-10, (p, x) -> p + x));

🚀4. 原子引用

原子引用类型有

  • AtomicReference
  • AtomicMarkableReference
  • AtomicStampedReference

DecimalAccount 实现,使用 CAS 保证安全的取款操作

public interface DecimalAccount {
	BigDecimal getBalance();

	void withdraw(BigDecimal amount);

	/**
	 * 方法内会启动 1000 个线程,每个线程做 -10 元 的操作    
     * 如果初始余额为 10000 那么正确的结果应当是 0
	 */
	static void demo(DecimalAccountImpl account) {
		List<Thread> ts = new ArrayList<>();
		long start = System.nanoTime();
		for (int i = 0; i < 1000; i++) {
			ts.add(new Thread(() -> {
				account.withdraw(BigDecimal.TEN);
			}));
		}
		ts.forEach(Thread::start);
		ts.forEach(t -> {
			try {
				t.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});
		long end = System.nanoTime();
		System.out.println(account.getBalance() + " cost: " + (end - start) / 1000_000 + " ms");
	}
}

class DecimalAccountImpl implements DecimalAccount {
	//原子引用,泛型类型为小数类型
	AtomicReference<BigDecimal> balance;

	public DecimalAccountImpl(BigDecimal balance) {
		this.balance = new AtomicReference<BigDecimal>(balance);
	}

	@Override
	public BigDecimal getBalance() {
		return balance.get();
	}

	@Override
	public void withdraw(BigDecimal amount) {
		while(true) {
			BigDecimal pre = balance.get();
			BigDecimal next = pre.subtract(amount);
			if(balance.compareAndSet(pre, next)) {
				break;
			}
		}
	}

	public static void main(String[] args) {
		DecimalAccount.demo(new DecimalAccountImpl(new BigDecimal("10000")));
	}
}

🚀5. ABA 问题及解决

static AtomicReference<String> ref = new AtomicReference<>("A");

public static void main(String[] args) {
   new Thread(()->{
       String pre = ref.get();   // A
       System.out.println("change");
       try {
           other();
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       try {
           Thread.sleep(1000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       System.out.println("change A->C "+ref.compareAndSet(pre, "C"));
   }).start();
}
static void other() throws InterruptedException {
    new Thread(()->{
        System.out.println("change A->B " + ref.compareAndSet("A", "B"));
    }).start();
    Thread.sleep(500);
    new Thread(()->{
        System.out.println("change B->A "+ref.compareAndSet("B","A"));
    }).start();
}

运行结果如下

change
change A->B true
change B->A true
change A->C true

那么,观察上面的运行结果可以发现,主线程仅能判断出共享变量的值与初值 A 是否相同,不能感知到这种从 A 改为 B 又 改回 A 的情况,如果主线程希望:

  • 只要有其它线程【动过了】共享变量,那么自己的 cas 就算失败,这时,仅比较值是不够的,需要再加一个版本号

AtomicStampedReference

 static AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 0);

 public static void main(String[] args) {
    new Thread(()->{
        String pre = ref.getReference();
        //获得版本号
        int stamp = ref.getStamp();
        System.out.println("change");
        try {
            other();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //把str中的A改为C,并比对版本号,如果版本号相同,就执行替换,并让版本号+1
        System.out.println("change A->C stamp " + stamp + ref.compareAndSet(pre, "C", stamp, stamp+1));
    }).start();
 }
 static void other() throws InterruptedException {
     new Thread(()->{
         int stamp = ref.getStamp();
         System.out.println("change A->B stamp " + stamp + ref.compareAndSet("A", "B", stamp, stamp+1));
     }).start();
     Thread.sleep(500);
     new Thread(()->{
         int stamp = ref.getStamp();
         System.out.println("change B->A stamp " + stamp +  ref.compareAndSet("B", "A", stamp, stamp+1));
     }).start();
 }

运行结果如下

change
change A->B stamp 0true
change B->A stamp 1true
change A->C stamp 0false

AtomicStampedReference 可以给原子引用加上版本号,追踪原子引用整个的变化过程,如: A -> B -> A -> C ,通过 AtomicStampedReference,我们可以知道,引用变量中途被更改了几次。但是有时候,并不关心引用变量更改了几次,只是单纯的关心是否更改过,所以就有了 AtomicMarkableReference.

AtomicMarkableReference

static AtomicMarkableReference<String> ref = new AtomicMarkableReference<>("A", true);

public static void main(String[] args) {
   new Thread(()->{
       String pre = ref.getReference();
       System.out.println("change");
       try {
           other();
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       try {
           Thread.sleep(1000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       System.out.println("change A->C mark " +  ref.compareAndSet(pre, "C", true, false));
   }).start();
}
static void other() throws InterruptedException {
    new Thread(()->{
        System.out.println("change A->A mark " + ref.compareAndSet("A", "A", true, false));
    }).start();
}

运行结果如下

change
change A->A mark true
change A->C mark false

两者的区别

  • AtomicStampedReference 需要我们传入整型变量作为版本号,来判定是否被更改过(可以追踪变化过程)
  • AtomicMarkableReference 需要我们传入布尔变量作为标记,来判断是否被更改过(只关心是否更改过)

🚀6. 原子数组和原子字段更新器

原子数组类型有

  • AtomicIntegerArray
  • AtomicLongArray
  • AtomicReferenceArray

原子字段更新器类型有

  • AtomicReferenceFieldUpdater
  • AtomicIntegerFieldUpdater
  • AtomicLongFieldUpdater

利用字段更新器,可以针对对象的某个域(Field)进行原子操作,只能配合 volatile 修饰的字段使用,否则会出现异常 Exception in thread "main" java.lang.IllegalArgumentException: Must be volatile type.

public class Test5 {
    private volatile int field;

    public static void main(String[] args) {
        AtomicIntegerFieldUpdater fieldUpdater = AtomicIntegerFieldUpdater.newUpdater(Test5.class, "field");

        Test5 test5 = new Test5();
        fieldUpdater.compareAndSet(test5, 0, 10);
        //修改成功 field=10
        System.out.println(test5.field);
        //修改成功 field=20
        fieldUpdater.compareAndSet(test5, 10, 20);
        System.out.println(test5.field);

        //修改失败 field=20
        fieldUpdater.compareAndSet(test5, 10, 30);
        System.out.println(test5.field);

    }
}

运行结果如下

10
20
20

原子字段更新器初始化过程

从上面的例子可以看出,原子更新器是通过 newUpdater 来获取实例的。其中传入了三个参数

  • 拥有属性的类的 Class
  • 属性的 Class
  • 属性的名称

未完待续…

举报

相关推荐

0 条评论