问题描述
我正在尝试不使用锁而是使用多版本并发控制来解决多线程银行帐户问题*。工作正常有点慢。如何加快速度？
（*）我有5个用户，每个用户以200开始-每个用户随机提取100，然后将100存入另一个用户拥有的另一个银行帐户。我预计到运行结束时银行存款余额总计为1000。不应损失或创造任何金钱。这部分与下面的实现相同。
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
public class ConcurrentWithdrawer {
private Map<String,Integer> database = new HashMap<>();
private int transactionCount = 0;
private final List<Transaction> transactions = Collections.synchronizedList(new ArrayList<>());
public static void main(String[] args) {
try {
new ConcurrentWithdrawer().run();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static int getRandomNumberInRange(int min,int max) {
if (min >= max) {
throw new IllegalArgumentException("max must be greater than min");
}
Random r = new Random();
return r.nextInt((max - min) + 1) + min;
}
public void run() throws ExecutionException,InterruptedException {
int startAmount = 200;
int numberAccounts = 5;
int totalMoney = 0;
for (int i = 0; i < numberAccounts; i++) {
database.put(String.format("account%d",i),startAmount);
totalMoney += startAmount;
}
ThreadPoolExecutor executor =
(ThreadPoolExecutor) Executors.newFixedThreadPool(5);
List<Future> futures = new ArrayList<Future>();
for (int i = 0; i < 5; i++) {
futures.add(executor.submit(new Callable<Integer>() {
@Override
public Integer call() {
for (int j = 0; j < 5; j++) {
Transaction transaction = beginTransaction(transactions,database);
transaction.read("fromBalance","fromAccountName",(context) -> {
int fromAccount = getRandomNumberInRange(0,4);
String fromAccountName = String.format("account%d",fromAccount);
return fromAccountName;
}).read("toBalance","toAccountName",(context) -> {
int toAccount = getRandomNumberInRange(0,4);
String toAccountName = String.format("account%d",toAccount);
while (toAccountName.equals(context.lookupName("fromAccountName"))) {
toAccount = getRandomNumberInRange(0,4);
toAccountName = String.format("account%d",toAccount);
}
return toAccountName;
}).write("fromAccountName",(writeContext) -> {
int difference;
TransactionContext context = writeContext.context;
if (context.get("fromBalance") >= 100) {
difference = 100;
} else {
difference = 0;
}
context.write(writeContext.writeStep,context.get("fromBalance") - difference);
context.put("difference",difference);
}).write("toAccountName",(writeContext) -> {
TransactionContext context = writeContext.context;
context.write(writeContext.writeStep,context.get("toBalance") + context.get("difference"));
}).commit();
}
int foundMoney = 0;
for (int j = 0; j < numberAccounts; j++) {
Integer foundMoney1;
String account = String.format("account%d",j);
foundMoney1 = database.get(account);
foundMoney += foundMoney1;
}
return foundMoney;
}
}));
}
List<Integer> monies = new ArrayList<>();
for (Future f : futures) {
int foundMoney = (Integer) f.get();
monies.add(foundMoney);
}
System.out.println("Totals while running");
for (Integer money : monies) {
System.out.println(money);
}
System.out.println("Expected money");
System.out.println(totalMoney);
System.out.println("Final money");
int foundMoney = 0;
for (int j = 0; j < numberAccounts; j++) {
Integer foundMoney1;
foundMoney1 = database.get(String.format("account%d",j));
System.out.println(String.format(String.format("account%d %d",j,foundMoney1)));
foundMoney += foundMoney1;
}
System.out.println(foundMoney);
executor.shutdown();
}
private Transaction beginTransaction(List<Transaction> transactions,Map<String,Integer> database) {
transactionCount = transactionCount + 1;
Transaction transaction = new Transaction(transactions,transactionCount,database);
this.transactions.add(transaction);
return transaction;
}
private class Transaction {
public Long readTimestamp = 0L;
public Long writeTimestamp = 0L;
public List<String> readTargets = new ArrayList<>();
private List<Transaction> transactions;
private final int id;
private Map<String,Integer> database;
private List<TransactionStep> steps = new ArrayList<>();
private TransactionContext transactionContext = new TransactionContext();
private boolean active = true;
private boolean cancel = false;
private long transactionFinish;
private long transactionStart;
private int reread;
private boolean valid;
public Transaction(List<Transaction> transactions,int id,Integer> database) {
this.transactions = transactions;
this.id = id;
this.database = database;
}
public Transaction read(String field,String name,Function<TransactionContext,String> keyGetter) {
ReadStep step = new ReadStep(this,field,keyGetter);
steps.add(step);
transactionContext.registerStep(name,step);
return this;
}
public Transaction write(String fieldName,Consumer<WriteContext> writer) {
steps.add(new WriteStep(this,fieldName,writer));
return this;
}
public boolean invalid() {
long largestWrite = 0L;
long largestRead = 0L;
List<Transaction> cloned = new ArrayList<>(transactions);
cloned.sort(new Comparator<Transaction>() {
@Override
public int compare(Transaction o1,Transaction o2) {
return (int) (o1.transactionStart - o2.transactionStart);
}
});
for (Transaction transaction : cloned) {
ArrayList<TransactionStep> clonedSteps = new ArrayList<>(transaction.steps);
for (TransactionStep step : clonedSteps) {
for (TransactionStep thisStep : steps) {
if (step instanceof ReadStep && thisStep instanceof ReadStep) {
ReadStep thisReadStep = (ReadStep) thisStep;
ReadStep readStep = (ReadStep) step;
if (thisReadStep.key.equals(readStep.key)) {
if (thisReadStep.timestamp > readStep.timestamp) {
return true;
}
}
}
if (step instanceof WriteStep && thisStep instanceof WriteStep) {
WriteStep thisWriteStep = (WriteStep) thisStep;
WriteStep writeStep = (WriteStep) step;
if (thisWriteStep.timestamp > writeStep.timestamp) {
return true;
}
}
}
}
}
return false;
}
public void commit() {
boolean needsRunning = true;
int retryCount = 0;
transactionStart = System.nanoTime();
while (needsRunning || invalid()) {
readTimestamp = 0L;
writeTimestamp = 0L;
readTargets.clear();
retryCount++;
active = true;
for (TransactionStep step : steps) {
step.run(transactionContext);
}
needsRunning = false;
if (cancel) {
needsRunning = true;
cancel = false;
}
}
System.out.println(String.format("Retry count was %d",retryCount));
for (TransactionStep step : steps) {
if (step instanceof ReadStep) {
String key = ((ReadStep) step).key;
Integer value = transactionContext.context.get(key);
database.put(key,value);
}
}
transactions.remove(this);
transactionFinish = System.nanoTime();
}
}
private interface TransactionStep {
TransactionContext run(TransactionContext context);
}
private class ReadStep implements TransactionStep {
private final String field;
private final Function<TransactionContext,String> keyGetter;
private boolean activated;
private String key;
public long timestamp;
Transaction transaction;
public ReadStep(Transaction transaction,String field,Function keyGetter) {
this.transaction = transaction;
this.field = field;
this.keyGetter = keyGetter;
this.activated = false;
}
public TransactionContext run(TransactionContext context) {
if (!activated) {
key = (String) this.keyGetter.apply(context);
}
activated = true;
timestamp = System.nanoTime();
context.put(field,database.get(key));
if (transaction.readTimestamp == 0L) {
transaction.readTimestamp = timestamp;
}
transaction.readTargets.add(key);
return context;
}
}
private class TransactionContext {
public final HashMap<String,Integer> context;
private Map<String,ReadStep> readSteps = new HashMap<>();
public TransactionContext() {
this.context = new HashMap<>();
}
public void registerStep(String name,ReadStep readStep) {
readSteps.put(name,readStep);
}
public void put(String field,Integer integer) {
this.context.put(field,integer);
}
public String lookupName(String name) {
return readSteps.get(name).key;
}
public void write(WriteStep writeStep,Integer newValue) {
String key = lookupName(name);
writeStep.key = key;
context.put(key,newValue);
}
public Integer get(String field) {
return this.context.get(field);
}
}
private class WriteStep implements TransactionStep {
public String key;
private boolean activated;
private String fieldName;
private final Consumer<WriteContext> writer;
public long timestamp;
Transaction transaction;
public WriteStep(Transaction transaction,String fieldName,Consumer<WriteContext> writer) {
this.transaction = transaction;
this.fieldName = fieldName;
this.writer = writer;
activated = false;
}
@Override
public TransactionContext run(TransactionContext context) {
timestamp = System.nanoTime();
transaction.writeTimestamp = timestamp;
writer.accept(new WriteContext(this,context));
activated = true;
return context;
}
}
private class WriteContext {
private final WriteStep writeStep;
private final TransactionContext context;
public WriteContext(WriteStep writeStep,TransactionContext context) {
this.writeStep = writeStep;
this.context = context;
}
}
}
我收到的输出：
Retry count was 4511
Retry count was 671
Retry count was 5956
Retry count was 140
Retry count was 3818
Retry count was 3102
Retry count was 34
Retry count was 580
Retry count was 106
Retry count was 46
Retry count was 22
Retry count was 11478
Retry count was 199
Retry count was 33
Retry count was 715
Retry count was 263
Retry count was 6186
Retry count was 6846
Retry count was 7012
Retry count was 301
Retry count was 93
Retry count was 148
Retry count was 11
Retry count was 355
Retry count was 7
Totals while running
1000
1000
1000
1000
1000
Expected money
1000
Final money
account0 200
account1 700
account2 100
account3 0
account4 0
1000
BUILD SUCCESSFUL in 515ms
如何提高效率？我确信Postgres不允许交易在接受之前运行数千次。
有人说代码被混淆了。此代码读取值（并记录读取），就像SQL读取语句一样。该代码需要访问要访问的字段的名称以及要访问的实际值。这就是为什么代码是这样写的。下面的代码说：读取此函数生成的字段名称，将该名称存储到fromAccountName中，并将结果值存储在fromBalance中。
transaction.read("fromBalance",fromAccount);
return fromAccountName;
})
解决方法
要测试的系统的简短描述，从问题中扣除并查看代码：
- 帐户保存在一个“数据库”中，该数据库由映射到帐户值的帐户名称组成。
- 帐户之间的交易本质上应该始终是原子性的，因此，除了最初的200英镑以外，没有其他货币会被创建或销毁。
- 另一个假设是帐户不能具有负值。
- 在您的示例代码中，数据库被实现为
Map<String,Integer>。
如下所述，可以通过使用帐户数据库BankAccounts解决问题转移的原子性。这不包括测试代码，但解决了一致性问题。
public class BankAccounts {
/**
* The number of retries for each transfer
*/
public static final int RETRIES = 10;
/**
* The account database
*/
private Map<String,AtomicInteger> accounts = new HashMap<>();
/**
* Creates accounts with each 200 initial balance.
*/
public BankAccounts() {
// fill the accounts initially
// Left as an exercice to the reader
}
/**
* Return a set with all account names.
*/
public Set<String> accountNames() {
return Collections.unmodifiableSet(accounts.keySet());
}
/**
* Get the balance value for the specified account.
*/
public int getBalanceFor(String accountName) {
AtomicInteger account = accounts.get(accountName);
return account != null ? account.get() : 0;
}
/**
* Atomically transfers an amount from one account to another and returns {@code true} if that was successful.
*/
public boolean transfer(String fromAccountName,String toAccountName,int amount) {
AtomicInteger fromBalance = accounts.get(fromAccountName);
AtomicInteger toBalance = accounts.get(toAccountName);
if (amount < 0 || fromBalance == null || toBalance == null) {
return false;
}
for (int retry = 0; retry < RETRIES; retry++) {
int fromValue = fromBalance.get(); // get from-account balance value
if (fromValue >= amount) { // check if enough money
if (fromBalance.compareAndSet(fromValue,fromValue - amount)) {
toBalance.addAndGet(amount); // Adding money is always allowed ;-)
return true;
} else {
// value of fromBalance was changed concurrently
Thread.yield(); // optional.
}
}
}
return false;
}
}
设置时间 控制面板
错误1：Request method ‘DELETE‘ not supported 错误还原：...