Concurrency: Sharing Resources (2)

Author: nex3z 2016-07-03

3 Atomicity and Volatility

　　原子操作（Atomic Operation）指的是不会被线程调度打断的操作。对除了long和double以外基本类型的简单操作（如赋值和返回值）是原子性的。JVM允许以两次32位操作的形式读写64位的值（long和double型变量），导致读写期间可能发生上下文切换。为了使long和double得到原子性，可以使用volatile来修饰long和double型变量。

　　凭借原子操作不会被线程调度打断的特性，专家可以写出没有锁的代码，不需要使用同步机制。但不要轻易尝试。移除同步通常意味着不成熟的优化，通常会带来更多问题。

　　在多处理器系统上，可见性（Visibility）通常比原子性（Atomicity）更容易引发问题。在一个任务中进行的修改，即便使用的是原子操作、不会被打断，这个修改在另一个任务中也可能是不可见的，因为修改可能被暂存在处理器的缓存中，对其他任务不可见。那么问题来了：不同任务对当前应用程序的状态有不同的观测。同步机制会强制一个任务的修改可以在整个应用程序中可见；没有同步机制，一个任务的修改何时能对其他任务可见是不确定的。

　　volatile关键字也可以确保可见性。如果声明某个变量为volatile，那么对该变量的任何修改，都会立即对整个程序可见。对volatile变量的修改会立即写入主内存，并从主内存读出，不会使用本地缓存。

　　原子性和volatile是不同的概念。对非volatile的原子操作不一定会被写入主内存，故而不确保对其他任务的可见性。如果多个任务都要访问某个变量，则该变量应当是volatile的；如果不使用volatile，那么该变量应当仅能够以同步的方式访问。反过来说，如果一个变量仅能从同步的方法或代码块中访问，那就不需要声名其为volatile。

　　在同一个任务中的修改会对该任务可见，如果变量仅在一个任务中用到，就不需要声名为volatile。

　　“原子操作不需要同步”的观点是错误的。

import java.util.concurrent.*;

public class AtomicityTest implements Runnable {
    private int i = 0;
    public int getValue() { return i; }
    private synchronized void evenIncrement() { i++; i++; }
    public void run() {
        while(true)
            evenIncrement();
    }
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        AtomicityTest at = new AtomicityTest();
        exec.execute(at);
        while(true) {
            int val = at.getValue();
            if(val % 2 != 0) {
                System.out.println(val);
                System.exit(0);
            }
        }
    }
}
/* Output
11
*/

在上面的例子中，自加方法evenIncrement()是同步的；getValue()中只是返回i，返回操作是原子操作，但还是出现了奇数的情况。原因在于，getValue()中对i的读取缺乏同步，如果读取发生在i的自加操作中间，就可能读取到处于奇数状态的i。getValue()和evenIncrement()都需要同步。

　　下面的例子中，nextSerialNumber()用于产生唯一的序列号：

public class SerialNumberGenerator {
    private static volatile int serialNumber = 0;
    public static int nextSerialNumber() {
        return serialNumber++; // Not thread-safe
    }
}

在Java中，自增不是原子操作，它包含了读取和写入两种操作，serialNumber通过volatile保证了可见性，其他任务调用了nextSerialNumber()后，可以及时观测到serialNumber的变化。但这里的问题是，nextSerialNumber()没有使用同步的方式访问serialNumber，所以其他任务可能会在serialNumber的自增操作期间访问serialNumber，导致读取到不确定的结果。volatile并不能解决“自增操作不是原子操作、可能被打断”这一事实。

　　简单来说，如果一个变量会被多个任务同时访问，且至少有一个任务会修改该变量，就应该使用volatile。举例来说，用于停止任务的标志位必须是volatile，否则的话，该标志位可能被缓存在寄存器里，当其他任务修改了该标志位，缓存的值并不会改变，导致无法及时停止任务。

　　下面的例子为了测试SerialNumberGenerator，使用了一个循环存储CircularSet，避免内存不足：

import java.util.concurrent.*;

class CircularSet {
    private int[] array;
    private int len;
    private int index = 0;
    public CircularSet(int size) {
        array = new int[size];
        len = size;
        // Initialize to a value not produced
        // by the SerialNumberGenerator:
        for(int i = 0; i < size; i++)
            array[i] = -1;
    }
    public synchronized void add(int i) {
        array[index] = i;
        // Wrap index and write over old elements:
        index = ++index % len;
    }
    public synchronized boolean contains(int val) {
        for(int i = 0; i < len; i++)
            if(array[i] == val) return true;
        return false;
    }
}

public class SerialNumberChecker {
    private static final int SIZE = 10;
    private static CircularSet serials =
            new CircularSet(1000);
    private static ExecutorService exec =
            Executors.newCachedThreadPool();
    static class SerialChecker implements Runnable {
        public void run() {
            while(true) {
                int serial =
                        SerialNumberGenerator.nextSerialNumber();
                if(serials.contains(serial)) {
                    System.out.println("Duplicate: " + serial);
                    System.exit(0);
                }
                serials.add(serial);
            }
        }
    }
    public static void main(String[] args) throws Exception {
        for(int i = 0; i < SIZE; i++)
            exec.execute(new SerialChecker());
        // Stop after n seconds if there’s an argument:
        if(args.length > 0) {
            TimeUnit.SECONDS.sleep(new Integer(args[0]));
            System.out.println("No duplicates detected");
            System.exit(0);
        }
    }
}
/* Output
Duplicate: 3213
*/

