Spring数据不保存实体

Question

我有一个问题，Spring数据没有保存实体。应用程序逻辑如下：

1. 另一个应用程序正在监听负载相当重的Kafka主题（每秒数十条消息）并将消息插入数据库中具有“NEW”状态的表中。
2. @调度方法加载一个具有“NEW”状态的实体列表，这些实体被一个接一个地传输到FixedThreadPool（20个线程），它们的状态设置为“PROCESSING”和一个saveAll方法调用同一个表。

@Scheduled(fixedDelay = 10_000)
public void scheduled() {
    Pageable pageable = PageRequest.of(0, pageSize);
    List<EntityClass> entities = entityService.getAllByStatusOrderByIdDesc(NEW_STATUS, pageable);

    while (!entities.isEmpty()) {
        entities.forEach(entity -> {
            entity.setStatus(PROCESSING_STATUS);
            process(entity);
        });

        entityService.saveAll(entities);
        loggers = entityService.getAllByStatusOrderByIdDesc(NEW_STATUS, pageable);
    }

private void process(Entity entity) {
    threadPool.execute(
         () -> processor.execute(
            TaskBuilder.entity(entity).build()
         )
    );
}

...
log.info("[SaveStep] [execute] status before {}", entity.getStatus());
Entity entity = entityService.save(entity);
log.info("[SaveStep] [execute] status after {}", entity.getStatus());

这两种日志方法都显示正确的状态（“OK”或“ERROR”），但数据库行始终包含“PROCESSING”状态。我试着重做一个实体服务。save（）有多种变体（entityRepository是一种常规的JPARepository）：

1. 常规repo.save（）不起作用：

public Entity save(Entity entity) {
   return repository.save(logger);
}

@Transactional
public Entity save(Entity entity) {
   Entity saved = repository.getFirstById(entity.getId());
   saved.setStatus(entity.getStatus());
   return repository.save(saved);
}

@Transactional
public Entity save(Entity entity) {
   Entity saved = repository.getFirstById(entity.getId());
   log.info("[Service] [save] status before {}", saved.getStatus());
   saved.setStatus(entity.getStatus());
   log.info("[Service] [save] status after {}", saved.getStatus());
   return repository.save(saved);
}

@Modifying
@Transactional
@Query(value = "UPDATE entity_table_name set status = :status where id = :id", nativeQuery = true)
void updateStatus(@Param("id") Long id, @Param("status") String status);

目前我没有办法尝试。主要的侮辱是，大多数实体最终被数据库正确地消化，只有大约20%被save（）忽略，没有抛出任何异常。

Answer 1

谢谢泰勒，他绝对正确地指出了我的问题。即使在调度器点击saveAll（“处理”）行之前，实体列表已经被并行处理，并以足够快的速度以正确的状态保存，这会导致状态重写。

Answer 2

您将消息提交到池后设置为“处理”，这意味着池可能会先访问它，而“处理”更新可能会在之后发生。不确定这是否是您的问题，但您正在做的事情有可能。在提交到池之前尝试设置并保存/提交“处理”更新。