用Java写一个MVCC例子

MVCC（Multi-Version Concurrency Control，多版本并发控制）是一种用于数据库管理系统中处理并发访问数据的技术。它允许事务在查询时看到数据的一个快照版本，而不是直接查看最新的提交数据。这种方法不仅提高了并发性能，还减少了锁定带来的开销，因为读操作不会阻塞写操作，反之亦然。

代码基于Java21

Data：数据，对应一个数据库表

public class Data {String value;int DB_TRX_ID; // 记录的事务idData next; // 下一个版本public Data(String value, int DB_TRX_ID, Data next) {this.value = value;this.DB_TRX_ID = DB_TRX_ID;this.next = next;}@Overridepublic String toString() {return "Data{" +"value='" + value + '\'' +", DB_TRX_ID=" + DB_TRX_ID +", next=" + next +'}';}public String getValue() {return value;}public void setValue(String value) {this.value = value;}public int getDB_TRX_ID() {return DB_TRX_ID;}public void setDB_TRX_ID(int DB_TRX_ID) {this.DB_TRX_ID = DB_TRX_ID;}public Data getNext() {return next;}public void setNext(Data next) {this.next = next;}public static Data of(Data data) {return new Data(data.getValue(), data.getDB_TRX_ID(), data.getNext());}
}

Transaction：事务，具有ACID特性，事务每次修改数据时会生成一个数据版本（快照）

public class Transaction {private int id;private String state = "create";private IsolationLevel isolationLevel = REPEATABLE_READ;enum IsolationLevel {READ_COMMITTED, REPEATABLE_READ}public Transaction(int id) {this.id = id;}public void beginTransaction() {this.state = "begin";}public void commit() {this.state = "commit";}public IsolationLevel getIsolationLevel() {return isolationLevel;}public void setIsolationLevel(IsolationLevel isolationLevel) {this.isolationLevel = isolationLevel;}public int getId() {return id;}public void setId(int id) {this.id = id;}public String getState() {return state;}public void setState(String state) {this.state = state;}
}

ReadView：读视图，MVCC提取数据的依据,记录并维护系统当前活跃的事务(未提交的)id

import java.util.Set;/*** m_ids:当前活跃的事务id集合* min_trx_id:当前活跃事务中最小的事务id* max_trx_id:系统将要分配给下一个事务的id* creator_trx_id:生成当前readview的事务id*/
public class ReadView {private Set<Integer> txs;private int min_trx_id;private int max_trx_id;private int creator_trx_id;// 构造函数，用于创建ReadView对象public ReadView(Transaction tx, Set<Integer> txs) {// 将传入的事务对象赋值给creator_trx_idthis.creator_trx_id = tx.getId();this.txs = txs;// 使用stream流获取传入的事务集合中的最小事务idmin_trx_id = txs.stream().min(Integer::compareTo).get();// 使用stream流获取传入的事务集合中的最大事务id，并加1max_trx_id = txs.stream().max(Integer::compareTo).get() + 1;}public Set<Integer> getTxs() {return txs;}public int getMin_trx_id() {return min_trx_id;}public int getMax_trx_id() {return max_trx_id;}public int getCreator_trx_id() {return creator_trx_id;}
}

DataStore：数据存储，维护数据版本链、活跃事务

