kafka高吞吐持久化方案（1）

需求背景

kafka消费消息丢入线程池，upsert数据库（mysql）。随着线程池并发加大，mq tps升高，锁冲突会加剧

优化方案

核心思路

方案1

按照唯一key有序缓存至ConcurrentSkiplistMap。多线程并发截取Map进行batchUpsert

• 利用 ConcurrentSkipListMap<UniqueKey, Data> 实现 按唯一索引有序缓存。
• 每个线程从 map.subMap(…) 中获取一个局部有序子集，例如每次提取 N=100 条。
• 执行一次批量 upsert（避免单条写导致频繁锁竞争）。

✅ 优点

🧠 实现建议

1. 定义唯一键类

public class UniqueKey implements Comparable<UniqueKey> {
private final Long uid;
private final Long cid;

// equals 和 hashCode 用于 map 去重
@Override
public int compareTo(UniqueKey o) {
int cmp = uid.compareTo(o.uid);
if (cmp != 0) return cmp;
return cid.compareTo(o.cid);
}

// equals(), hashCode(), constructor 省略
}

2. 定义缓存结构

ConcurrentSkipListMap<UniqueKey, YourDataDTO> buffer = new ConcurrentSkipListMap<>();

3. 批量提取 & 删除

// 每个线程循环执行，提取 batchSize 条
int batchSize = 100;
List<Map.Entry<UniqueKey, YourDataDTO>> batch = new ArrayList<>();

// 多线程批量抢占式消费保证读写原子性
for (UniqueKey key : buffer.keySet()) {
buffer.computeIfPresent(key, (k, v) -> {
if (batch.size() < batchSize) {
batch.add(Map.entry(k, v));
return null; // 返回 null 表示删除该 key
} else {
return v; // 保留该 key
}
});
if (batch.size() >= batchSize) {
break;
}
}

// 批量 upsert 到数据库
if (!batch.isEmpty()) {
List<YourDataDTO> records = batch.stream()
.map(Map.Entry::getValue)
.collect(Collectors.toList());

yourDao.batchUpsert(records);
}

🧱 数据库层优化建议

INSERT INTO your_table(uid, cid, xxx)
VALUES (?, ?, ?), (?, ?, ?)
ON DUPLICATE KEY UPDATE xxx=VALUES(xxx)

INSERT … ON DUPLICATE KEY UPDATE col = IF(VALUES(col) > col, VALUES(col), col)

⚠️ 注意事项

🧩 进阶建议

• 可以用 ReentrantLock 控制同一 uid 分组处理（hash分片思想）。
• 或者用 Disruptor / RingBuffer 等高性能缓冲结构替代 SkipListMap。

方案2

Disruptor多 EventHandler + Routing

• 每个 EventHandler 是 独占线程；
• 可以让某个 handler 只处理 (uid+cid).hash % N == i 的事件；
• 在 handler 中**同步调用 ****handleUpsert()**，无需线程池

✅ 核心目标

按照 uid+cid 为唯一键，做到 同一个 **uid+cid** 的 upsert 严格顺序、有序处理，同时允许不同键的并发处理。

✳️ 实现思路概览（支持局部有序）

class UpsertEvent {
UniqueKey key; // uid + cid
YourDataDTO data;
}

Disruptor<UpsertEvent> disruptor = new Disruptor<>(
UpsertEvent::new,
ringBufferSize,
Executors.defaultThreadFactory(),
ProducerType.MULTI,
new BlockingWaitStrategy()
);

具体方式有两种：

🚀 方式一：多 EventHandler + Routing（推荐）

拆分思路：

• 创建 N 个 EventHandler（例如 8 个）。
• 根据 key.hashCode() % N 将事件路由到某个 handler。
• 每个 handler 内部队列是串行消费的，所以同一个 key 总在同一个 handler 中串行处理，不同 key 可并发。

示例代码（概念版）：

