在Java Stream实现大型查询

栏目: Java · 发布时间: 6年前

内容简介:Spring JdbcTemplate从1.0版开始就一直在使用这个类,并且它的发展很好,但我希望在版本5中它会包含一些流式处理功能,用于查询很大数据结果,可惜没有发生。尽管如此,有时我需要执行返回数百万行的查询,而且我不能使用JdbcTemplate方法来返回列表,RowCallbackHandler非常适合,但是如果收到Stream会好得多,不是吗?特别是如果你有自定义RowMappers ...所以,我决定编写自己的Stream生成器以与JdbcTemplate一起使用。在这个过程中,我最终创建了一

Spring JdbcTemplate从1.0版开始就一直在使用这个类,并且它的发展很好,但我希望在版本5中它会包含一些流式处理功能,用于查询很大数据结果,可惜没有发生。

尽管如此,有时我需要执行返回数百万行的查询,而且我不能使用JdbcTemplate方法来返回列表,RowCallbackHandler非常适合,但是如果收到Stream会好得多,不是吗?特别是如果你有自定义RowMappers ...

所以,我决定编写自己的Stream生成器以与JdbcTemplate一起使用。在这个过程中,我最终创建了一个更通用的Stream生成器,我觉得这很好,所以我想与需要类似东西的人分享。

挑战

首先,我们需要考虑流是惰性的,当你得到一个流并定义要在其上完成的操作时,其实没有任何事情发生,直到你实现最终操作,它需要实际遍历元素并应用对它的操作。有些操作遍历整个流(例如计数或将元素收集到另一个集合中),并且存在短路操作(例如确定是否有任何元素通过某些过滤器)。

因此,我们希望得到一个流,并在其上定义操作,并且没有任何反应,直到流需要遍历,然后需要运行查询(这意味着与数据库建立开放连接)。如果发生错误,查询需要停止(并且JdbcTemplate将负责清理连接和其他资源)。

我发现我能够完成这项工作的唯一方法是使用两个线程:一个生成器线程,其中运行查询,行以某种方式提供给流,以及一个消费者线程,它是流的读者。

我们需要一个缓冲区,生产者将在其中存储元素,消费者将从中获取元素。LinkedBlockingQueue似乎是完美的。

<b>public</b> <b>static</b> <T> Stream<T> streamForQuery(<b>int</b> bufferSize, T endOfStreamMarker,
                                               Consumer<Consumer<T>> query) {
        <b>final</b> LinkedBlockingQueue<T> queue = <b>new</b> LinkedBlockingQueue<>(bufferSize);
        <font><i>//This is the consumer that is usually passed to queries;</i></font><font>
        </font><font><i>//it will receive each item from the query and put it in the queue</i></font><font>
        Consumer<T> filler = t -> {
            <b>try</b> {
                </font><font><i>//Try to add to the queue, waiting up to 1 second</i></font><font>
                </font><font><i>//Honestly if after 1 second the queue is still full, either the stream consumer</i></font><font>
                </font><font><i>//needs some serious optimization or, more likely, a short-circuit terminal</i></font><font>
                </font><font><i>//operation was performed on the stream.</i></font><font>
                <b>if</b> (!queue.offer(t, 1, TimeUnit.SECONDS)) {
                    </font><font><i>//If the queue is full after 1 second, time out.</i></font><font>
                    </font><font><i>//Throw an exception to stop the producer queue.</i></font><font>
                    log.error(</font><font>"Timeoud waiting to feed elements to stream"</font><font>);
                    <b>throw</b> <b>new</b> BufferOverflowException();
                }
            } <b>catch</b> (InterruptedException ex) {
                System.err.println(</font><font>"Interrupted trying to add item to stream"</font><font>);
                ex.printStackTrace();
            }
        };
        </font><font><i>//For the stream that we return, we use a Spliterator.</i></font><font>
        <b>return</b> StreamSupport.stream(() -> <b>new</b> Spliterators.AbstractSpliterator<T>(Long.MAX_VALUE, Spliterator.ORDERED) {
            </font><font><i>//We need to know if the producer thread has been started</i></font><font>
            <b>private</b> <b>boolean</b> started = false;
            </font><font><i>//If there's an exception in the producer, keep it here</i></font><font>
            <b>private</b> <b>volatile</b> Throwable boom;
            </font><font><i>/** This method is called once, before advancing to the first element.
             * It will start the producer thread, which runs the query, passing it our
             * queue filler.
             */</i></font><font>
            <b>private</b> <b>void</b> startProducer() {
                </font><font><i>//Get the consumer thread</i></font><font>
                Thread interruptMe = Thread.currentThread();
                </font><font><i>//First time this is called it will run the query in a separate thread</i></font><font>
                </font><font><i>//This is the producer thread</i></font><font>
                <b>new</b> Thread(() -> {
                    <b>try</b> {
                        </font><font><i>//Run the query, with our special consumer</i></font><font>
                        query.accept(filler);
                    } <b>catch</b> (BufferOverflowException ignore) {
                        </font><font><i>//The filler threw this, means the queue is not being consumed fast enough</i></font><font>
                        </font><font><i>//(or, more likely, not at all)</i></font><font>
                    } <b>catch</b> (Throwable thr) {
                        </font><font><i>//Something bad happened, store the exception and interrupt the reader</i></font><font>
                        boom = thr;
                        interruptMe.interrupt();
                    }
                }).start();
                started = <b>true</b>;
            }
            @Override
            <b>public</b> <b>boolean</b> tryAdvance(Consumer<? <b>super</b> T> action) {
                <b>if</b> (!started) {
                    startProducer();
                }
                <b>try</b> {
                    </font><font><i>//Take an item from the queue and if it's not the end of stream maker, pass it</i></font><font>
                    </font><font><i>//to the action consumer.</i></font><font>
                    T t = queue.take();
                    <b>if</b> (t != endOfStreamMarker) {
                        action.accept(t);
                        <b>return</b> <b>true</b>;
                    }
                } <b>catch</b> (InterruptedException ex) {
                    <b>if</b> (boom == <b>null</b>) {
                        System.err.println(</font><font>"Interrupted reading from stream"</font><font>);
                        ex.printStackTrace();
                    } <b>else</b> {
                        </font><font><i>//Throw the exception from the producer on the consumer side</i></font><font>
                        <b>throw</b> <b>new</b> RuntimeException(boom);
                    }
                }
                <b>return</b> false;
            }
        }, Spliterator.IMMUTABLE, false);
    }