import java.util.*;
import java.util.concurrent.TimeUnit;public class DataStore {// 数据版本链private final List<Data> dataList = new LinkedList<>();// 当前活跃事务，用于生成readview readview决定事务读到哪个版本的数据private final Set<Integer> activeTrxIds = new TreeSet<>();// 读视图缓存：隔离性为可重复读时返回第一次创建的读视图private final Map<Transaction, ReadView> map = new HashMap<>();private final StringBuffer sb = new StringBuffer();public ReadView getReadView(Transaction tx) {if (tx.getIsolationLevel() == Transaction.IsolationLevel.REPEATABLE_READ) {if (map.containsKey(tx)) {return map.get(tx);}ReadView readView = new ReadView(tx, activeTrxIds);map.put(tx, readView);return readView;} else if (tx.getIsolationLevel() == Transaction.IsolationLevel.READ_COMMITTED) {return new ReadView(tx, activeTrxIds);}return null;}public Data readData(Transaction tx) {tx.beginTransaction();
//        activeTrxIds.add(tx.getId());System.out.println("当前事务ID:" + tx.getId());System.out.println("活跃事务:" + activeTrxIds);ReadView rv = getReadView(tx);int creatorTrxId = rv.getCreator_trx_id();int minTrxId = rv.getMin_trx_id();int maxTrxId = rv.getMax_trx_id();Set<Integer> txs = rv.getTxs(); // 活跃事务集合System.out.println("视图: 创建者ID:" + creatorTrxId + " 最小事务ID:" + minTrxId + " 最大事务ID:" + maxTrxId + " 活跃事务:" + txs);Data first = dataList.getFirst(); // 当前记录int dbTrxId = first.getDB_TRX_ID();// 沿着版本链找数据 与读视图比较Data head = dataList.getFirst();while (head != null) {System.out.println("当前记录事务ID:" + dbTrxId);// 是否创建者if (dbTrxId == creatorTrxId) {System.out.println("创建者... 可见");return head;}// 是否比最小事务还小if (dbTrxId < minTrxId) {System.out.println("比最小事务还小... 可见");return head;}// 是否在中间if (dbTrxId >= minTrxId && dbTrxId <= maxTrxId && !txs.contains(dbTrxId)) {System.out.println("在活跃事务最小值和最大值中间并且不是活跃事务... 可见");return head;}head = head.getNext();if (head != null) {dbTrxId = head.getDB_TRX_ID();}System.out.println("轮询...");}tx.commit();return first;}/*** 每次事务更新数据时生成一个版本*/public Data update(Transaction tx, Data data, String val) {tx.beginTransaction();Data data1 = null;synchronized (dataList) {activeTrxIds.add(tx.getId());if (dataList.isEmpty()) {dataList.addFirst(data);sb.append(data.getDB_TRX_ID()).append("<-");} else {data1 = new Data(val, tx.getId(), dataList.getFirst());dataList.addFirst(data1);sb.append(data1.getDB_TRX_ID()).append("<-");}}try {TimeUnit.SECONDS.sleep(new Random().nextInt(5));} catch (InterruptedException e) {throw new RuntimeException(e);}tx.commit();System.out.println("事务:" + tx.getId() + "提交了...");activeTrxIds.remove(tx.getId());return data1;}public Data getFirstData() {return dataList.getFirst();}public String getTransactionChain() {String substring = sb.substring(0, sb.length() - 2);return substring;}}

Client：测试类 设置事务不同隔离级别会读到不同版本数据

import java.security.SecureRandom;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;public class Client {private static AtomicInteger counter = new AtomicInteger(1);public static void main(String[] args) {// 事务修改数据，记录进logDataStore log = new DataStore();for (int i = 1; i <= 10; i++) {Thread.ofPlatform().start(() -> {Transaction tx = new Transaction(counter.getAndIncrement());Data data = new Data("100", tx.getId(), null);String s = new SecureRandom().nextInt(100) + "";log.update(tx, data, s);});}// 延迟1秒读数据，等版本链生成try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}// 当前记录Data curr = log.getFirstData();System.out.println("当前记录:" + curr);System.out.println("事务处理链:" + log.getTransactionChain());// 读数据Transaction tx = new Transaction(new Random().nextInt(4) + 1);
//        tx.setIsolationLevel(Transaction.IsolationLevel.READ_COMMITTED);Data readData = log.readData(tx);System.out.println("选择快照:" + readData);System.out.println("+++++++++++++++++++++++++++++++++++");try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}Transaction tx2 = new Transaction(new Random().nextInt(9) + 1);
//        tx2.setIsolationLevel(Transaction.IsolationLevel.READ_COMMITTED);readData = log.readData(tx2);System.out.println("选择快照:" + readData);}
}

一种测试结果

事务:8提交了...
事务:7提交了...
事务:5提交了...
当前记录:Data{value='30', DB_TRX_ID=5, next=Data{value='88', DB_TRX_ID=3, next=Data{value='47', DB_TRX_ID=1, next=Data{value='62', DB_TRX_ID=10, next=Data{value='55', DB_TRX_ID=6, next=Data{value='14', DB_TRX_ID=8, next=Data{value='65', DB_TRX_ID=4, next=Data{value='56', DB_TRX_ID=9, next=Data{value='53', DB_TRX_ID=2, next=Data{value='100', DB_TRX_ID=7, next=null}}}}}}}}}}
事务处理链:7<-2<-9<-4<-8<-6<-10<-1<-3<-5
当前事务ID:1
活跃事务:[1, 2, 3, 4, 6, 9, 10]
视图: 创建者ID:1 最小事务ID:1 最大事务ID:11 活跃事务:[1, 2, 3, 4, 6, 9, 10]
当前记录事务ID:5
在活跃事务最小值和最大值中间并且不是活跃事务... 可见
选择快照:Data{value='30', DB_TRX_ID=5, next=Data{value='88', DB_TRX_ID=3, next=Data{value='47', DB_TRX_ID=1, next=Data{value='62', DB_TRX_ID=10, next=Data{value='55', DB_TRX_ID=6, next=Data{value='14', DB_TRX_ID=8, next=Data{value='65', DB_TRX_ID=4, next=Data{value='56', DB_TRX_ID=9, next=Data{value='53', DB_TRX_ID=2, next=Data{value='100', DB_TRX_ID=7, next=null}}}}}}}}}}
+++++++++++++++++++++++++++++++++++
事务:4提交了...
事务:1提交了...
当前事务ID:2
活跃事务:[2, 3, 6, 9, 10]
视图: 创建者ID:2 最小事务ID:2 最大事务ID:11 活跃事务:[2, 3, 6, 9, 10]
当前记录事务ID:5
在活跃事务最小值和最大值中间并且不是活跃事务... 可见
选择快照:Data{value='30', DB_TRX_ID=5, next=Data{value='88', DB_TRX_ID=3, next=Data{value='47', DB_TRX_ID=1, next=Data{value='62', DB_TRX_ID=10, next=Data{value='55', DB_TRX_ID=6, next=Data{value='14', DB_TRX_ID=8, next=Data{value='65', DB_TRX_ID=4, next=Data{value='56', DB_TRX_ID=9, next=Data{value='53', DB_TRX_ID=2, next=Data{value='100', DB_TRX_ID=7, next=null}}}}}}}}}}
事务:2提交了...
事务:9提交了...
事务:3提交了...
事务:10提交了...
事务:6提交了...

MVCC基本概念

基本概念
当前读 读取的是记录的最新版本
快照读 读取的是记录的历史版本Read Committed:每次select,都生成一个快照读Repeatable Read:开启事务后第一个select生成一个快照读,后续select都是基于这个快照读Serializable:快照读退化为当前读
MVCC:三个隐式字段、undo log、readview
1:三个隐式字段DB_TRX_ID: 最近修改事务的idDB_ROLL_PTR: 回滚指针,指向上一个版本数据DB_ROW_ID: 行id,唯一标识一条记录
2:undo log版本链不同事务或相同事务对同一条记录进行修改,会导致该记录的undo log生成一条记录版本链表,链表头部是最新的旧记录,尾部是最早旧的记录
3:readview 快照读SQL执行时,MVCC提取数据的依据,记录并维护系统当前活跃的事务(未提交的)idreadview包含四个核心字段:m_ids:当前活跃的事务id集合min_trx_id:当前活跃事务中最小的事务idmax_trx_id:系统将要分配给下一个事务的idcreator_trx_id:生成当前readview的事务id版本可见性判断1:如果被访问数据的DB_TRX_ID与readview的creator_trx_id相同,说明当前事务在访问自己修改过的记录,版本可见2:如果被访问数据的DB_TRX_ID小于readview的min_trx_id,说明当前事务在访问已经提交的事务修改过的记录,版本可见3:如果被访问数据的DB_TRX_ID大于readview的max_trx_id,说明当前事务在访问将来要执行的事务修改过的记录,版本不可见4:如果被访问数据的DB_TRX_ID在readview的min_trx_id和max_trx_id之间,说明当前事务在访问其他事务修改过的记录,版本不可见,需要通过DB_ROLL_PTR找到上一个版本数据,然后判断版本可见性如果DB_TRX_ID不在m_ids中是可以访问该版本的