Spring Boot中如何优雅实现分布式锁与事务一致性 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Spring Boot 分布式锁与事务一致性：一种实战方案

大家好，今天我们来聊聊 Spring Boot 中如何优雅地实现分布式锁与事务一致性。这是一个在分布式系统中非常常见且重要的课题，处理不当会导致数据不一致，业务逻辑混乱等严重问题。我们将深入探讨几种常见的解决方案，并重点关注如何使用 Redis 分布式锁配合 Spring 的事务管理，来实现最终一致性。

1. 问题背景：分布式锁与事务的挑战

在单体应用中，我们可以依靠 JVM 提供的锁机制（如 synchronized 关键字或 ReentrantLock）来保证并发安全。同时，数据库事务能保证一系列操作的原子性，要么全部成功，要么全部失败。

但在分布式环境中，JVM 锁只能保证单个 JVM 实例内的并发安全，无法控制跨多个服务实例的并发访问。此时，我们就需要用到分布式锁。而分布式事务，特别是强一致性分布式事务，实现起来非常复杂，性能开销很大。因此，在很多场景下，我们会选择最终一致性方案，即允许短暂的数据不一致，但最终会达到一致状态。

2. 分布式锁的几种实现方案

常用的分布式锁实现方案有以下几种：

方案	优点	缺点	适用场景
基于数据库	实现简单，容易理解	性能较差，存在单点故障风险，锁的释放依赖数据库连接状态，可能导致死锁	并发量较低，对性能要求不高的场景
基于 Redis	性能高，实现相对简单	需要额外的 Redis 依赖，需要考虑 Redis 的可用性问题，锁的自动释放依赖过期时间，可能导致误删锁或锁未释放	并发量较高，对性能有要求的场景
基于 ZooKeeper	可靠性高，提供有序性和watch机制	实现复杂，性能相对 Redis 较差，需要额外的 ZooKeeper 依赖	对可靠性要求极高，需要强一致性的场景，例如配置中心，注册中心

3. 基于 Redis 的分布式锁：Redisson

这里我们选择 Redis 作为分布式锁的实现方案，因为它性能高，使用广泛。同时，我们使用 Redisson 客户端，它提供了丰富的分布式锁 API，并解决了 Redis 分布式锁的一些常见问题，例如：

锁的自动续期（Watchdog）: 防止锁过期被自动释放，导致并发问题。
可重入性: 同一个线程可以多次获取同一个锁。
公平锁: 按照请求顺序获取锁。

3.1 添加 Redisson 依赖

首先，在 pom.xml 文件中添加 Redisson 的依赖：

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.26.0</version> <!-- 请根据最新版本调整 -->
</dependency>

3.2 配置 Redisson

在 application.properties 或 application.yml 文件中配置 Redisson 连接信息：

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password: your_redis_password # 如果有密码

3.3 使用 Redisson 获取分布式锁

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class RedisLockService {

    @Autowired
    private RedissonClient redissonClient;

    public boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit timeUnit) throws InterruptedException {
        RLock lock = redissonClient.getLock(lockKey);
        return lock.tryLock(waitTime, leaseTime, timeUnit);
    }

    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        if (lock.isLocked() && lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }

    public void forceUnlock(String lockKey) {
      RLock lock = redissonClient.getLock(lockKey);
      lock.forceUnlock();
    }
}

这段代码封装了 Redisson 客户端的锁操作，包括：

tryLock(): 尝试获取锁，可以设置等待时间和锁的过期时间。
unlock(): 释放锁，释放前会检查锁是否被当前线程持有。
- forceUnlock(): 强制解锁，不建议使用，除非确定锁状态异常。

4. Spring 事务管理

Spring 提供了强大的事务管理机制，可以通过注解或编程式方式来管理事务。我们这里使用注解方式，在需要事务的方法上添加 @Transactional 注解即可。

5. 分布式锁与事务一致性：最终一致性方案

现在我们将 Redis 分布式锁和 Spring 的事务管理结合起来，实现最终一致性。假设我们有一个场景：扣减库存。我们需要保证在高并发情况下，库存不会出现超卖的情况。

5.1 核心业务逻辑

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;

@Service
public class InventoryService {

    @Autowired
    private RedisLockService redisLockService;

    @Autowired
    private InventoryRepository inventoryRepository; // 假设有一个操作库存的 Repository

    private static final String INVENTORY_LOCK_KEY_PREFIX = "inventory:lock:";