这里使用了10个任务并行地调用SerialNumberGenerator.nextSerialNumber()，打印输出了Duplicate: 3213，出现了重复的结果，验证了上面的分析。解决方法很简单，只需为SerialNumberGenerator的nextSerialNumber()方法加上synchronized。

4. Atomic Classes

　　原子类（Atomic Class）是Java SE 5中引入的，提供原子性的操作。

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.*;

public class AtomicIntegerTest implements Runnable {
    private AtomicInteger i = new AtomicInteger(0);
    public int getValue() { return i.get(); }
    private void evenIncrement() { i.addAndGet(2); }
    public void run() {
        while(true)
            evenIncrement();
    }
    public static void main(String[] args) {
        new Timer().schedule(new TimerTask() {
            public void run() {
                System.err.println("Aborting");
                System.exit(0);
            }
        }, 5000); // Terminate after 5 seconds 
        ExecutorService exec = Executors.newCachedThreadPool();
        AtomicIntegerTest ait = new AtomicIntegerTest();
        exec.execute(ait);
        while(true) {
            int val = ait.getValue();
            if(val % 2 != 0) {
                System.out.println(val);
                System.exit(0);
            }
        }
    }
}

这里使用了AtomicInteger，不再需要使用synchronized。上面的例子会在5秒后打印Aborting并结束，不会检出奇数的情况。

　　另一个例子：

import java.util.concurrent.atomic.*;

public class AtomicEvenGenerator extends IntGenerator {
    private AtomicInteger currentEvenValue = new AtomicInteger(0);
    public int next() {
        return currentEvenValue.addAndGet(2);
    }
    public static void main(String[] args) {
        EvenChecker.test(new AtomicEvenGenerator());
    }
}

同样也不会出现奇数的情况。

　　原子类被设计用于实现java.util.concurrent中的类，只应当在特殊情况下使用。通常来说使用synchronized或Lock进行锁定更加安全。

5. Critical Sections

　　通过synchronized关键字可以创建一块临界区（Critical Section），也称为同步块（Synchronized Block）:

synchronized(syncObject) { 
    // This code can be accessed by only one task at a time 
}

其中的代码同时只能被一个线程访问。由此可以限制多个线程同时访问方法中的一段代码，而不直接同步整个方法。在进入同步块前，必须要获得syncObject的锁定，如果其他任务已经获取了锁定，那么在锁定释放前，其他任务就无法进入同步块。

　　同步块能为其他任务提供更多的访问机会，提高资源的使用效率：

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.*;

class Pair { // Not thread-safe
    private int x, y;
    public Pair(int x, int y) {
        this.x = x;
        this.y = y;
    }
    public Pair() { this(0, 0); }
    public int getX() { return x; }
    public int getY() { return y; }
    public void incrementX() { x++; }
    public void incrementY() { y++; }
    public String toString() {
        return "x: " + x + ", y: " + y;
    }
    public class PairValuesNotEqualException extends RuntimeException {
        public PairValuesNotEqualException() {
            super("Pair values not equal: " + Pair.this);
        }
    }
    // Arbitrary invariant -- both variables must be equal:
    public void checkState() {
        if(x != y)
            throw new PairValuesNotEqualException();
    }
}

// Protect a Pair inside a thread-safe class:
abstract class PairManager {
    AtomicInteger checkCounter = new AtomicInteger(0);
    protected Pair p = new Pair();
    private List<Pair> storage = Collections.synchronizedList(new ArrayList<Pair>());
    public synchronized Pair getPair() {
        // Make a copy to keep the original safe:
        return new Pair(p.getX(), p.getY());
    }
    // Assume this is a time consuming operation
    protected void store(Pair p) {
        storage.add(p);
        try {
            TimeUnit.MILLISECONDS.sleep(50);
        } catch(InterruptedException ignore) {}
    }
    public abstract void increment();
}

// Synchronize the entire method:
class PairManager1 extends PairManager {
    public synchronized void increment() {
        p.incrementX();
        p.incrementY();
        store(getPair());
    }
}

// Use a critical section:
class PairManager2 extends PairManager {
    public void increment() {
        Pair temp;
        synchronized(this) {
            p.incrementX();
            p.incrementY();
            temp = getPair();
        }
        store(temp);
    }
}

class PairManipulator implements Runnable {
    private PairManager pm;
    public PairManipulator(PairManager pm) {
        this.pm = pm;
    }
    public void run() {
        while(true)
            pm.increment();
    }
    public String toString() {
        return "Pair: " + pm.getPair() + " checkCounter = " + pm.checkCounter.get();
    }
}

class PairChecker implements Runnable {
    private PairManager pm;
    public PairChecker(PairManager pm) {
        this.pm = pm;
    }
    public void run() {
        while(true) {
            pm.checkCounter.incrementAndGet();
            pm.getPair().checkState();
        }
    }
}