</font>

这就是你使用JdbcTemplate的方式:

<b>final</b> MyRow marker = <b>new</b> MyRow();
Stream<MyRow> stream = streamForQuery(100, marker, callback -> {
    <font><i>//Pass a RowCallbackHandler that passes a MyRow to the callback</i></font><font>
    jdbcTemplate.query(</font><font>"SELECT * FROM really_big_table_with_millions_of_rows"</font><font>,
                       rs -> { callback.accept(myRowMapper.mapRow(rs, 0)); }
    );
    </font><font><i>//Pass the marker to the callback, to signal end of stream</i></font><font>
    callback.accept(marker);
});
</font>

此时,尚未执行查询。你可以做以下事情:

stream = stream.filter(row -> row.isPretty());

但仍然没有任何反应。当你做这样的事情时:

Optional<MyRow> row = stream.skip(100_000).limit(1000).findAny();

然后执行查询,将读取(并跳过)前十万行,并且每行将通过过滤器,直到漂亮地读取了一千行。

请,请,请不要使用它作为良好的WHERE子句和正确索引表的替代品。我主要使用这个东西生成报告,通过将元素映射到一个公共类型进行进一步处理来连接不相交类型的流(基本上,弥补了 Java 中缺少的联合类型)。

话虽如此,能够以流方式从数据库中读取行是非常酷的。


以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

设计沟通十器

设计沟通十器

Daniel M. Brown / 樊旺斌 / 机械工业出版社 / 2008-12 / 49.00元

本书提供了网站设计时所需的可交付文档资料包括:概念模型,站点地图,可用性报告等,这些文档资料是设计人员和客户进行交流的主要工具。本书深入讨论了文档推介和风险规避技巧,向你展示了如何将文档资料按要求制作成有效的交流工具。 本书内容全面,结构清晰,讲解详细。可作为网站设计人员的参考用书。 关于网站设计的多数讨论好像都着眼于流程的创建,然而,要想把概念变为现实,需要一整套强大的可交付文档资料......一起来看看 《设计沟通十器》 这本书的介绍吧!

HTML 压缩/解压工具
HTML 压缩/解压工具

在线压缩/解压 HTML 代码

正则表达式在线测试
正则表达式在线测试

正则表达式在线测试

RGB CMYK 转换工具
RGB CMYK 转换工具

RGB CMYK 互转工具