java - Camel : How to stream a from SQL component using StreamList

Question

我正在尝试使用Camels SQL组件通过outputType=StreamList从数据库进行流式传输。我使用 ConsumerTemplate 从 Java 类获取 ResultIterator:

public Flux<String> CreateFlux () {
ConsumerTemplate consumer = camelContext.createConsumerTemplate();

    ResultSetIterator resultSetIterator = consumer.receiveBody(
            "sql:SELECT DATA FROM TRANSAKSJON WHERE REQ_ID='" + recId + "'?outputType=StreamList", ResultSetIterator.class);
    ...
    while (result.hasNext()) {
        Map<String, String> map = (Map<String, String>) result.next();
        String data = map.get("DATA");

    }
}

尝试迭代 ResultsetIterator 时出现以下错误:

org.h2.jdbc.JdbcSQLException:对象已关闭 [90007-197]

经检查，我发现连接已关闭。连接 = {HikariProxyConnection@16287} "HikariProxyConnection@1048081993 包装com.zaxxer.hikari.pool.ProxyConnection.ClosedConnection"

如何使用camel SQL组件进行流式传输？我必须从不在 Camel 路线内的 bean 中使用它。我发现只有当我在 Camel 路由中使用 SQL 组件时，流式传输才起作用。

Camel 版本是:2.24.1

Update1:​​查看源代码后，这是有意的。 onDone 关闭连接。我正在尝试在 defaultExchange 上设置 UnitOfWork，通过将交换标记为未完成来保持连接打开。

更新2:设法通过设置工作单元使其工作:

        ProducerTemplate pTmp = camelContext.createProducerTemplate();

        DefaultExchange defaultExchange = new DefaultExchange(camelContext);
        UnitOfWork unitOfWork = new DefaultUnitOfWork(defaultExchange);
        defaultExchange.setUnitOfWork(unitOfWork);
        pTmp.send("direct:DbStream", defaultExchange);

路由DbStream执行上述SQL select

Answer 1

不要使用receiveBody，而只是使用receive来取回Exchange。然后就可以从消息体中获取迭代器，使用后就可以进行交换了(参见javadoc)