之所以会有并发问题,是因为有个共享变量被很多人去读/写(这里面必定有一个写)
0. 我们处理锁还能做什么
很多并发问题都是通过对共享变量进行上锁,来达到访问的串行化, 让一个如何解决并发问题变成了, 如何将并发问题变成一个串行问题,并解决这个串行问题
当然本文参照的书中也提到,当你上锁的时候,所谓的线程优先级就已经不存在了, 除非在获得锁的时候才能体现这种优先级, 一个优先级再高的线程,不可能抢过已经持有锁的线程
还有没有不用锁,就可以解决并发问题的方法
1. 一个小小的CAS能干啥
一个小小的CAS只是提供了一种语义,但有了这个语义再加上while 循环就可以解决并发问题
CAS的语义更强一些,里面暗含了变量状态
CAS之保证现场只有一个人:
// 用java模拟一个CAS的操作
@ThreadSafe
public class SimulatedCAS {
@GuardedBy("this") private int value;
public synchronized int get() {
return value;
}
public synchronized int compareAndSwap(int expectedValue, int newValue) {
int oldValue = this.value;
if (oldValue == expectedValue) {
this.value = newValue;
}
return oldValue;
}
public synchronized boolean compareAndSet(int expectedValue, int newValue) {
return (expectedValue == compareAndSwap(expectedValue, newValue));
}
}
2. 非阻塞Stack
import java.util.concurrent.atomic.AtomicReference;
public class ConcurrentStack<E> {
AtomicReference<Node<E>> top = new AtomicReference<>();
public void push(E item) {
Node<E> newHead = new Node<E>(item);
Node<E> oldHead;
do {
oldHead = this.top.get();
newHead.next = oldHead;
} while (top.compareAndSet(oldHead, newHead));
}
public E pop() {
Node<E> oldHead;
Node<E> newHead;
do {
oldHead = this.top.get();
if (oldHead == null) {
return null;
}
newHead = oldHead.next;
// 注意这里使用了newHead来处理,而不是top.get() = top.get().next()
} while (!top.compareAndSet(oldHead, newHead));
return oldHead.item;
}
private static class Node<E> {
public final E item;
public Node<E> next;
public Node(E item) {
this.item = item;
}
}
}
3. 非阻塞Queue
import java.util.concurrent.atomic.AtomicReference;
public class LinkedQueue<E> {
private static class Node<E> {
final E item;
final AtomicReference<Node<E>> next;
public Node(E item, Node<E> next) {
this.item = item;
this.next = new AtomicReference<>(next);
}
}
private final Node<E> dummy = new Node<>(null, null);
private final AtomicReference<Node<E>> head = new AtomicReference<>(dummy);
private final AtomicReference<Node<E>> tail = new AtomicReference<>(dummy);
public boolean put(E item) {
Node<E> newNode = new Node<>(item, null);
while (true) {
Node<E> curTail = this.tail.get();
Node<E> tailNext = curTail.next.get();
if (curTail == this.tail.get()) {
if (tailNext != null) {
// Queue in intermediate state, advance tail
this.tail.compareAndSet(curTail, tailNext);
} else {
// In quiescent state, try inserting new node
if (curTail.next.compareAndSet(null, newNode)) {
// Insertion succeeded, try advancing tail
this.tail.compareAndSet(curTail, newNode);
return true;
}
}
}
}
}
}
4. 问题
- 为什么
ConcurrentStack#Node的next不是AtomicReference类型 - 为什么
LinkedQueue#Node的next是AtomicReference类型
5. 解答
我们可以将上面2个问题合并起来看, 同样是添加元素, 一个是往头部添加, 一个是往尾部添加
但同样都是2个步骤:
- 修改
next - 修改
head
但是步骤1,在2种不同的数据结构中它的可见性是不同的:
| 步骤 | ConcurrentStack |
LinkedQueue |
|---|---|---|
修改next
|
修改新元素的next,不可见 |
修改已存在元素的next, 可见 |
要有并发问题,首先是要有可见性
+-----+ +-----+ +-----+ +-----+
| | | | | | | |
| H +---> | A +------> B +---->+ C |
| | | | | | | |
+-----+ +-----+ +-----+ +-----+
ConcurrentStack 添加元素, 即便是H的next节点已经指到了A节点, 但是因为其他线程看到的还是以A节点为头的一个Stack 所以这样H 的next怎么修改都不影响
相反, LinkedQueue 则是在已有元素上修改则会有一个中间态出来, 直接从书中截图出来:
linked_queue_1.png
linked_queue_2.png
从上图可以看出来2节点再插入新节点的时候会有一个中间态, 此时next 需要做并发处理, 所以类型为 AtomicReference
还有一个点很重要, 当我们发现出于中间态的数据的时候,并不是什么都不做, 而是积极促成这样一件事, 这样cpu就算是干活了, 原文的摘抄如下:
Rather than wait for that thread to finish, the current thread helps it by finishing the operation for it, advancing the tail pointer
6. 总结
- 可见性决定是否使用
Atomic*类型 - 非阻塞会极大浪费CPU
- 看到其他线程正在修改的数据, 主动提供帮助会提高CPU利用率 (对比第2点)
ref:
- Java Concurrency in Practice, chapter 15
网友评论