Java中的不变性和同步

Question

自从我阅读了Java Concurrency in Practice 这本书后，我想知道如何使用不变性来简化线程之间的同步问题。

我完全理解不可变对象是线程安全的。它的状态在初始化后不能改变，所以根本不可能有“共享可变状态”。但是不可变对象必须正确使用才能被认为在同步问题中有用。

以这段代码为例，它描述了一家拥有许多账户的银行，并公开了一种我们可以在账户之间转账的方法。

public class Bank {

    public static final int NUMBER_OF_ACCOUNT = 100;

    private double[] accounts = new double[NUMBER_OF_ACCOUNT];

    private Lock lock;
    private Condition sufficientFunds;

    public Bank(double total) {
        double singleAmount = total / 100D;
        for (int i = 0; i < NUMBER_OF_ACCOUNT; i++) {
            accounts[i] = singleAmount;
        }
        lock = new ReentrantLock();
        sufficientFunds = lock.newCondition();
    }

    private double getAdditionalAmount(double amount) throws InterruptedException {
        Thread.sleep(1000);
        return amount * 0.04D;
    }

    public void transfer(int from, int to, double amount) {
        try {
            // Not synchronized operation
            double additionalAmount = getAdditionalAmount(amount);
            // Acquiring lock
            lock.lock();
            // Verifying condition
            while (amount + additionalAmount > accounts[from]) {
                sufficientFunds.await();
            }
            // Transferring funds
            accounts[from] -= amount + additionalAmount;
            accounts[to] += amount + additionalAmount;
            // Signaling that something has changed
            sufficientFunds.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
   }   

   public double getTotal() {
       double total = 0.0D;
       lock.lock();
       try {
           for (int i = 0; i < NUMBER_OF_ACCOUNT; i++) {
               total += accounts[i];
           }
       } finally {
           lock.unlock();
       } 
       return total;
    }

    public static void main(String[] args) {
        Bank bank = new Bank(100000D);

        for (int i = 0; i < 1000; i++) {
            new Thread(new TransferRunnable(bank)).start();
        }
    }
}

在上面的例子中，来自Core Java Volume I一书，它通过显式锁使用同步。该代码显然难以阅读且容易出错。

我们如何使用不变性来简化上面的代码？我试图创建一个不可变的Accounts 类来保存帐户值，给Bank 类一个volatile 的Accounts 实例。但是我还没有达到我的目标。

谁能解释我是否可以使用不变性来简化同步？

---编辑---

可能我没有很好地解释自己。我知道不可变对象一旦创建就无法更改其状态。而且我知道，对于在 Java Memory Model (JSR-133) 中实现的规则，可以保证在初始化后看到不可变对象完全构造（有一些 distingua）。

然后我尝试使用这些概念从Bank 类中删除显式同步。我开发了这个不可变的Accounts 类：

class Accounts {
    private final List<Double> accounts;

    public Accounts(List<Double> accounts) {
        this.accounts = new CopyOnWriteArrayList<>(accounts);
    }

    public Accounts(Accounts accounts, int from, int to, double amount) {
        this(accounts.getList());
        this.accounts.set(from, -amount);
        this.accounts.set(to, amount);
    }

    public double get(int account) {
        return this.accounts.get(account);
    }

    private List<Double> getList() {
        return this.accounts;
    }
}

Bank 类的帐户属性必须使用volatile 变量发布：

private volatile Accounts accounts;

很明显，Bank类的传输方式会相应改变：

public void transfer(int from, int to, double amount) {
    this.accounts = new Accounts(this.accounts, from, to, amount);
}

使用不可变对象 (Accounts) 来存储类的状态 (Bank) 应该是一种发布模式，这在 JCIP 一书的第 3.4.2 段中有所描述。

但是，在某处仍然存在竞态条件，我不知道在哪里（以及为什么！！！）。

Answer 1

您的Account 值本质上是可变的（具有不可变余额的银行帐户不是很有用），但是您可以通过使用类似Actor Model 的方式封装可变状态来降低复杂性。你的Account 类实现了Runnable，每个Account 对象负责更新它的value。

public class Bank {
    // use a ConcurrentMap so that all threads will see updates to it
    private final ConcurrentMap<Integer, Account> accounts;
    private final ExecutorService executor = Executors.newCachedThreadPool();

    public void newAccount(int acctNumber) {
        Account newAcct = new Account();
        executor.execute(newAcct);
        accounts.put(acctNumber, newAcct);
    }

    public void transfer(int from, int to, double amount) {
        Account fromAcct = accounts.get(from);
        Account toAcct = accounts.get(to);
        if(fromAcct == null || toAcct == null) throw new IllegalArgumentException();
        fromAcct.transfer(amount, toAcct);
    }
}

public interface Message {
    public double getAmount();
}

public class Transfer implements Message {
    // initialize in constructor, implement getters
    private final double amount;
    private final Account toAcct;
}

public class Credit implements Message {
    // initialize in constructor, implement getters
    private final double amount;
}

public class Account implements Runnable {
    private volatile double value;
    private final BlockingQueue<Message> queue = new ArrayBlockingQueue<>(8);

    public void transfer(double amount, Account toAcct) {
        queue.put(new Transfer(amount, toAcct));
    }

    public void credit(double amount) {
        queue.put(new Credit(amount));
    }

    public void run() {
        try {
            while(true) {
                Message message = queue.take();
                if(message instanceof Transfer) {
                    Transfer transfer = (Transfer)message;
                    if(value >= transfer.getAmount()) {
                        value -= transfer.getAmount();
                        transfer.getToAcct().credit(transfer.getAmount());
                    } else { /* log failure */ }
                } else if(message instanceof Credit) {
                    value += message.getAmount();
                } else { /* log unrecognized message */ }
            }
        } catch(InterruptedException e) {
            return;
        }
    }
}

Account#transfer 和 Account#credit 方法可以从任何线程安全地调用，因为 BlockingQueue 是线程安全的。 value 字段仅在帐户的run 方法中修改，因此不存在并发修改的风险； value 必须是 volatile 以便所有线程都可以看到更新（您使用 ThreadPoolExecutor 来执行所有 Accounts 所以不能保证 Account's run 方法将执行每次都在同一个Thread）。

您还应该在执行转移之前将转移记录在Bank 类中，以便您可以从系统故障中恢复 - 如果在从账户被借记但在账户被贷记之前服务器崩溃，那么您需要一种在服务器恢复后重新建立一致性的方法。