java-db 整合 Guava 的 Striped 锁优化
在现代高并发应用中，确保数据的一致性和线程安全性是至关重要的。本文将详细介绍如何使用 Guava 库提供的Striped锁机制来优化 Java 应用中的并发控制。通过一个具体的业务案例，展示Striped锁在实际应用中的优势和实现方法。
业务背景
假设我们有一个系统，用于处理教授的学生评论数据，并生成对应的汇总摘要。具体的业务逻辑包括：
- 汇总评论：对指定教授的学生评论进行汇总，生成汇总后的文本。
- 保存摘要：将生成的摘要保存到
RumiRmpProfessorRatingSummary表中，以实现缓存，提升并发处理能力。
在高并发环境下，多个线程可能会同时请求对同一个professorId进行摘要生成和保存操作。为了避免重复生成摘要和数据不一致的问题，需要有效的线程同步机制。
面临的挑战
在多线程环境下，针对同一个professorId进行摘要生成和保存操作时，可能会出现以下问题：
- 重复生成摘要：多个线程同时检测到摘要不存在，进而同时生成并保存摘要，导致重复数据。
- 数据不一致：并发写操作可能导致数据库中的数据不一致或覆盖。
为了解决这些问题，需要在应用层实现有效的锁机制，确保同一时间只有一个线程能够处理特定的professorId。
传统的同步方法及其局限性
传统的同步方法如sychronized关键字，可以确保同一时间只有一个线程访问被锁定的代码块。然而，在高并发场景下，synchronized可能会成为性能瓶颈，尤其是在锁粒度较大或锁竞争激烈的情况下。
示例代码（使用sychronized）
public class RmpAiService {
private static final Map<Long, Object> locks = new ConcurrentHashMap<>();
public String summary(Long professorId) {
if (Db.exists(RumiRmpProfessorRatingSummary.tableName, "id", professorId)) {
return Db.queryStr(String.format("select summary from %s where id=?", RumiRmpProfessorRatingSummary.tableName), professorId);
}
Object lock = locks.computeIfAbsent(professorId, id -> new Object());
synchronized (lock) {
try {
if (Db.exists(RumiRmpProfessorRatingSummary.tableName, "id", professorId)) {
return Db.queryStr(String.format("select summary from %s where id=?", RumiRmpProfessorRatingSummary.tableName), professorId);
}
// 生成摘要并保存
String summary = generateAndSaveSummary(professorId);
return summary;
} finally {
locks.remove(professorId);
}
}
}
private String generateAndSaveSummary(Long professorId) {
// 摘要生成逻辑
return "摘要内容";
}
}局限性：
- 性能开销：
synchronized在高并发场景下可能导致性能下降。 - 锁管理复杂性：需要手动管理锁对象的生命周期，防止内存泄漏。
使用 Guava 的Striped锁机制
为了解决上述问题，可以采用 Guava 库提供的Striped锁机制。Striped锁通过分段锁定的方式，减少锁的粒度和锁竞争，提高并发性能。
什么是Striped锁？
Striped锁是一种基于哈希分段的锁机制。它将多个锁分散到不同的锁段中，通过哈希算法将特定的键（如professorId）映射到相应的锁段。这种方式有效地减少了锁的数量，同时避免了全局锁带来的性能瓶颈。
Striped锁的优点
- 高效的锁分段：通过分段锁定，减少了锁竞争，提高了并发处理能力。
- 自动管理锁对象：无需手动管理锁对象的生命周期，避免内存泄漏。
- 灵活的锁粒度：可以根据需要调整锁的分段数量，平衡锁的粒度和数量。
Striped锁的数量
在 Guava 的并发工具中，Striped 代表一种“分条（分段）”的锁或信号量实现，其内部维护了多个锁（或信号量），通过被加锁对象的哈希值来决定具体使用其中哪一个锁，从而减少锁竞争。
如果只针对单词 “striped” 做中文翻译，常见的表述方式有 “条带式”、“分段式” 或 “分片式”。在此上下文中，一般会翻译成 “分段（锁）” 或 “条带（锁）”，表示它是把锁“分割”成多个小单元（即多条带、分段）来使用。
所以可以将
private static final Striped<Lock> stripedLocks = Striped.lock(1024);
理解为**“创建了 64 段（条带）的锁”，而 striped 就可以直译为 “分段的”、“分片的”** 或 “条带式的”。
当调用 Striped.lock(1024) 时，Guava 会在内部创建 64 个独立的 Lock 实例。每当使用 stripedLocks.get(key) 获取锁时，它会根据 key 的哈希值选择其中一个锁来加锁，从而减少竞争并提高并发性能。
实现步骤
引入 Guava 依赖： 确保项目中引入了 Guava 库。如果使用 Maven，可以在
pom.xml中添加以下依赖：xml<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>31.1-jre</version> </dependency>配置
Striped锁： 使用Striped.lock(int stripes)方法创建一个具有固定数量锁段的Striped锁。例如，创建 64 个锁段：javaprivate static final Striped<Lock> stripedLocks = Striped.lock(1024);使用
Striped锁进行同步： 在需要同步的代码块中，通过stripedLocks.get(key)获取对应的锁对象，并进行锁定。javaLock lock = stripedLocks.get(professorId); lock.lock(); try { // 关键业务逻辑 } finally { lock.unlock(); }
优化后的代码示例
以下是基于 Guava 的Striped锁机制优化后的RmpAiService类，实现了高效的并发控制：
package nexus.io.open.chat.services.rmp;
import com.google.common.util.concurrent.Striped;
import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.db.utils.MarkdownTableUtils;
import nexus.io.open.chat.constants.TableNames;
import nexus.io.open.chat.model.RumiRmpProfessorRatingSummary;
import nexus.io.openai.chat.ChatResponseUsage;
import nexus.io.openai.chat.OpenAiChatResponse;
import nexus.io.openai.client.OpenAiClient;
import nexus.io.openai.consts.OpenAiModels;
import java.util.List;
import java.util.concurrent.locks.Lock;
public class RmpAiService {
private String summary_prompt_en = "You are a teaching assistant, and I will provide you with student feedback data for a professor. Please summarize this data and output the summarized content.";
private String summary_prompt = summary_prompt_en;
// 使用Guava的Striped锁，设置64个锁段
private static final Striped<Lock> stripedLocks = Striped.lock(1024);
/**
* 对指定教授的学生评论数据进行汇总，并保存摘要到数据库。
*
* @param professorId 教授的唯一标识ID
* @return 汇总后的摘要内容
*/
public String summary(Long professorId) {
// 第一次检查摘要是否存在
if (Db.exists(RumiRmpProfessorRatingSummary.tableName, "id", professorId)) {
return Db.queryStr(String.format("SELECT summary FROM %s WHERE id=?", RumiRmpProfessorRatingSummary.tableName), professorId);
}
// 获取对应的锁并锁定
Lock lock = stripedLocks.get(professorId);
lock.lock();
try {
// 再次检查摘要是否存在，防止其他线程已生成
if (Db.exists(RumiRmpProfessorRatingSummary.tableName, "id", professorId)) {
return Db.queryStr(String.format("SELECT summary FROM %s WHERE id=?", RumiRmpProfessorRatingSummary.tableName), professorId);
}
// 进行数据库查询和数据处理
List<Row> rows = Db.findByField(TableNames.rumi_rmp_rating, "teacher_id", professorId);
for (Row row : rows) {
row.remove("teacher_id")
.remove("school_id")
.remove("created_by_user")
.remove("rating_tags")
.remove("source_url")
.remove("remark")
.remove("creator")
.remove("create_time")
.remove("updater")
.remove("update_time")
.remove("deleted")
.remove("tenant_id")
.remove("class_name_vector");
}
String markdownTable = MarkdownTableUtils.to(rows);
String prompt = summary_prompt + "\r\n" + markdownTable;
long start = System.currentTimeMillis();
OpenAiChatResponse chat = OpenAiClient.chatWithModel(OpenAiModels.GPT_4O_MINI, "user", prompt);
long end = System.currentTimeMillis();
long elapsed = end - start;
ChatResponseUsage usage = chat.getUsage();
Integer prompt_tokens = usage.getPrompt_tokens();
Integer completion_tokens = usage.getCompletion_tokens();
Integer total_tokens = usage.getTotal_tokens();
String system_fingerprint = chat.getSystem_fingerprint();
String model = chat.getModel();
String content = chat.getChoices().get(0).getMessage().getContent();
// 创建并保存摘要实体
RumiRmpProfessorRatingSummary entity = new RumiRmpProfessorRatingSummary();
entity.setId(professorId)
.setSummary(content)
.setCompletionTokens(completion_tokens)
.setPromptTokens(prompt_tokens)
.setTotalTokens(total_tokens)
.setModel(model)
.setElapsed(elapsed)
.setSystemFingerprint(system_fingerprint);
entity.save();
return content;
} finally {
// 确保锁被释放
lock.unlock();
}
}
}关键点解析
创建
Striped锁：javaprivate static final Striped<Lock> stripedLocks = Striped.lock(1024);这里创建了一个包含 64 个锁段的
Striped锁。锁的数量可以根据实际需求进行调整，以平衡锁的粒度和系统的并发性能。获取并锁定特定的锁段：
javaLock lock = stripedLocks.get(professorId); lock.lock();通过
stripedLocks.get(professorId)方法，根据professorId获取对应的锁对象，并进行锁定。双重检查机制： 在获取锁后，再次检查摘要是否存在，防止在等待锁的过程中，其他线程已经生成了摘要。
javaif (Db.exists(RumiRmpProfessorRatingSummary.tableName, "id", professorId)) { return Db.queryStr(String.format("SELECT summary FROM %s WHERE id=?", RumiRmpProfessorRatingSummary.tableName), professorId); }业务逻辑处理： 在锁定的情况下，执行摘要生成和保存操作，确保同一时间只有一个线程能够处理特定的
professorId。确保锁的释放： 使用
try-finally块，确保无论业务逻辑是否成功执行，锁都会被正确释放，避免死锁的发生。javafinally { lock.unlock(); }
优势总结
通过采用 Guava 的Striped锁机制，能够有效提升应用的并发性能，具体优势包括：
- 减少锁竞争：通过分段锁定，降低了锁的粒度，减少了不同
professorId之间的锁竞争，提高了系统的吞吐量。 - 简化锁管理：
Striped锁自动管理锁对象的生命周期，无需手动创建和销毁锁对象，降低了开发和维护的复杂性。 - 灵活配置：可以根据系统的并发需求，灵活调整锁段的数量，平衡锁的粒度和系统的性能。
结论
在高并发的 Java 应用中，选择合适的锁机制对于系统的性能和稳定性至关重要。Guava 的Striped锁通过分段锁定的方式，提供了一种高效且易于管理的并发控制方案。在实际业务中，通过合理配置和使用Striped锁，可以显著提升系统的并发处理能力，确保数据的一致性和线程安全。