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`锁的优点

高效的锁分段：通过分段锁定，减少了锁竞争，提高了并发处理能力。
自动管理锁对象：无需手动管理锁对象的生命周期，避免内存泄漏。
灵活的锁粒度：可以根据需要调整锁的分段数量，平衡锁的粒度和数量。

实现步骤

引入 Guava 依赖：确保项目中引入了 Guava 库。如果使用 Maven，可以在pom.xml中添加以下依赖：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>31.1-jre</version>
</dependency>

配置Striped锁：使用Striped.lock(int stripes)方法创建一个具有固定数量锁段的Striped锁。例如，创建 64 个锁段：
```
private static final Striped<Lock> stripedLocks = Striped.lock(64);
```

使用Striped锁进行同步：在需要同步的代码块中，通过stripedLocks.get(key)获取对应的锁对象，并进行锁定。

Lock lock = stripedLocks.get(professorId);
lock.lock();
try {
    // 关键业务逻辑
} finally {
    lock.unlock();
}

优化后的代码示例

以下是基于 Guava 的Striped锁机制优化后的RmpAiService类，实现了高效的并发控制：

package com.litongjava.open.chat.services.rmp;

import com.google.common.util.concurrent.Striped;
import com.litongjava.db.activerecord.Db;
import com.litongjava.db.activerecord.Row;
import com.litongjava.db.utils.MarkdownTableUtils;
import com.litongjava.open.chat.constants.TableNames;
import com.litongjava.open.chat.model.RumiRmpProfessorRatingSummary;
import com.litongjava.openai.chat.ChatResponseUsage;
import com.litongjava.openai.chat.ChatResponseVo;
import com.litongjava.openai.client.OpenAiClient;

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(64);

    /**
     * 对指定教授的学生评论数据进行汇总，并保存摘要到数据库。
     *
     * @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();
            ChatResponseVo chat = OpenAiClient.chat(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锁：
```
private static final Striped<Lock> stripedLocks = Striped.lock(64);
```
这里创建了一个包含 64 个锁段的Striped锁。锁的数量可以根据实际需求进行调整，以平衡锁的粒度和系统的并发性能。
获取并锁定特定的锁段：
```
Lock lock = stripedLocks.get(professorId);
lock.lock();
```
通过stripedLocks.get(professorId)方法，根据professorId获取对应的锁对象，并进行锁定。

双重检查机制：在获取锁后，再次检查摘要是否存在，防止在等待锁的过程中，其他线程已经生成了摘要。

if (Db.exists(RumiRmpProfessorRatingSummary.tableName, "id", professorId)) {
    return Db.queryStr(String.format("SELECT summary FROM %s WHERE id=?", RumiRmpProfessorRatingSummary.tableName), professorId);
}

业务逻辑处理：在锁定的情况下，执行摘要生成和保存操作，确保同一时间只有一个线程能够处理特定的professorId。
确保锁的释放：使用try-finally块，确保无论业务逻辑是否成功执行，锁都会被正确释放，避免死锁的发生。
```
finally {
    lock.unlock();
}
```

优势总结

通过采用 Guava 的Striped锁机制，能够有效提升应用的并发性能，具体优势包括：

减少锁竞争：通过分段锁定，降低了锁的粒度，减少了不同professorId之间的锁竞争，提高了系统的吞吐量。
简化锁管理：Striped锁自动管理锁对象的生命周期，无需手动创建和销毁锁对象，降低了开发和维护的复杂性。
灵活配置：可以根据系统的并发需求，灵活调整锁段的数量，平衡锁的粒度和系统的性能。

结论

在高并发的 Java 应用中，选择合适的锁机制对于系统的性能和稳定性至关重要。Guava 的Striped锁通过分段锁定的方式，提供了一种高效且易于管理的并发控制方案。在实际业务中，通过合理配置和使用Striped锁，可以显著提升系统的并发处理能力，确保数据的一致性和线程安全。