如何在并发事务中实现高性能对象隔离解决方案

Question

我正在创建此方法 issueTransfer 以实现从一个帐户到另一个帐户的转帐。

我的解决方案是：

public void issueTransfer(final int amount, final Account src,
        final Account dst) {
    /*
     * TODO implement issueTransfer using object-based isolation instead of
     * global isolation, based on the reference code provided in
     * BankTransactionsUsingGlobalIsolation. Keep in mind that isolation
     * must be applied to both src and dst.
     */

    isolated(src, dst, () -> {
        if (src.withdraw(amount)) {
            dst.deposit(amount);
        }
    });
}

全局隔离的解决方案是：

public void issueTransfer(final int amount, final Account src,
        final Account dst) {
    isolated(() -> {
        src.performTransfer(amount, dst);
    });
}

应用的全局和对象隔离方法定义如下：

public static void isolated(Runnable runnable) {
    isolatedManager.acquireAllLocks();

    try {
        runnable.run();
    } finally {
        isolatedManager.releaseAllLocks();
    }
}

public static void isolated(Object obj1, Object obj2, Runnable runnable) {
    Object[] objArr = new Object[]{obj1, obj2};
    isolatedManager.acquireLocksFor(objArr);

    try {
        runnable.run();
    } finally {
        isolatedManager.releaseLocksFor(objArr);
    }
}

辅助方法（获取和释放）是：

public void acquireAllLocks() {
    for(int i = 0; i < this.locks.length; ++i) {
        this.locks[i].lock();
    }
}

public void releaseAllLocks() {
    for(int i = this.locks.length - 1; i >= 0; --i) {
        this.locks[i].unlock();
    }
}

public void acquireLocksFor(Object[] objects) {
    TreeSet<Object> sorted = this.createSortedObjects(objects);
    Iterator var3 = sorted.iterator();

    while(var3.hasNext()) {
        Object obj = var3.next();
        int lockIndex = this.lockIndexFor(obj);
        this.locks[lockIndex].lock();
    }
}

public void releaseLocksFor(Object[] objects) {
    TreeSet<Object> sorted = this.createSortedObjects(objects);
    Iterator var3 = sorted.iterator();

    while(var3.hasNext()) {
        Object obj = var3.next();
        int lockIndex = this.lockIndexFor(obj);
        this.locks[lockIndex].unlock();
    }
}

private int lockIndexFor(Object obj) {
    return Math.abs(obj.hashCode()) % 64;
}

private TreeSet<Object> createSortedObjects(Object[] objects) {
    TreeSet<Object> sorted = new TreeSet(new Comparator<Object>() {
        public int compare(Object o1, Object o2) {
            return IsolatedManager.this.lockIndexFor(o1) - IsolatedManager.this.lockIndexFor(o2);
        }
    });
    Object[] var3 = objects;
    int var4 = objects.length;

    for(int var5 = 0; var5 < var4; ++var5) {
        Object obj = var3[var5];
        sorted.add(obj);
    }

    return sorted;
}

如您所见，我应该按照文档的要求应用第二种方法（对象隔离）。测试顺利通过：

public void testObjectIsolation() {
    testDriver(new BankTransactionsUsingGlobalIsolation());
    final long globalTime = testDriver(
            new BankTransactionsUsingGlobalIsolation());

    testDriver(new BankTransactionsUsingObjectIsolation());
    final long objectTime = testDriver(
            new BankTransactionsUsingObjectIsolation());
    final double improvement = (double)globalTime / (double)objectTime;

    final int ncores = getNCores();
    final double expected = (double)ncores * 0.75;
    final String msg = String.format("Expected an improvement of at " +
            "least %fx with object-based isolation, but saw %fx", expected,
            improvement);
    assertTrue(msg, improvement >= expected);
}

但是用于评估的平台说我没有通过 2 核或 4 核的测试。取决于我何时进行，有时我会通过其中一项测试（我假设是 2 个核心测试。

正如您从我通过的测试中看到的那样，我的对象隔离解决方案比我的全局隔离更快（每个内核的 1:0.75 比率）。 是平台故障还是我的代码可以改进？我试过使用lock、unlock和trylock，但我的解决方案似乎工作得更快但还不够。

Answer 1

您可以尝试 Brian Goetz 和 Tim Pierels 在 "Java Concurrency in Practice"

书中的这种方法

http://jcip.net/listings/InduceLockOrder.java

Answer 2

您可以将 Taooka distributed lock manager 与一些 SQL 数据库一起使用来编写如下代码：

if taooka.Lock(srcUser.ID, time.Minute) {

    if srcUser.Balance < transferSum {
        taooka.Unlock(srcUser.ID)
        return "Not enough money"
    }

    if taooka.Lock(dstUser.ID, time.Minute) {

        srcSQLServer = GetSQLServerFor(srcUser)
        dstSQLServer = GetSQLServerFor(dstUser)

        transferID = journalSQLServer.do("INSERT INTO money_transfer SET start_time=NOW()")

        srcSQLServer.do("INSERT INTO money_log SET transfer_id=", transferID, ", user_id=", srcUser.ID, ", sum=", -transferSum)
        dstSQLServer.do("INSERT INTO money_log SET transfer_id=", transferID, ", user_id=", dstUser.ID, ", sum=", transferSum)

        journalSQLServer.do("DELETE FROM money_transfer WHERE id=", transferID)

        taooka.Unlock(dstUser.ID)
        taooka.Unlock(srcUser.ID)

        return "Ok"
    }

    taooka.Unlock(srcUser.ID)
    return "wait please..."

}
else {
    return "wait please..."
}

和运行在后台此代码：

for {
    partialTransferID = journalSQLServer.do("SELECT id FROM money_transfer WHERE start_time < DATE_SUB(NOW(), INTERVAL 1 MINUTE) LIMIT 1")

    if partialTransferID > 0 {
        foreach sql in SQLServers {
            sql.do("DELETE FROM money_log WHERE transfer_id=", partialTransferID)
        }
        journalSQLServer.do("DELETE FROM money_transfer WHERE id=", partialTransferID)
        continue
    }

    Sleep(time.Second)
}

但不幸的是，Taooka 仍然没有 Java 绑定。

如何在并发事务中实现高性能对象隔离解决方案

How do I achieve a performant object isolation solution in concurrent transactions

java

concurrency

locking

transaction-isolation