傳統高併發依賴鎖 (synchronized、Lock、CAS)，上下文切換與爭用開銷大。LMAX Disruptor 證明：**單線程循環 (Event Loop) + 環形緩衝區** 可達百萬級 TPS。本文架構採用相同哲學：
┌─────────────┐ ┌──────────────────┐ ┌─────────────────┐ ┌──────────┐
│ Web Thread │────▶│ L1 Cache (Thread│────▶│ Disruptor │────▶│ DB │
│ (Producer) │ │ Local / Shared)│ │ RingBuffer │ │ Writer │
└─────────────┘ └──────────────────┘ └────────┬────────┘ └──────────┘
│
┌─────────────────────────────┘
▼
┌──────────────────┐
│ L2 Cache (Shared│
│ ConcurrentHashMap)│
└──────────────────┘
public class DbEvent {
public enum Op { INSERT, UPDATE, DELETE }
public final Op op;
public final String table;
public final Object pk; // 主鍵
public final Map<String, Object> columns; // 列值
public final long seq; // 全局遞增序號 (排序/去重用)
// 建構子、getter 省略
}public class DualCache<K, V> {
// L1: 僅單線程消費者寫入，多線程讀取 (無鎖讀)
private final Map<K, V> l1Cache = new HashMap<>();
// L2: 併發環境讀寫 (ConcurrentHashMap)
private final ConcurrentHashMap<K, V> l2Cache = new ConcurrentHashMap<>();
// 讀取：優先 L1，再 L2
public V get(K key) {
V v = l1Cache.get(key);
return v != null ? v : l2Cache.get(key);
}
// 併發環境寫入：直接寫 L2
public void putL2(K key, V value) { l2Cache.put(key, value); }
// 消費者寫入：寫 L1，異步刷新 L2
public void putL1AndFlushL2(K key, V value) {
l1Cache.put(key, value);
// 提交異步任務刷新 L2 (或由消費者批次刷新)
asyncFlush(key, value);
}
// 失效：同時清理兩級
public void invalidate(K key) {
l1Cache.remove(key);
l2Cache.remove(key);
}
}public class DbEventProcessor implements EventHandler<DbEvent> {
private final JdbcTemplate jdbc;
private final DualCache<String, Object> cache;
private final List<DbEvent> batch = new ArrayList<>(1000);
private static final AtomicLong SEQ = new AtomicLong();
@Override
public void onEvent(DbEvent event, long sequence, boolean endOfBatch) {
batch.add(event);
// 批次觸發條件：滿 1000 條 或 批次結束
if (batch.size() >= 1000 || endOfBatch) {
flushBatch();
}
}
private void flushBatch() {
if (batch.isEmpty()) return;
// 1. 排序：按 table -> pk -> seq，便於合併同主鍵操作
batch.sort(Comparator.comparing((DbEvent e) -> e.table)
.thenComparing(e -> e.pk.toString())
.thenComparingLong(e -> e.seq));
// 2. 去重優化：同 table+pk 只保留最後一條 UPDATE；有 DELETE 則前序無效
List<DbEvent> optimized = optimizeBatch(batch);
// 3. 批次執行
executeBatch(optimized);
// 4. 更新 L1 Cache (單線程安全)
for (DbEvent e : optimized) {
String cacheKey = e.table + ":" + e.pk;
if (e.op == DbEvent.Op.DELETE) {
cache.invalidate(cacheKey);
} else {
cache.putL1AndFlushL2(cacheKey, e.columns);
}
}
batch.clear();
}
private List<DbEvent> optimizeBatch(List<DbEvent> batch) {
List<DbEvent> res = new ArrayList<>();
DbEvent last = null;
for (DbEvent e : batch) {
if (last != null && last.table.equals(e.table) && last.pk.equals(e.pk)) {
if (e.op == DbEvent.Op.DELETE) {
res.remove(last); // 刪除操作使前序無效
last = e;
} else if (e.op == DbEvent.Op.UPDATE) {
last = e; // 只保留最後一次 UPDATE
}
} else {
res.add(e);
last = e;
}
}
return res;
}
}@RestController
public class OrderController {
private final RingBuffer<DbEvent> ringBuffer;
private final DualCache<String, Object> cache;
@PostMapping("/order")
public Result createOrder(@RequestBody OrderDTO dto) {
// 1. 業務邏輯、生成主鍵
String orderId = IdGenerator.next();
// 2. 寫入 L1 Cache (供後續同請求或其他請求立即讀取)
String cacheKey = "t_order:" + orderId;
Map<String, Object> cols = toColumns(dto);
cache.putL1AndFlushL2(cacheKey, cols); // 或直接 putL2
// 3. 發布事件到 RingBuffer (無鎖、極快)
long seq = ringBuffer.next();
try {
DbEvent event = ringBuffer.get(seq);
event.op = DbEvent.Op.INSERT;
event.table = "t_order";
event.pk = orderId;
event.columns = cols;
event.seq = SEQ.getAndIncrement();
} finally {
ringBuffer.publish(seq);
}
return Result.ok(orderId);
}
}public final class IdGenerator {
private static final AtomicLong SEQ = new AtomicLong(0);
public static long nextSeq() { return SEQ.getAndIncrement(); }
// 雪花算法變體 (含時間戳、機器碼、序號)
public static String nextId() { /* ... */ return ""; }
}| 優化 | 說明 |
|---|---|
| 批次合併同主鍵操作 | 減少 DB 往返，避免重複 UPDATE |
| DELETE 失效前序操作 | 同批次中出現 DELETE，前面的 INSERT/UPDATE 直接丟棄 |
| 異步落庫 | Web 線程不阻塞於 DB I/O，響應時間穩定 |
| L1 Cache 單線程寫 | 消費者線程獨佔寫入，讀取無鎖，保證強一致性 |
| L2 Cache 併發讀寫 | ConcurrentHashMap 支援高併發讀取 |
| 序號排序 | 保證跨線程操作的全局順序，支援冪等/去重 |
「大循環 + 雙級緩存 + 異步批次」是高併發無鎖寫入的經典模式。核心在於：**接受最終一致性、將隨機 I/O 轉為順序批次 I/O、用單線程換取無鎖吞吐**。實施時務必配合監控 (RingBuffer 積壓、DB 延遲、緩存命中率) 與熔斷降級機制。
