在Java Stream实现大型查询

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

内容简介: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 中缺少的联合类型)。

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


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

查看所有标签

猜你喜欢:

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

算法基础

算法基础

[美] 托马斯 H.科尔曼(Thomas H.Cormen) / 王宏志 / 机械工业出版社 / 2015-12 / 59.00

本书介绍了什么是计算机算法,如何描述它们,以及如何来评估它们。这些计算机算法将提供:利用计算机搜索信息的简单方式;解决各种排序问题的方法;利用有向无环图和最短路径法来解决基本问题的方法(可用于建模公路网络,任务间的依赖及金融关系);解决字符串(例如DNA结构)问题的方法;密码学背后的基本原理;数据压缩的基础知识;以及甚至一些没有人能够理解如何在计算机上用相当长的时间来解决的问题。 本书适合作......一起来看看 《算法基础》 这本书的介绍吧!

URL 编码/解码
URL 编码/解码

URL 编码/解码

MD5 加密
MD5 加密

MD5 加密工具

UNIX 时间戳转换
UNIX 时间戳转换

UNIX 时间戳转换