Java - 将 JMS 消息流式传输到 Oracle 表中

Posted 2023-03-27

【中文标题】Java - 将 JMS 消息流式传输到 Oracle 表中

【英文标题】：Java - Streaming JMS messages into an Oracle table

【发布时间】：2017-04-03 14:03:39

【问题描述】：

我们正在使用 Spring Boot 和 JDBC 来捕获 JMS 消息并将它们保存到数据库表中。

为了尽可能减少内存占用，我们希望将这些消息流式传输到数据库中并批量提交。

是否可以将这些消息流式传输到数据库中？

目前，我们有多个 JMS 侦听器正在使用我们的 JMS 队列。这些侦听器共享一个 LinkedBlockingQueue（它由服务包装），他们将消息写入该队列。然后使用此队列，直到它耗尽，结果存储在列表中。这些结果使用 Spring 的 JdbcTemplate 进行持久化。

我们希望降低内存占用，因为这些消息存储在中间列表中只是为了允许批量保存。

在这种情况下有什么建议或经过验证的模式可以遵循吗？

【问题讨论】：

什么是事务边界？如果消息的消费是一个事务，那么结果似乎应该作为该事务的一部分被持久化。如果 JVM、网络等出现故障，如何恢复队列中的结果？

关于事务边界的要点。我认为更好的选择可能是一次使用一个消息列表，例如 2000 条消息，然后简单地保留它们而不添加任何花哨的批处理机制。事务边界位于侦听器级别。这样，我们仍然可以与已消费/未消费的内容保持一致。我看到只是将它们推送到队列会丢失交易信息。

【参考方案1】：

我设法使用方便的 StreamEx 库对流进行了分块。 但是，我们仍然在流实现时遇到 JVM 内存问题。我们需要运行一些进一步的性能和垃圾回收测试。

要流式传输的代码如下：

try (StreamEx<MyEntity> stream = StreamEx.of(myEntities)) 
            stream.groupRuns((prev, next) -> recordCounter.incrementAndGet() % myApplicationProperties.getBatchSize() != 0)
                    .forEach((chunk) -> 
                        if (chunk.size() != 1) 
                            jmsTemplate.convertAndSend(chunk);
                         else 
                            jmsTemplate.convertAndSend(chunk.get(0));
                        
                    );