DTeam 技术日志

Doer、Delivery、Dream

利用Java的Stream API处理海量数据

冯宇 Posted at — Mar 17, 2019 阅读

前言

最近优化一个旧的 HBase 查询程序，在 HBase 查询出来的数据量太大的时候就会出现性能下降、内存溢出的问题。遂着手优化这个问题，将解决过程记录下来。

问题描述

一个旧的 HBase 查询程序，会从 HBase 数据库查询所需要的记录，将查询结果以 JSON 形式返回给客户端，整个程序基于 Vertx 之上构建。但是我们都知道 HBase 存储着海量的数据，当数据量太大的时候，这个程序就会出现响应速度慢，甚至内存溢出等现象。最近准备迁移旧的系统，于是着手修复这个问题。

问题分析

首先很容易想到的就是查询数据 –> 响应给客户端这部分可能出了问题，先看看这部分程序的代码(groovy)。大致如下(代码示例而已，非完整可执行代码):

List<JsonObject> scanResult() {
    Scan scan = new Scan() // 其他查询条件忽略
    ResultScanner resultScanner = hTable.getScanner(scan)
    LinkedList list = new LinkedList()
    resultScanner.each { Result result ->
        List data = []
        result.rawCells().each {
            data << new JsonObject(Bytes.toString(CellUtil.cloneValue(it))
         }
         list.addFirst(data)
     }
     resultScanner.close()

     list
}

Closure responseToClinet(RoutingContext context) {
        { List result ->
            JsonObject data = new JsonObject()
            data.put('success', true)
            data.put('result', result)
            context.response().end(data.encodePrettily(), 'utf-8')
        }
}

将从 HBase 中查询的结果放入一个List<JsonObject>中，再将这个 json 格式化之后响应给客户端。所以这个问题就很明显了，如果这个Scan的结果集特别大，将会造成整个 List 内存溢出，而且将 List 格式化成 Json 也需要消耗大量的时间，这就造成客户端请求的时候往往等上半天。

第一次优化

首先解决一下 Json 格式化的消耗问题，将一个超大的 Map 在内存中格式化成 Json 输出给客户端，这个操作本身就需要消耗大量的时间和内存，所以先从这里入手，规避在最后把一个超大的 JsonObject 对象格式化，而是在输出之前我就手工格式化成 json 字符串流的形式，不需要在输出的时候做。正好 vertx 在HttpServerResponse是支持io.vertx.core.buffer.Buffer的，那么我在格式化List<JsonObject>这里的时候就手工拼接成一个String.join(",")的形式就好了

Buffer scanResult() {
    Scan scan = new Scan() // 其他查询条件忽略
    ResultScanner resultScanner = hTable.getScanner(scan)
    Buffer buffer = Buffer.buffer()
    resultScanner.each { Result result ->
        result.rawCells().each {
            String cellValue = Bytes.toString(CellUtil.cloneValue(it))
            if (buffer.length() > 0) {
                buffer.appendString(',' + cellValue)
            } else {
                buffer.appendString(cellValue)
            }
         }
     }
     resultScanner.close()

     buffer
}

Closure responseToClinet(RoutingContext context) {
        { Buffer result ->
            // 期望响应格式{"success":true,"result":[e1,e2,...,en]}
            HttpServerResponse response = context.response()
            response.setChunked(true)
            response.putHeader('Content-Type', 'application/json')
            response.write('{"success":true,"result":[')
            if (result) {
                response.write(result)
            }
            response.end(']}')
        }
}

很好，我们用手工拼接 Buffer 的形式，拼接出了一个超大的 json 字符串，这样节省了大量格式化 json 的时间。目前我们将一个原有 5 秒才能出结果的查询，降低到了 0.5 秒出结果，已经将速度大幅提升了。

但是这还不够，我们放开查询条件，进一步扩大 Scan 的结果集，依旧会出现内存溢出的错误。进一步审计代码和堆栈，发现这里出了问题:

    Buffer buffer = Buffer.buffer()
    resultScanner.each { Result result ->
        result.rawCells().each {
            String cellValue = Bytes.toString(CellUtil.cloneValue(it))
            if (buffer.length() > 0) {
                buffer.appendString(',' + cellValue)
            } else {
                buffer.appendString(cellValue)
            }
         }
     }

我们在循环中不断追加 buffer，但是结果集超大，超出我们的预期的时候，buffer 还是会把内存撑爆。看来我们还是得找更好的方案。

第二次优化

进一步审计ResultScanner的代码，发现它是这么定义的:

/**
 * Interface for client-side scanning.
 * Go to {@link Table} to obtain instances.
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public interface ResultScanner extends Closeable, Iterable<Result>

ResultScanner是一个Iterable对象。那我们可以这么考虑，能不能不要在迭代中一直拼接 Buffer，而是迭代中每格式化一个 Buffer 之后，就把 Buffer 写入响应流中呢？这样不就解决了拼接 buffer 造成内存溢出的问题了吗？而且可以让响应时间进一步缩短。

这里就是 Java 8 新引入的java.util.Stream API 了，不同于 nio 的Stream，这里的 Stream 是为了实现流式处理的接口，而且可以和 java 8 新加入的 Lambda 表达式进行更加灵活的处理。Stream 就像水流一样，处理完就丢掉了，不会向前遍历，可以很方便的构造数据处理管道，正好符合我们的这个场景。一个 Stream 的典型处理场景如下:

image.png

而Iterable对象是非常容易转成Stream的，在 java 8 就提供了一个工具类java.util.stream.StreamSupport，可以很方便的把Iterable对象转成Stream。于是我们的代码改造如下:

Itertor<Buffer> scanResult() {
    Scan scan = new Scan() // 其他查询条件忽略
    resultScanner = hTable.getScanner(scan)
    boolean first = true
    StreamSupport.stream(resultScanner.spliterator(), false).map({ Result result ->
        Buffer buffer = Buffer.buffer()
        result.rawCells().each {
            String cellValue = Bytes.toString(CellUtil.cloneValue(it))
            if (first) {
                buffer.appendString(cellValue)
            } else {
                buffer.appendString(',' + cellValue)
            }
            first = false
        }
        buffer
    }).iterator()
}

Closure responseToClinet(RoutingContext context) {
    { Iterator<Buffer> bufferIterator ->
        // 期望响应格式{"success":true,"result":[e1,e2,...,en]}
        HttpServerResponse response = context.response()
        response.setChunked(true)
        response.putHeader('Content-Type', 'application/json')
        response.write('{"success":true,"result":[')
        bufferIterator?.each {
            response.write(it)
        }
        response.end(']}')
    }
}

// 调用部分示例，需要在迭代完成close掉ResultScanner
{ Iterator<Buffer> bufferIterator ->
    resultScanner?.withCloseable {
        responseToClinet(context).call(bufferIterator)
    }
}

至此，所有问题都得到了解决，查询再大的数据量也不会造成内存溢出了。

小结

本章我们通过一个实际的项目案例分析，展示了大数据处理的思想，以及使用 Java 8 新增的 Stream API 解决我们实际的问题。

觉得有帮助的话，不妨考虑购买付费文章来支持我们 🙂 ：

付费文章

友情链接

相关文章