java - Spring Webflux 中实体管理器的问题

Question

在我开始使用 Spring WebFlux 的 Web 客户端中，出现以下异常:

reactor.core.Exceptions$ErrorCallbackNotImplemented: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process 'persist' call
Caused by: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process 'persist' call

代码:

@Service
@Transactional
@Slf4j
public class PersonSyncServiceImpl
        extends HotelApiCommunicationService implements PersonSyncService {

    private static final String REST_ENDPOINT_PATH = "/api/sync/person";

    @PersistenceContext
    private EntityManager em;
    @Autowired
    private SynchronizationService syncMgr;


    @Override
    public int initialSyncPersonData(ObiektDTO obiekt) {

        WebClient client = WebClient.create("http://" + obiekt.getAdresIp());
        AtomicInteger count = new AtomicInteger(0);

        Flux<Person> personFlux = client.get()
                .uri(REST_ENDPOINT_PATH)
                .retrieve()
                .bodyToFlux(Person.class);

        personFlux.subscribe(person -> {
            CNPerson cnPerson = syncMgr.convertObject(person, CNPerson.class);
            cnPerson.setObiektLoid(obiekt.getLoid());
            em.persist(cnPerson);
            count.getAndIncrement();
        });

        return count.get();
    }
}

我知道问题出在这条线上，因为 reactor 无法获取实体管理器。

em.persist(cnPerson);

服务器上的 Controller 方法:

    @GetMapping(value = "/person", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<Person> getPersonList() {

        return Flux.fromStream(
                personDao.findAll().stream());
    }

如何解决？如何使用 react 器访问事务并使用 JPA 将记录保存到数据库？

Answer 1

首先，Spring Data R2DBC 不适用于 JPA/Hibernate 和阻塞 JDBC 驱动程序。因此，您需要摆脱 webflux 或 JPA。

使用 JPA

如果假设您决定继续使用 JPA，那么您可以使用 Spring 调用 Person Sync API RestTemplate或者像这样的 WebClient 阻塞调用:

   List<Person> persons = client.get()
                .uri(REST_ENDPOINT_PATH)
                .retrieve()
                .bodyToFlux(Person.class)
                .collectList()
                .block();

像你一样改变它们；

List<CNPerson> cnPersons = new ArrayList();
persons.forEach(person -> {
  CNPerson cnPerson = syncMgr.convertObject(person, CNPerson.class);
  cnPerson.setObiektLoid(obiekt.getLoid());
  cnPersons.add(cnPerson);
});

然后使用您的 org.springframework.data.jpa.repository.JpaRepository.saveAll() 一次性保存它们。 [为此，您需要为 CNPerson 创建一个存储库。实体。]
或手动管理交易并保存每个 CNPerson其中。
最后，回答您的问题是否有其他方法可以在客户端获取流？ - 答案是 否 如果你想在一笔交易中做到这一点。因为您必须在内存中加载 Person Sync API 的整个响应。

没有 JPA(使用 Spring Webflux 和 R2DBC)

您仍然可以使用 Spring Data R2DBC 进行事务处理。您可以阅读更多相关信息 here .请注意，对于 R2DBC，您必须为正在使用的任何类型的数据库使用非阻塞驱动程序。给出列表 here .