Java读取Level-1行情dbf文件极致优化

Posted 2020-07-29 Binhua 的技术博客

最近架构一个项目，实现行情的接入和分发，需要达到极致的低时延特性，这对于证券系统是非常重要的。接入的行情源是可以配置，既可以是Level-1，也可以是Level-2或其他第三方的源。虽然Level-1行情没有Level-2快，但是作为系统支持的行情源，我们还是需要优化它，使得从文件读取，到用户通过socket收到行情，端到端的时延尽可能的低。本文主要介绍对level-1行情dbf文件读取的极致优化方案。相信对其他的dbf文件读取应该也有借鉴意义。

Level-1行情是由行情小站，定时每隔几秒把dbf文件（上海是show2003.dbf，深圳是sjshq.dbf）更新一遍，用新的行情替换掉旧的。我们的目标就是，在新文件完成更新后，在最短时间内将文件读取到内存，把每一行转化为对象，把每个列转化为对应的数据类型。

我们一共采用了6种优化方式。

 

我们在上文《Java读取Level-1行情dbf文件极致优化（1）》中，介绍了2种我们使用的优化策略：

优化一：采用内存硬盘（RamDisk）

优化二：采用JNotify，用通知替代轮询

 

本文继续介绍：

 

优化三：采用NIO读取文件

对于Dbf文件的读写，有许多的开源的实现，选择和改进它们是这里的重要策略。

 

有许多Dbf库是基于流的I/O实现的，即InputStream和OutStream。我们应该采用NIO的方式，即基于RandomAccessFile，FileChannel和ByteBuffer。流的方式是一边处理数据，一边从文件中读取，而采用NIO可以一次性把整个文件加载到内存中。有测试表明（见《Java程序性能优化》一书），NIO的方式大概比流的方式快5倍左右。我这里提供采用NIO实现的dbf读取库供大家下载学习（最原始的出处已不可考了。这个代码被改写了，其中也已经包含我之后将要提出的优化策略），如果你的项目已经有dbf库，建议基于本文的优化策略进行改进，而不是直接替换为我提供的。

DBFReader库

 

其中，DBFReader.java中有如下代码片段：

创建FileChannel代码为：

this.dbf = new RandomAccessFile(file, "r");
this.fc = dbf.getChannel();

 

把指定的文件片段加载到ByteBuffer的代码为

private ByteBuffer loadData(int offset, int length) throws IOException {
        // return fc.map(MapMode.READ_ONLY, offset, length).load();
        ByteBuffer b = ByteBuffer.allocateDirect(length);
        fc.position(offset);
        fc.read(b);
        b.rewind();
        return b;

    }

 

以上，我们使用ByteBuffer.allocateDirect(length)创建ByteBuffer。 allocateDirect方法创建的是DirectBuffer，DirectBuffer分配在”内核缓存区”，比普通的ByteBuffer快一倍，这也有利于我们程序的优化。但是DirectBuffer的创建和销毁更耗时，在我们接下来的优化中将要解决这一问题。

（我不打算详细介绍NIO的相关知识（可能我也讲不清楚），也不打算详细介绍DbfReader.java的代码，只重点讲解和性能相关的部分，接下来也是如此。）

 

优化四：减少读取文件时内存反复分配和GC

以上我提供的DBFReader.java文件读取的文件的基本步骤是 ：

1，把整个文件（除了文件头）读取到ByteBuffer当中（其实为DirectBuffer）

2，再把每一行从ByteBuffer读取到一个个byte[]数组中。

3，把这些byte[]数组封装在一个一个Record对象中（Record对象提供了从byte[]中读取列的各种方法）。

见以下loadRecordsWithOutDel方法：

private List<Record> loadRecordsWithOutDel() throws IOException {

        ByteBuffer bb = loadData(getDataIndex(), getCount() * getRecordLength());

        List<Record> rds = new ArrayList<Record>(getCount());
        for (int i = 0; i < getCount(); i++) {
            byte[] b = new byte[getRecordLength()];
            bb.get(b);

            if ((char) b[0] != \'*\') {
                Record r = new Record(b);
                rds.add(r);
            }
        }

        bb.clear();

        return rds;
    }

 

private ByteBuffer loadData(int offset, int length) throws IOException {
        // return fc.map(MapMode.READ_ONLY, offset, length).load();
        ByteBuffer b = ByteBuffer.allocateDirect(length);
        fc.position(offset);
        fc.read(b);
        b.rewind();
        return b;

    }

 