public class CriticalSection {
    // Test the two different approaches:
    static void
    testApproaches(PairManager pman1, PairManager pman2) {
        ExecutorService exec = Executors.newCachedThreadPool();
        PairManipulator pm1 = new PairManipulator(pman1), 
                pm2 = new PairManipulator(pman2);
        PairChecker pcheck1 = new PairChecker(pman1), 
                pcheck2 = new PairChecker(pman2);
        exec.execute(pm1);
        exec.execute(pm2);
        exec.execute(pcheck1);
        exec.execute(pcheck2);
        try {
            TimeUnit.MILLISECONDS.sleep(500);
        } catch(InterruptedException e) {
            System.out.println("Sleep interrupted");
        }
        System.out.println("pm1: " + pm1 + "\npm2: " + pm2);
        System.exit(0);
    }
    public static void main(String[] args) {
        PairManager pman1 = new PairManager1(), 
                pman2 = new PairManager2();
        testApproaches(pman1, pman2);
    }
}
/* Output
pm1: Pair: x: 55, y: 55  checkCounter = 3
pm2: Pair: x: 56, y: 56 checkCounter = 154288579
*/

Pair不是线程安全的，其中的自加操作不是原子的，存在被打断的情况，incrementX()和incrementY()也没有同步。这里使用PairManager持有Pair并外部对Pair的访问，实现线程安全。PairManager由两个具体子类，PairManager1同步了整个increment()方法，PairManager2只同步increment()方法中的一部分。PairManager的store()用于把Pair保存到同步列表，不需要进行同步。PairChecker在单独线程上不停地访问PairManager，用checkCounter记录成功访问的次数，由此测定PairManager各具体子类处于解锁状态的时间在整个测试期间所占的比重。从打印可见，pm2，也就是使用同步块的PairManager2，占用并锁定共享资源的时间较少，可以让共享的资源在多数时间保持解锁、供其他任务使用。

　　Thinking in Java原书在这里还有一个比较同步整个方法和手动为代码块加锁的例子，但在我的电脑上运行会出现问题，详见混用同步块和同步方法时的问题。

6. Synchronizing On Other Objects

　　使用同步块时，必须指定要同步的对象。通常使用当前对象自身进行同步，即synchronized(this)。但有时候也需要同步于其他对象，此时必须确保其他所有相关的任务都同步于同一个对象。

class DualSynch {
    private Object syncObject = new Object();
    public synchronized void f() {
        for(int i = 0; i < 5; i++) {
            System.out.println("f()");
            Thread.yield();
        }
    }
    public void g() {
        synchronized(syncObject) {
            for(int i = 0; i < 5; i++) {
                System.out.println("g()");
                Thread.yield();
            }
        }
    }
}

public class SyncObject {
    public static void main(String[] args) {
        final DualSynch ds = new DualSynch();
        new Thread() {
            public void run() {
                ds.f();
            }
        }.start();
        ds.g();
    }
}
/* Output: (Sample)
g()
f()
g()
f()
g()
f()
...
*/

这里f()同步于this，g()同步于syncObject，二者之间的同步是独立的。

7. Thread Local Storage

　　使用线程本地存储（Thread Local Storage）也可以解决多任务使用共享资源发生冲突的问题。线程本地存储可以在不同的线程上为同一个变量创建副本，各线程只使用自己的副本。这使得线程具有了状态。
　　java.lang.ThreadLocal用于创建线程存储：

import java.util.concurrent.*;
import java.util.*;

class Accessor implements Runnable {
    private final int id;
    public Accessor(int idn) { id = idn; }
    public void run() {
        while(!Thread.currentThread().isInterrupted()) {
            ThreadLocalVariableHolder.increment();
            System.out.println(this);
            Thread.yield();
        }
    }
    public String toString() {
        return "#" + id + ": " +
                ThreadLocalVariableHolder.get();
    }
}

public class ThreadLocalVariableHolder {
    private static ThreadLocal<Integer> value =
            new ThreadLocal<Integer>() {
                private Random rand = new Random(47);
                protected synchronized Integer initialValue() {
                    return rand.nextInt(10000);
                }
            };
    public static void increment() {
        value.set(value.get() + 1);
    }
    public static int get() { return value.get(); }
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        for(int i = 0; i < 5; i++)
            exec.execute(new Accessor(i));
        TimeUnit.MILLISECONDS.sleep(10);  // Run for a while
        exec.shutdownNow();         // All Accessors will quit
    }
}
/* Output
...
#2: 6776
#4: 1010
#1: 601
#0: 9300
#3: 1901

ThreadLocalVariableHolder的ThreadLocal<Integer> value具有随机的初始值，由打印可见，每个线程都有自己的ThreadLocal副本。ThreadLocal对象通常使用static。ThreadLocal对象只能通过get()和set()访问，get()会返回对应线程上的副本，set()则用于更新对应线程上的副本，并返回旧值。注意ThreadLocalVariableHolder的increment()和get()没有同步，因为ThreadLocal本身可以确保不会发生竞争。