如何实现列级乐观锁

Question

    目前的并发控制机制，悲观锁和乐观锁，都是基于行来实现，对于一些很多模块都会访问并修改的数据（修改的列并不一样），这种粒度是不够的，因为不同模块的业务，修改的字段并不一定是同一个（或者同一类）。
在这些场景下，基于列的并发就比较有意义了，可以很大的提高并发事务执行能力。本文要介绍的乐观锁机制，是基于内存实现的，锁的获取和并发检测，完全是在内存中，不会和数据库有任何交互，所以效率会非常高。首先是增加事务拦截器，在事务开始时登记事务信息，在事务结束后清除事务信息
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
// 获取事务信息
TransactionStatus transactionStatus = TransactionUtils
.getCurrentTransactionStatus();
TransactionInfo transactionInfo = null;
boolean bind = false;
if (transactionStatus != null && transactionStatus.isNewTransaction()) {
transactionInfo = new TransactionInfo(invocation,transactionStatus);
transactionInfoManager.addTransactionInfo(transactionInfo);
transactionInfo.bindToThread();
bind = true;
try {
TransactionSynchronizationManager
.registerSynchronization(new OptimisticLockTransactionSynchronizationAdapter(
optimisticLockManager, transactionInfo));
} catch (Exception e) {
e.printStackTrace();
}
}
Object rv;
try {
rv = invocation.proceed();
} finally {
if(transactionInfo != null){
transactionInfoManager.removeTransactionInfo(transactionInfo);
}
if (bind) {
transactionInfo.restoreThreadLocalStatus();
}
}
return rv;
}
OptimisticLockTransactionSynchronizationAdapter的代码如下
public class OptimisticLockTransactionSynchronizationAdapter extends
TransactionSynchronizationAdapter { private TransactionInfo transactionInfo = null;
private OptimisticLockManager optimisticLockManager; public OptimisticLockTransactionSynchronizationAdapter(
OptimisticLockManager optimisticLockManager,
TransactionInfo transactionInfo) {
super();
this.transactionInfo = transactionInfo;
this.optimisticLockManager = optimisticLockManager;
} @Override
public void afterCompletion(int status) {
// 事务完成时，清除事务相关的锁
optimisticLockManager.releaseTransactionLocks(transactionInfo);
}
}
TransactionInfo为事务信息类，在当前线程会有一个栈，每次新建事务时会新生成一个TransactionInfo对象，并且入栈。当事务结束后，会将当前TransactionInfo出栈，如下图：
以下是相关的核心类
类 OptimisticLockManager
属性 private ConcurrentHashMap<String, EntityLocksInfo> entityToLocksMap
public void requestLock(String entity, String rowId,String colId) {

EntityLocksInfo entityLocksInfo = entityToLocksMap.get(entity);

if (entityLocksInfo == null) {
EntityLocksInfo newEntityLocksInfo = new EntityLocksInfo(entity);
entityLocksInfo = entityToLocksMap.putIfAbsent(entity,
newEntityLocksInfo);
if (entityLocksInfo == null) {
entityLocksInfo = newEntityLocksInfo;
}
}

entityLocksInfo.requestLock(rowId,colId);
}类 EntityLocksInfo
属性 private ConcurrentHashMap<String, RowLocksInfo> idToRowLocksInfoMap
public void requestLock(String rowId, String colId) {
TransactionInfo transactionInfo = TransactionUtils
.currentTransactionInfo();
if (transactionInfo == null) {
throw new OptimisticLockNoTransactionException();
}
RowLocksInfo rowLocksInfo = idToRowLocksInfoMap.get(rowId); if (rowLocksInfo == null) {
RowLocksInfo newRowLocksInfo = new RowLocksInfo(entity);
rowLocksInfo = idToRowLocksInfoMap.putIfAbsent(rowId,
newRowLocksInfo);
if (rowLocksInfo == null) {
rowLocksInfo = newRowLocksInfo;
}
}
rowLocksInfo.requestLock(colId);
}类 RowLocksInfo
属性 private ConcurrentHashMap<String, OptimisticLock> colIdToLockMap
方法 public void requestLock(String colId);
public void requestLock(String colId) {
TransactionInfo transactionInfo = TransactionUtils
.currentTransactionInfo();
if (transactionInfo == null) {
throw new OptimisticLockNoTransactionException();
}
OptimisticLock existingOptimisticLock = colIdToLockMap.get(colId);
if (existingOptimisticLock == null) {
OptimisticLock newOptimisticLock = new OptimisticLock();
newOptimisticLock.setId(colId);
newOptimisticLock.setTransactionInfo(transactionInfo); existingOptimisticLock = colIdToLockMap.putIfAbsent(colId,
newOptimisticLock);
if (existingOptimisticLock == null) {
transactionInfo.addLock(newOptimisticLock);
}
}
if (existingOptimisticLock != null) {
if (existingOptimisticLock.getTransactionInfo() != transactionInfo) {
throw new OptimisticLockConflictException(colId,
transactionInfo);
}
} }类 OptimisticLock 为真正的所对象，记录了锁所属的实体名，行id，列id，等等private String entity;
private String rowId;
private String colId;
如何在业务类中使用在业务manager中注入OptimisticLockManager
例如有一个产品管理系统，现在有3个产品，数据库中数据如下id name stock salesAmount   1 intel 9900K 1000 500
2 intel 9700K 1000 300
3 intel 8700K 10         10000现在有订单事务，购买intel 9900K，数量10个
那么需要锁定stock和salesAmountoptimisticLockManager.requestLock("product", "1","stock");
optimisticLockManager.requestLock("product", "1","salesAmount");然后执行逻辑
update product set stock = stock-10 where id=1;
update product set salesAmount = salesAmount+10 where id=1;对于入库单（产品入库/进货），入库intel 9900K，数量10个，只需要修改库存，而不需要修改销售量
那么只需要锁定stock
optimisticLockManager.requestLock("product", "1","stock");
然后执行逻辑
update product set stock = stock+10 where id=1;这里只是API调用说明，此处的订单事务和入库事务，其实是不能并发执行（因为修改了共同的列stock），但是其他场景
如果修改的列是不冲突的，那就完全可以并发执行当多个事务请求同一个表，同一条记录的同个属性时，后请求的将会抛出异常（可以将entity,rowId,colId以及事务名字都提示出来），例如上面的2个事务，如果事务2先开始（入库事务）并且未提交时，事务1（订单事务）也同时执行，将会提示"入库操作正在修改id为1的产品记录的库存，请稍后重试!"

