使用MySQL在分布式Spring Boot应用中实现幂等性

CREATE TABLE idempotency_key (
       idem_key      VARCHAR(128) NOTNULL,
       status        ENUM('IN_PROGRESS','COMPLETED','FAILED') NOTNULL,
       request_hash  CHAR(64) NULL,
       response_json JSONNULL,
       created_at    TIMESTAMPNOTNULLDEFAULTCURRENT_TIMESTAMP,
       updated_at    TIMESTAMPNOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,
       PRIMARY KEY (idem_key)
) ENGINE=InnoDB;

spring.application.name=IdempotencyCheck

spring.datasource.url=jdbc:mysql://localhost:3306/product?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
spring.datasource.username=root
spring.datasource.password=password
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

spring.jpa.hibernate.ddl-auto=update
spring.jpa.show-sql=true
spring.jpa.properties.hibernate.format_sql=true
spring.jpa.open-in-view=false

spring.datasource.hikari.auto-commit=false
spring.datasource.hikari.maximum-pool-size=10
spring.datasource.hikari.minimum-idle=2
spring.datasource.hikari.idle-timeout=30000
spring.datasource.hikari.connection-timeout=20000
spring.datasource.hikari.max-lifetime=1800000

package repository;

import jakarta.persistence.*;
import lombok.Getter;
import lombok.Setter;

@Entity
@Table(name = "idempotency_key")
@Getter@Setter
publicclass IdempotencyKeyEntity {

    @Id
    @Column(name="idem_key", length = 128)
    private String key;

    @Enumerated(EnumType.STRING)
    private Status status;

    @Column(name = "request_hash", length = 64)
    private String requestHash;

    @Column(name = "response_json", columnDefinition = "json")
    private String responseJson;

    publicenum Status { IN_PROGRESS, COMPLETED, FAILED }

}

import jakarta.persistence.LockModeType;
import org.springframework.data.jpa.repository.Lock;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import org.springframework.stereotype.Repository;

import java.util.Optional;

@Repository
publicinterface IdempotencyRepositoryImpl {

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("select idem from IdempotencyKeyEntity idem where idem.key = :key")
    Optional<IdempotencyKeyEntity> lockByKey(@Param("key") String key);

}

@Transactional
public Optional<String> getCompletedResponse(String key, String requestHash){

    //Find the key or else create a record and insert it into the database and return the entity
    repo.findById(key).orElseGet(() -> {

        IdempotencyKeyEntity entity = new IdempotencyKeyEntity();
        entity.setKey(key);
        entity.setStatus(IdempotencyKeyEntity.Status.IN_PROGRESS);
        entity.setRequestHash(requestHash);
        try{
            return repo.saveAndFlush(entity);
        }catch (DataIntegrityViolationException exception){
            returnnull;
        }
    });

    //Lock the row only one thread at a time
    IdempotencyKeyEntity locked = repo.lockByKey(key)
            .orElseThrow(() -> new IllegalStateException("Row must exist"));

    //If already completed return the cached response
    if (locked.getStatus() == IdempotencyKeyEntity.Status.COMPLETED) {
        if (!Objects.equals(locked.getRequestHash(), requestHash)) {
            thrownew ResponseStatusException(HttpStatus.CONFLICT,
                    "Idempotency-Key reuse with different request");
        }
        return Optional.ofNullable(locked.getResponseJson());
    }

    // If in progress and hash differs conflict (Not blocking here)
    if (locked.getRequestHash() != null &&
            !Objects.equals(locked.getRequestHash(), requestHash)) {
        thrownew ResponseStatusException(HttpStatus.CONFLICT,
                "Idempotency-Key reuse with different request");
    }

    //Not yet completed, the caller should do the work and mark it completed
    return Optional.empty();
}

@Transactional
public void completed(String key, String responseJson) {
    IdempotencyKeyEntity locked = repo.lockByKey(key)
            .orElseThrow(() -> new IllegalStateException("Row must exist"));
    locked.setStatus(IdempotencyKeyEntity.Status.COMPLETED);
    locked.setResponseJson(responseJson);
    repo.save(locked);
}

@Transactional
public void failed(String key) {
    IdempotencyKeyEntity locked = repo.lockByKey(key)
            .orElseThrow(() -> new IllegalStateException("Row must exist"));
    locked.setStatus(IdempotencyKeyEntity.Status.FAILED);
    repo.save(locked);
}

@RestController
@RequiredArgsConstructor
publicclass IdempotentController {

    privatefinal IdempotencyService idempotencyService;

    @PostMapping("/payments")
    public ResponseEntity<String> createPayments(@RequestHeader("Idempotency-Key") String idemKey,
                                                 @RequestBody PaymentRequest req){

        //Using Google Guava for 256 hashing
        String hashReq = Hashing.sha256()
                .hashString(req.toString(), StandardCharsets.UTF_8)
                .toString();

        //check for cachedInDB
        Optional<String> cachedInDB = idempotencyService.getCompletedResponse(idemKey, hashReq);
        if(cachedInDB.isPresent()){
            return ResponseEntity.ok(cachedInDB.get());
        }

        //Do the business logic
        try{
            String results = idempotencyService.doWork();
            //Mark the state as completed for the idempotent key
            idempotencyService.completed(idemKey, results);

            return ResponseEntity.ok(results);
        }catch (Exception ex){
            //if the transaction fails, mark the idempotent key as failed to be processed later by other threads
            idempotencyService.failed(idemKey);
            throw ex;
        }
    }
}