    @Transactional
    public boolean decreaseInventory(Long productId, int quantity) {
        String lockKey = INVENTORY_LOCK_KEY_PREFIX + productId;
        boolean locked = false;
        try {
            locked = redisLockService.tryLock(lockKey, 10, 30, TimeUnit.SECONDS); // 等待10秒，锁过期时间30秒
            if (locked) {
                // 1. 获取库存信息
                Inventory inventory = inventoryRepository.findByProductId(productId);
                if (inventory == null) {
                    throw new IllegalArgumentException("Product not found.");
                }

                // 2. 检查库存是否足够
                if (inventory.getStock() < quantity) {
                    throw new IllegalArgumentException("Insufficient stock.");
                }

                // 3. 扣减库存
                inventory.setStock(inventory.getStock() - quantity);
                inventoryRepository.save(inventory);

                // 模拟异常，测试事务回滚
                // if (true) {
                //     throw new RuntimeException("Simulated error for rollback.");
                // }

                return true; // 扣减成功
            } else {
                // 获取锁失败，可以重试或返回错误信息
                System.out.println("Failed to acquire lock for product: " + productId);
                return false;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断状态
            System.err.println("Interrupted while acquiring lock for product: " + productId);
            return false;
        } finally {
            if (locked) {
                redisLockService.unlock(lockKey);
            }
        }
    }
}

代码解释:

@Transactional: 这个注解表示 decreaseInventory 方法是一个事务方法。 Spring 会自动管理事务的开始、提交和回滚。
redisLockService.tryLock(): 尝试获取分布式锁，如果获取成功，则执行后续的业务逻辑。等待时间设置为 10 秒，锁的过期时间设置为 30 秒。
inventoryRepository.findByProductId(): 从数据库中获取库存信息。
库存检查: 检查库存是否足够，如果不足，则抛出异常。
inventoryRepository.save(): 扣减库存，并更新数据库。
redisLockService.unlock(): 释放分布式锁。无论操作成功或失败，都要释放锁，确保不会出现死锁。
异常处理: 在 catch 块中，捕获 InterruptedException 异常，并恢复中断状态。在 finally 块中，确保锁被释放。

5.2 关键点分析

锁的粒度: 锁的粒度应该尽可能小，这里我们使用 productId 作为锁的 key，可以避免锁住整个库存表。
锁的过期时间: 锁的过期时间应该设置合理，既要避免锁过早释放导致并发问题，又要避免锁过期时间过长导致死锁。
事务的回滚: 如果在事务执行过程中发生异常，Spring 会自动回滚事务，保证数据的一致性。同时，由于我们使用了 Redis 分布式锁，即使事务回滚，锁也会被释放，不会影响其他线程的执行。
重试机制: 如果获取锁失败，可以考虑重试，或者返回错误信息给用户。重试机制可以提高系统的可用性，但需要注意避免死循环。

6. 解决潜在问题：锁的误删

如果在 decreaseInventory 方法中，执行时间超过了锁的过期时间，那么锁会被 Redis 自动释放。此时，如果另一个线程获取了锁，那么两个线程可能会同时执行扣减库存的操作，导致数据不一致。

Redisson 客户端通过 Watchdog 机制来解决这个问题。 Watchdog 机制会在锁过期之前，自动延长锁的过期时间，确保锁不会被自动释放。默认情况下，Watchdog 机制会每隔 lockWatchdogTimeout / 3 毫秒检查锁是否快要过期，如果快要过期，则自动延长锁的过期时间。 lockWatchdogTimeout 可以在 Redisson 的配置中进行设置，默认为 30 秒。

7. 幂等性

在分布式系统中，由于网络抖动等原因，可能会出现消息重复发送的情况。为了保证数据的一致性，我们需要保证接口的幂等性。幂等性是指，无论接口被调用多少次，其结果都是一样的。

常用的幂等性解决方案有以下几种：

唯一 ID: 为每个请求生成一个唯一的 ID，并将 ID 作为请求参数传递给接口。接口在处理请求之前，先检查 ID 是否已经存在，如果存在，则直接返回结果，否则执行业务逻辑。
Token 机制: 客户端先向服务器申请一个 Token，服务器将 Token 存储在 Redis 中。客户端在发起请求时，将 Token 传递给服务器。服务器在处理请求之前，先检查 Token 是否存在，如果存在，则执行业务逻辑，并删除 Token，否则拒绝请求。
乐观锁: 在数据库表中添加一个版本号字段，每次更新数据时，版本号加 1。在更新数据时，先比较版本号是否与数据库中的版本号一致，如果一致，则更新数据，否则拒绝更新。

8. 补偿机制

虽然我们使用了 Redis 分布式锁和 Spring 的事务管理，但是仍然可能出现数据不一致的情况。例如，由于网络故障，事务提交失败，导致数据回滚，但是 Redis 锁已经被释放。

为了解决这个问题，我们可以使用补偿机制。补偿机制是指，在事务执行失败后，通过一定的手段来恢复数据的一致性。

常用的补偿机制有以下几种：

重试: 如果事务执行失败，可以尝试重试。重试机制可以解决一些临时的网络故障。
人工干预: 如果重试多次仍然失败，可以考虑人工干预。人工干预可以解决一些复杂的异常情况。
消息队列: 将需要执行的操作放入消息队列中，由消费者来执行。如果消费者执行失败，可以重新放入消息队列中，等待下次执行。

9. 代码示例：添加幂等性控制

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Service
public class InventoryService {

