Camel：如何使用StreamList从SQL组件流式处理

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时，我得到以下错误：

组织。h2。jdbc。JdbcSQLException:对象已关闭[90007-197]

经检查，我发现连接已关闭。连接={HikariProxyConnection@16287} "HikariProxyConnection@1048081993包装com。扎克瑟。希卡里。水塘代理连接。“闭合连接”

如何使用camel SQL组件来流式传输？我必须从一个不在骆驼路线内的豆子上使用它。我发现，只有在驼峰路由中使用SQL组件时，流媒体才能工作。

骆驼版是：2.24.1

更新1：在看了源代码后，它被要求。onDone关闭连接。我试图在我的默认交换上设置UnitOfWork，通过标记交换未完成来保持连接打开。

Update2：通过设置UnitOf Work来让它工作：

        ProducerTemplate pTmp = camelContext.createProducerTemplate();

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

路由DbStream执行上述SQL选择

Answer 1

不要使用接收体，而只使用接收来获取Exchange。然后您可以从它的消息体中获取迭代器，使用后，您可以完成交换（参见javadoc）