【操作幂等和数据一致性】保障业务在MySQL和COS对象存储的一致

业务场景

发布信息，更新到数据库MySQL
COS操作，更新JSON文件

不过可能存在幂等性和数据一致性的问题。

// 批量存MySQL
entityPublishService.saveOrUpdateBatch(entityPublishList);
// 遍历批量存COS对象存储searchEntitys.forEach(req -> {//删除旧的json文件this.cosFileClient.deleteFile(Constant.ResourcePath.RESOURCE_PATH + req.getId());ObjectMapper mapper = new ObjectMapper();String jsonString = null;try {jsonString = mapper.writeValueAsString(req);// 将 JSON 字符串转换为 InputStream 并上传InputStream inputStream = new ByteArrayInputStream(jsonString.getBytes());this.cosFileClient.uploadFile(Constant.ResourcePath.RESOURCE_PATH + req.getId(), inputStream);} catch (Exception e) {throw new RuntimeException(e);}});

数据一致性与幂等

幂等性指的是无论操作执行多少次，结果都一致。
比如，如果这段代码被重复执行，应该不会导致数据错误或重复上传。
数据一致性则要确保数据库和COS中的JSON文件状态一致，
比如更新数据库后，必须成功上传新的JSON，否则应该回滚或处理失败情况。

业务潜在问题

当前的代码在保存或更新数据库后，遍历entitys，删除旧的JSON，然后生成新的并上传。这里有几个潜在的问题：

数据库操作和文件操作之间没有事务管理，如果上传文件失败，数据库已经提交了事务，导致数据不一致。
删除旧文件后，如果上传新文件失败，会导致数据丢失。
如果方法被重复调用，可能导致多次上传或删除，不幂等。

寻找解决方案

可能的解决方案是，将数据库的保存操作拆分为每个实体的单独操作，并在每个实体处理时，先上传文件，再更新数据库。这样，每个实体的处理是原子性的：上传文件成功，然后更新数据库。如果上传失败，数据库不更新。但这样会影响性能，因为批量操作变为单个操作。
为了数据一致性，可能需要在上传失败时，提供某种回滚机制，比如删除已更新的数据库记录。但这在批量操作中较为困难。
因此可能需要业务层面的处理，比如标记记录为待定状态，或者记录操作日志，用于后续的补偿。
正确的顺序应该是先保存数据库，再上传文件，这样即使上传失败，数据库已经更新，但文件未更新，其他系统可能无法获取最新数据，但至少数据库是正确的，可以触发后续的修复机制。

后续修复或补偿机制，比如在RabbitMQ消费者中业务，配置重试机制。人工操作记录人工补偿机制

为了确保操作的幂等性和数据一致性，尤其是批量操作，代码改进：

移除不必要的旧文件删除操作：直接通过覆盖上传实现更新，避免删除后上传失败导致的数据丢失。
添加上传重试机制：提高上传成功率，减少因临时问题导致的失败。
异常处理和事务分离：确保数据库操作后尽可能完成文件上传，若失败则通过异常通知上层逻辑处理。

// 批量存储MySQL
entityPublishService.saveOrUpdateBatch(entityPublishList);// 遍历批量对象存储COS
entitys.forEach(req -> {ObjectMapper mapper = new ObjectMapper();String jsonString;InputStream inputStream = null;final int maxRetries = 3;int retryCount = 0;boolean uploaded = false;while (retryCount < maxRetries && !uploaded) {try {// 序列化JSON数据jsonString = mapper.writeValueAsString(req);inputStream = new ByteArrayInputStream(jsonString.getBytes(StandardCharsets.UTF_8));// 幂等上传（覆盖模式）this.cosFileClient.uploadFile(Constant.ResourcePath.RESOURCE_PATH + req.getId(), inputStream);uploaded = true; // 标记上传成功} catch (JsonProcessingException e) {// JSON序列化失败，无需重试throw new RuntimeException("JSON序列化失败，ID: " + req.getId(), e);} catch (Exception e) {retryCount++;if (retryCount >= maxRetries) {throw new RuntimeException("文件上传失败，超过最大重试次数，ID: " + req.getId(), e);}// 指数退避等待try {Thread.sleep(1000 * retryCount);} catch (InterruptedException ie) {Thread.currentThread().interrupt();throw new RuntimeException("上传操作被中断，ID: " + req.getId(), ie);}} finally {if (inputStream != null) {try {inputStream.close();} catch (IOException e) {// 日志记录关闭异常，但不影响主流程}}}}if (!uploaded) {// 此处可根据业务需要记录失败状态或触发补偿机制throw new RuntimeException("上传未完成，ID: " + req.getId());}
});

改进说明：

幂等性保障：
- 覆盖上传：直接调用uploadFile覆盖同名文件，多次执行结果一致。
- 数据库操作：saveOrUpdateBatch本身应设计为幂等（如使用唯一约束或主键更新）。
数据一致性措施：
- 先持久化数据库：确保数据库更新后再同步文件，符合业务逻辑依赖。
- 重试机制：通过最多3次重试（含退避等待），减少网络抖动等临时问题的影响。
- 异常处理：上传失败时抛出异常，通知调用方处理（如日志记录、告警、人工干预或事务补偿）。
异常情况处理：
- 序列化错误：直接终止，无需重试（数据问题需修复）。
- 上传失败：重试后仍失败则向上抛出，由系统统一处理（如标记数据状态、异步任务修复）。