Quarkus反应式方法无法向Oracle数据库持久化数据求助

原Quarkus阻塞式流程可正常消费Kafka消息、调用内部API并持久化到Oracle，但每小时仅处理15000条消息，性能不达标。改用反应式架构后，数据库表能正常创建，但数据始终无法持久化，且无任何报错信息。架构流程：KafkaConsumer解析消息密钥后调用对应Service，Service通过RestClient调用API，再借助Mutiny.SessionFactory尝试持久化数据。

相关代码 #

KafkaConsumer代码 #

package com.acme.processor;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.acme.model.Message;
import com.acme.resource.KafkaKeyParser;
import com.acme.service.*;

import jakarta.enterprise.context.ApplicationScoped;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.microprofile.reactive.messaging.Incoming;

import io.smallrye.reactive.messaging.annotations.Blocking;
import io.smallrye.reactive.messaging.kafka.Record;
import jakarta.inject.Inject;
import java.util.Map;


@Slf4j
@ApplicationScoped
public class KafkaMessageConsumer {

    @Inject
    AccountContactEmailService accountEmailService;


    @Incoming("test")
    public Uni<Void> consume(KafkaRecord<String, Message> record) {
        // 密钥解析逻辑（确认正常工作）
        if (httpEndpoint.contains("relevantInfo")) {
            accountContactEmailService.storeId(key1);
        }
        // 其他分支逻辑...
        return Uni.createFrom().voidItem();
    }
}

AccountContactEmailService代码 #

@ApplicationScoped
public class AccountContactEmailService {
    @Inject
    Mutiny.SessionFactory sessionFactory;

    @RestClient
    AccountContactEmailRestClient restClient;

    public Uni<Void> storeId(String key) {
        return restClient.getById(key)
                .onItem().transformToUni(data -> {
                    AccountContactEmail accountContactEmail = data.getAccountContactEmail();
                    return sessionFactory.withTransaction(
                            (session, transaction) -> session.persist(accountContactEmail)
                    );
                 })
                  .onFailure().invoke(e -> {
                      throw new RuntimeException(e);
                 });
        }
}

RestClient代码 #

@Path("/PrettySureMyPathingWorks")
@RegisterRestClient(configKey="rest1")
@ClientBasicAuth(username="user", password= "pass")
public interface AccountContactEmailRestClient {

    @GET
    Uni<AccountContactEmailResponse> getById(@QueryParam("id") String id);
}

问题排查与修复方案 #

1. 核心问题：异步流未被订阅，导致持久化逻辑未执行 #

在KafkaMessageConsumer的consume方法中，调用accountContactEmailService.storeId(key1)后，没有将这个返回的Uni纳入最终返回的流中，而是直接返回了空的Uni.createFrom().voidItem()。反应式操作是懒加载的，不被订阅就不会执行，因此持久化逻辑根本没有触发。

修复代码：

@Incoming("test")
public Uni<Void> consume(KafkaRecord<String, Message> record) {
    // 密钥解析逻辑...
    if (httpEndpoint.contains("relevantInfo")) {
        // 返回Service的Uni，让Reactive Messaging框架自动订阅执行
        return accountEmailService.storeId(key1);
    }
    // 其他分支逻辑...
    return Uni.createFrom().voidItem();
}

2. 异常处理逻辑缺陷，导致异常静默 #

AccountContactEmailService中的onFailure().invoke(e -> throw new RuntimeException(e))会包装异常，但如果上游Uni未被订阅，异常不会传播到消费端。建议添加日志并重新抛出异常，避免异常被吞掉。

优化后的Service代码：

public Uni<Void> storeId(String key) {
    return restClient.getById(key)
            .onItem().transformToUni(data -> {
                AccountContactEmail accountContactEmail = data.getAccountContactEmail();
                return sessionFactory.withTransaction(
                        (session, transaction) -> session.persist(accountContactEmail)
                );
            })
            .onFailure().invoke(e -> log.error("持久化失败: {}", e.getMessage(), e)) // 添加错误日志
            .onFailure().rethrow(); // 重新抛出异常，让消费端感知并处理
}

3. 确认反应式依赖完整性 #

确保项目依赖中包含Quarkus反应式相关组件：

• quarkus-hibernate-reactive
• quarkus-reactive-oracle-client
• quarkus-rest-client-reactive

4. 实体类与事务上下文检查 #

• 实体类继承PanacheEntity后，需确认主键生成策略适配Oracle（例如使用@GeneratedValue(strategy = GenerationType.SEQUENCE)并配置对应数据库序列）
• 确认session.persist(accountContactEmail)操作的实体是新实例，未被其他上下文托管

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