考虑到我们系统的实际应用的情况：行情dbf文件每隔几秒就会刷新一遍，刷新后的大小基本上差不多，格式是完全一样的，每行的大小是一样的。

 

注意看以上代码中高亮的部分，会反复创建ByteBuffer和byte数组。在我们的应用场景下，完全可以使用一种缓存机制来重复使用他们，避免反复创建。要知道一个行情文件有5000多行之多，避免如此之多的new和GC，肯定对性能有好处。

 

我添加了一个CacheManager类来完成这个工作：

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

public class CacheManager {
    
    private ByteBuffer byteBuffer = null;
    private int bufSize = 0;
    
    private List<byte[]> byteArrayList = null;
    private int bytesSize = 0;
    
    public CacheManager()
    {
    }
    
    public ByteBuffer getByteBuffer(int size)
    {
        if(this.bufSize < size)
        {
            byteBuffer = ByteBuffer.allocateDirect(size + 1024*8); //多分配一些，避免下次重新分配
            this.bufSize = size + 1024*8;
        }
        byteBuffer.clear();
        return byteBuffer;
    }
    
    public List<byte[]> getByteArrayList(int rowNum, int byteLength) //rowNum为行数，即需要的byte[]数量，byteLength为byte数组的大小 
    {
        if(this.bytesSize!=byteLength)
        {
            byteArrayList = new ArrayList<byte[]>();
            this.bytesSize = byteLength;
        }
        
        if(byteArrayList.size() < rowNum)
        {
            int shouldAddRowCount = rowNum - byteArrayList.size()+100; //多分配100行
            for(int i=0; i<shouldAddRowCount; i++) 
            {
                byteArrayList.add(new byte[bytesSize]);
            }
        }
        
        return byteArrayList;
    }
    
}

 

CacheManager 管理了一个可以反复使用的ByteBuffer，以及可以反复使用的byte[]列表。

 

其中，getByteBuffer方法用于返回一个缓存的ByteBuffer。其只有当缓存的ByteBuffer小于指定的大小时，才重新创建ByteBuffer。（为了尽量避免这种情况，我们总是分配比实际需要大一些的ByteBuffer）。

 

其中，getByteArrayList方法用于返回缓存的byte[]列表。其只有当需要的Byte[]数量小于需要的数量时，创建更多的byte[]； 如果缓存的byte[]们的长度和需要的不符，就重新创建所有的byte[]（这种情况不可能发生，因为每行的大小不会变，代码只是以防万一而已）。

 

将loadRecordsWithOutDel改造为recordsWithOutDel_efficiently，采用缓存机制：

public List<byte[]> recordsWithOutDel_efficiently(CacheManager cacheManager) throws IOException {

        ByteBuffer bb = cacheManager.getByteBuffer(getCount() * getRecordLength());
        fc.position(getDataIndex());
        fc.read(bb);
        bb.rewind();
        List<byte[]> rds = new ArrayList<byte[]>(getCount());
        List<byte[]> byteArrayList = cacheManager.getByteArrayList(getCount(), getRecordLength());
        for (int i = 0; i < getCount(); i++) {
            byte[] b = byteArrayList.get(i);
            bb.get(b);

            if ((char) b[0] != \'*\') {
                rds.add(b);
            }
        }

        bb.clear();
        return rds;
    }

 

在新的recordsWithOutDel_efficiently中，我们从CacheManager中分配缓存的ByteBuffer和缓存的byte[]。而不是从系统分配，从而减少了反复的内存分配和GC。（另外，recordsWithOutDel_efficiently直接返回byte[]列表，而不是Record列表了）

 

我的测试发现，优化步骤四，即使用缓存的方式，大概把时间从5ms左右降到了2ms多，提高大概一倍。

 

到此，我们只是完成了文件到内存的读取。接着是为每一行创建一个行情对象，从byte[]中把每一列数据读取出来。  我发现，其耗时远远超过文件读取，在没有优化的情况下，对5000多行数据的转换超过70ms。这是我们接下来需要介绍的优化策略。

 

待续。。。

 

 

Binhua Liu原创文章，转载请注明原地址http://www.cnblogs.com/Binhua-Liu/p/5615299.html