基于请求体属性值同步Spring WebFlux POST请求的最优方案咨询

针对你的需求，我们可以通过通用化同步组件解决代码冗余问题，同时结合Spring WebFlux的特性优化性能，确保同一属性值的请求串行处理并优雅失败。

一、通用本地锁同步组件（单机场景） #

将原有的referralIdSync抽象为通用方法，支持任意属性的同步，避免重复代码：

import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

@Component
public class GenericSync {
    
    // 存储锁对象，key为"属性类型:属性值"的组合，避免不同属性的key冲突
    private final ConcurrentHashMap<String, AtomicInteger> lockMap = new ConcurrentHashMap<>();
    
    /**
     * 通用同步方法
     * @param keyType 属性类型标识（如"REFERRAL"、"SLOT"）
     * @param keyValue 属性具体值
     * @param supplier 要执行的业务逻辑
     * @return 业务逻辑执行结果
     */
    public <T> T sync(String keyType, String keyValue, Supplier<T> supplier) {
        String compositeKey = String.format("%s:%s", keyType, keyValue);
        AtomicInteger lock = lockMap.putIfAbsent(compositeKey, new AtomicInteger(0));
        if (lock == null) {
            lock = lockMap.get(compositeKey);
        }
        lock.incrementAndGet();
        
        try {
            // 基于原子对象加锁，保证同一属性值的请求串行执行
            synchronized (lock) {
                return supplier.get();
            }
        } finally {
            // 当计数归0时移除锁对象，避免内存泄漏
            if (lock.decrementAndGet() == 0) {
                lockMap.remove(compositeKey);
            }
        }
    }
}

使用示例 #

不管是referralId还是slotId，都可以复用同一个方法：

// 基于referralId同步处理
DbAppointmentResult referralResult = genericSync.sync("REFERRAL", referralId, 
    () -> daoApi.setappointment(organizationId, referralId, patientId, slotId, previousSlotId));

// 基于slotId同步处理
DbAppointmentResult slotResult = genericSync.sync("SLOT", slotId, 
    () -> daoApi.setappointment(organizationId, referralId, patientId, slotId, previousSlotId));

二、WebFlux场景适配注意事项 #

由于Spring WebFlux是异步非阻塞模型，直接使用synchronized会阻塞事件循环线程，影响应用性能。需要将同步逻辑放到弹性线程池中执行：

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

// 在WebFlux Handler/Controller中调用
Mono<DbAppointmentResult> appointmentResultMono = Mono.fromCallable(() -> 
    genericSync.sync("SLOT", slotId, 
        () -> daoApi.setappointment(organizationId, referralId, patientId, slotId, previousSlotId))
).subscribeOn(Schedulers.boundedElastic());

三、集群场景：分布式锁实现 #

如果服务是集群部署，本地锁无法跨节点生效，可基于Spring Data Redis实现分布式锁：

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

@Component
public class DistributedSync {
    
    private final RedisTemplate<String, String> redisTemplate;
    // 锁超时时间，需根据业务执行时长调整
    private static final long LOCK_TIMEOUT = 30;

    public DistributedSync(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    
    public <T> T sync(String keyType, String keyValue, Supplier<T> supplier) {
        String compositeKey = String.format("%s:%s", keyType, keyValue);
        // 尝试获取锁，设置超时时间避免死锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(compositeKey, "LOCKED", LOCK_TIMEOUT, TimeUnit.SECONDS);
        
        if (locked == null || !locked) {
            // 锁获取失败，直接抛出冲突异常，上层返回优雅响应
            throw new AppointmentConflictException("当前预约时段正在处理，请稍后重试");
        }
        
        try {
            return supplier.get();
        } finally {
            // 释放锁
            redisTemplate.delete(compositeKey);
        }
    }
}

四、优雅失败处理 #

1. 定义自定义业务异常：

public class AppointmentConflictException extends RuntimeException {
    public AppointmentConflictException(String message) {
        super(message);
    }
}

2. 在Controller中捕获异常，返回友好响应：

@RestControllerAdvice
public class GlobalExceptionHandler {
    
    @ExceptionHandler(AppointmentConflictException.class)
    public ResponseEntity<String> handleConflict(AppointmentConflictException e) {
        return ResponseEntity.status(HttpStatus.CONFLICT).body(e.getMessage());
    }
}

五、最后防线：保留数据库约束 #

无论使用哪种锁机制，数据库层面的唯一约束（如针对slotId的唯一索引）必须保留，作为最后一道防线，避免锁机制失效导致的数据不一致。

内容的提问来源于stack exchange，提问作者Sergey Zolotarev