    @Autowired
    private RedisLockService redisLockService;

    @Autowired
    private InventoryRepository inventoryRepository;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private static final String INVENTORY_LOCK_KEY_PREFIX = "inventory:lock:";
    private static final String DECREASE_INVENTORY_KEY_PREFIX = "decrease:inventory:";

    @Transactional
    public boolean decreaseInventory(Long productId, int quantity, String requestId) {
        String lockKey = INVENTORY_LOCK_KEY_PREFIX + productId;
        boolean locked = false;
        try {
            locked = redisLockService.tryLock(lockKey, 10, 30, TimeUnit.SECONDS);
            if (locked) {
                // 1. 幂等性校验
                String idempotentKey = DECREASE_INVENTORY_KEY_PREFIX + requestId;
                Boolean exists = stringRedisTemplate.hasKey(idempotentKey);
                if (exists != null && exists) {
                    System.out.println("Request already processed: " + requestId);
                    return true; // 已经处理过，直接返回成功
                }

                // 2. 获取库存信息
                Inventory inventory = inventoryRepository.findByProductId(productId);
                if (inventory == null) {
                    throw new IllegalArgumentException("Product not found.");
                }

                // 3. 检查库存是否足够
                if (inventory.getStock() < quantity) {
                    throw new IllegalArgumentException("Insufficient stock.");
                }

                // 4. 扣减库存
                inventory.setStock(inventory.getStock() - quantity);
                inventoryRepository.save(inventory);

                // 5. 记录请求已处理
                stringRedisTemplate.opsForValue().set(idempotentKey, "done", 1, TimeUnit.HOURS); // 设置过期时间

                return true;
            } else {
                System.out.println("Failed to acquire lock for product: " + productId);
                return false;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Interrupted while acquiring lock for product: " + productId);
            return false;
        } finally {
            if (locked) {
                redisLockService.unlock(lockKey);
            }
        }
    }

    public String generateRequestId() {
        return UUID.randomUUID().toString();
    }
}

在这个示例中，我们使用 Redis 来实现幂等性控制。每个请求都需要传递一个 requestId，在处理请求之前，先检查 Redis 中是否存在该 requestId，如果存在，则说明该请求已经被处理过，直接返回成功。否则，执行业务逻辑，并将 requestId 存储在 Redis 中，设置过期时间。

10. 其他注意事项

监控: 需要对分布式锁的获取和释放进行监控，以便及时发现问题。
日志: 需要记录详细的日志，以便排查问题。
测试: 需要进行充分的测试，包括并发测试、异常测试等，以确保系统的稳定性和可靠性。
避免长事务：尽量将事务拆分成更小的单元，减少锁的持有时间。
考虑使用消息队列：如果对一致性要求不高，可以考虑使用消息队列来实现最终一致性，将耗时操作异步化。

关于分布式锁和事务一致性的讨论结束

我们讨论了Spring Boot 中实现分布式锁与事务一致性的一个方案，包括使用 Redisson 作为分布式锁，Spring 的事务管理以及如何保证最终一致性。希望对大家有所帮助。记住，没有银弹，要根据实际场景选择最合适的方案。谢谢大家！

Spring Boot 分布式锁与事务一致性：一种实战方案

发表回复 取消回复

发表回复取消回复