int handlerCount = 8;
for (int i = 0; i < handlerCount; i++) {
final int index = i;
disruptor.handleEventsWith((event, sequence, endOfBatch) -> {
if ((event.getKey().hashCode() & Integer.MAX_VALUE) % handlerCount == index) {
// ✅ 在分区线程中直接执行
// 处理这个分区的数据
handleUpsert(event.getData());
}
});
}

🧠 优点：线程复用良好，避免全局串行，同 key 保证严格有序，不同 key 并发。

🌀 方式二：自定义 WorkPool + 内部分片缓存（低级方案）

这种方式使用 WorkPool 并发消费，但自己在每个消费者内部管理 Map<UniqueKey, Queue<Data>>，对于同一个 key，必须队列顺序处理，不如方式一清晰可靠。

🔐 为什么这样能保证局部有序？

• Disruptor 本身就是有序队列：其事件是严格顺序发布的（ringBuffer 顺序）。
• 我们人为将同一个 key 始终路由到一个 handler（分区）：
  • handler 是串行执行的 → 保证该 key 的操作严格按照到达顺序执行。
• 不同 key 的 handler 不同 → 允许并发处理。

方案3

使用ReentrantLock实现分段锁

ReentrantLock 方案是一个更轻量、易于控制的方式，特别适用于这种：

按 uid+cid 为唯一索引，需要实现相同 key 顺序处理、不同 key 可并发处理的 upsert 场景。

✅ 目标：基于 ReentrantLock 实现局部有序 upsert

• 同一个 (uid, cid)：串行处理（避免唯一索引冲突）
• 不同的 (uid, cid)：可以并发处理（提高吞吐）
• 可支持 batchUpsert（提高 DB 性能）

🧩 实现思路

一、定义一个可复用的锁容器（分段锁）

我们不为每个 key 分配一个 ReentrantLock（太耗内存），而是用固定数量的 ReentrantLock 数组，根据 uid+cid hash 映射到某一个锁。

public class KeyLockManager {

private final int lockSize = 256; // 根据并发量设置
private final ReentrantLock[] locks = new ReentrantLock[lockSize];

public KeyLockManager() {
for (int i = 0; i < lockSize; i++) {
locks[i] = new ReentrantLock();
}
}

public ReentrantLock getLock(Object key) {
int index = (key.hashCode() & Integer.MAX_VALUE) % lockSize;
return locks[index];
}
}

二、在并发 upsert 中使用锁保护同一组 key

KeyLockManager lockManager = new KeyLockManager();

public void safeUpsert(YourDataDTO data) {
UniqueKey key = new UniqueKey(data.getUid(), data.getCid());
ReentrantLock lock = lockManager.getLock(key);

lock.lock();
try {
// 这里可以进行 batchUpsert 也可以单条
doUpsert(data);
} finally {
lock.unlock();
}
}

如果希望每次一个线程处理一组数据（而不是一条），可以在锁内操作 batch：

三、批量获取并处理同一个 key 的多个数据

public void safeBatchUpsert(List<YourDataDTO> dataList) {
// 分组，同一个 key 一起处理
Map<UniqueKey, List<YourDataDTO>> groupMap = dataList.stream()
.collect(Collectors.groupingBy(d -> new UniqueKey(d.getUid(), d.getCid())));

for (Map.Entry<UniqueKey, List<YourDataDTO>> entry : groupMap.entrySet()) {
UniqueKey key = entry.getKey();
List<YourDataDTO> records = entry.getValue();

ReentrantLock lock = lockManager.getLock(key);
lock.lock();
try {
batchUpsert(records);
} finally {
lock.unlock();
}
}
}

⚠️ 注意事项

🧠 为什么这个方案可行？

• ReentrantLock 自身是线程安全、可重入的；
• Hash 分段锁可以显著减少锁的数量开销；
• 避免死锁：只对一个 key 获取一个锁，不需要多个锁组合；
• 保证有序：锁内处理同一个 key 的数据，天然是串行有序的。

✅ 总结

总结

3个方案各有利弊根据实际业务场景选择最适合的就是最好的方案。针对分段锁实现的进阶方案，由于篇幅问题，下篇文章再继续讨论进阶实现： 巧用弱引用缓存锁对象实现自动回收