Answer 1

赞！PS：其实我看不懂 java 。

Answer 2

没说明出处，没说明使用的框架，也没给出具体使用的jar包。没有具体的用于测试的代码，无法在本机上测试，我首先对这些代码的正确性持怀疑态度。更关键的是，不管是效率，功能性，还是方便程度，都不觉得比oracle原生的实现有什么优点。

Answer 3

框架是用的spring，事务代理加个拦截器，用来做事务同步处理。正确性毋庸置疑，我们产品已经在很多项目使用了，你说和原生的数据库比有什么优点，1，无须最后提交修改时才能检测，可以提前在内存中检测，比原生的效率高（不会导致执行事务一部分后回滚，浪费不必要的资源）。2，数据库并不支持列级别的并发，对于有些字段很多，修改它的业务比较多的表，列级别的并发是有意义的。3，对于冲突，可以知道是和哪些业务冲突（具体的事务和表，字段），数据库是不会有这么详细的信息的。4，对于有些场景，比如根据父表状态来对子表进行处理（修改），但不更新父表，此时用数据库的乐观锁无法实现

Answer 4

至于效率，就只取决于ConcurrentHashMap的效率，可以参考这篇文章 https://stackoverflow.com/questions/1378310/performance-concurrenthashmap-vs-hashmap  的测试结果，add的效率是100万次/100ms，get效率是100万次/22ms，remove是100万次/23ms，
对于一般的事务用到的锁不过几个或者数十个，也就是说，一秒钟平均支持数万个事务毫无问题，这里并不会成为瓶颈。如果一开始就分配系统支持的最大数量的锁，那么效率将会更高。