thread1
    TSOChannelHandler#channelRead
        AbstractRequestProcessor#timestampRequest 接收 client 請求，創(chuàng)建 RequestEvent 并 publish
thread2
    AbstractRequestProcessor#onEvent 處理 RequestEvent 請求
        AbstractRequestProcessor#handleRequest
            PersistenceProcessorImpl#addTimestampToBatch 創(chuàng)建 PersistEvent,當(dāng) batch 滿了發(fā)送事件
thread3
    PersistenceProcessorHandler#onEvent 持久化事件處理

private void handleCommit(RequestEvent event) throws Exception {
    long startTimestamp = event.getStartTimestamp(); // startTimestamp
    Iterable<Long> writeSet = event.writeSet(); // 寫入集，存儲的是 cellIds
    Collection<Long> tableIdSet = event.getTableIdSet();
    boolean isCommitRetry = event.isCommitRetry();
    boolean nonEmptyWriteSet = writeSet.iterator().hasNext(); // 檢查寫集合是否為空，即事務(wù)是否有寫操作
    if (startTimestamp > lowWatermark &&
        !hasConflictsWithFences(startTimestamp, tableIdSet) &&
        !hasConflictsWithCommittedTransactions(startTimestamp, writeSet)) { // 檢查事務(wù)是否滿足提交條件,通過 hasConflictsWithCommittedTransactions 判斷是否有事務(wù)寫沖突
        // 可以進(jìn)行事務(wù)提交
        long commitTimestamp = timestampOracle.next(); // 獲取提交時(shí)間戳
        Optional<Long> forwardNewWaterMark = Optional.absent();
        if (nonEmptyWriteSet) { // 寫集合非空
            long newLowWatermark = lowWatermark;
            for (long r : writeSet) { // 遍歷寫集合中的每個(gè)元素，更新其最新的寫入時(shí)間戳，并計(jì)算新的低水位線
                long removed = hashmap.putLatestWriteForCell(r, commitTimestamp); // 更新 cellId 對應(yīng)的 commitTimestamp, 返回之前的 oldest commitTimestamp
                newLowWatermark = Math.max(removed, newLowWatermark); // 更新低水位線
            }
            if (newLowWatermark != lowWatermark) { // 更新低水位線
                lowWatermark = newLowWatermark;
                forwardNewWaterMark = Optional.of(lowWatermark);
            }
        }
        forwardCommit(startTimestamp, commitTimestamp, c, event.getMonCtx(), forwardNewWaterMark);  // 持久化 commit 請求
    } else { // 事務(wù)不滿足提交條件
        if (isCommitRetry) { // Re-check if it was already committed but the client retried due to a lag replying
            forwardCommitRetry(startTimestamp, c, event.getMonCtx());  // 若是提交重試，再次檢查是否已提交以避免因響應(yīng)延遲導(dǎo)致的重復(fù)提交
        } else {
            forwardAbort(startTimestamp, c, event.getMonCtx()); // 否則，中止事務(wù)
        }
    }
}

public long set(long key, long value) {
    final int index = index(key); // cellId 取模返回下標(biāo)，可能會沖突
    int oldestIndex = 0;
    long oldestValue = Long.MAX_VALUE;
    for (int i = 0; i < associativity; ++i) {
        int currIndex = 2 * (index + i); // 計(jì)算 key 下標(biāo)
        if (cache[currIndex] == key) { // 相同事務(wù) cellId, 替換場景
            oldestValue = 0;
            oldestIndex = currIndex;
            break;
        }
        if (cache[currIndex + 1] <= oldestValue) { // 沒找到相同的key.通過和 oldestValue 比較會將最小的 timestamp 剔除
            oldestValue = cache[currIndex + 1];
            oldestIndex = currIndex;
        }
    }
    // 替換最舊的鍵值對，將其更新為新的鍵值對
    cache[oldestIndex] = key;
    cache[oldestIndex + 1] = value;
    return oldestValue;
}

public long get(long key) {
    final int index = index(key);
    for (int i = 0; i < associativity; ++i) { // associativity 里存儲的元素key應(yīng)該是相同的
        int currIndex = 2 * (index + i); // 計(jì)算 key 的下標(biāo)
        if (cache[currIndex] == key) { // 找到 cache key
            return cache[currIndex + 1]; // 返回對應(yīng)的 value
        }
    }
    return 0;
}

創(chuàng)建 event,添加到 current batch
如果 current batch is full
    triggerCurrentBatchFlush