簡介：我們在開發中後臺應用或者中介軟體的時候，會儲存一些資料在記憶體中以加快訪問速度。隨著資料量的增加，除了可以放置於堆外，還可以通過實時壓縮來緩解。今天就給大家介紹一種壓縮整形陣列的方式。

作者 | 玄胤
來源 | 阿里技術公眾號

我們在開發中後臺應用或者中介軟體的時候，會儲存一些資料在記憶體中以加快訪問速度。隨著資料量的增加，除了可以放置於堆外，還可以通過實時壓縮來緩解。今天就給大家介紹一種壓縮整形陣列的方式。

一 資料壓縮

陣列指 long[] 或者 int[] 型別，在 Java 中應用很廣。當資料量很大時，其記憶體佔用的問題便突顯出來，原因是一個 long 型別是佔 8 個位元組，而 int 也是佔用 4 個位元組，當有千萬級別的資料時，其佔用的空間便是上百 MB 級別的了。

1 去冗餘

首先想到的就是縮減每個數字佔用的空間。因為我們都知道就正數而言，int 3 個位元組以上即可表示 2^24 = 16M 即 1600 百萬個數，而再往後，即使用第 4 個位元組，絕大多數我們是用不到的，但也不能砍掉，萬一還會用到的；所以可以將高位去掉，是一種可行的思路，但必須動態去掉，該用的時候還是得用，這就需要儲存用到多少個位元組了（見圖：數字壓縮基本原理）。

數字壓縮基本原理

表示資料佔用位元組數有兩種方式：一是借用原資料的幾位來表示，就拿 long 來說，我們只需要借用 3 位就可以覆蓋用到的位元組數了（因為 2ˆ3 = 8），由於 2^60 以後已經是非常大的數了，幾乎用不到，所以我們借用也基本不會產生負面效果；另一種就是利用位元組最高位表示還有剩餘資料（見圖2），Facebook 在 Thrift 中就是使用此方法來壓縮傳輸資料的。總之，我們就是要把 long 或者 int 陣列壓縮成 byte 陣列，在使用時再依據 byte 陣列中儲存的資訊將對應的數字還原。

解壓時識別資料大小方法

以上壓縮思路在傳輸場景下可以很好的解決存取問題，因為都是前進先出的思路，但是如果我們需要壓縮後的結構仍然具備陣列的下標訪問能力怎麼辦？

這裡的難點是：之前每個數字都是固定長度，我們可以通過 “[單個數字佔用的位元組數]*[第幾個]” 很快地找到對應的記憶體地址，但是壓縮過後每個數字佔用的空間不是一樣的，這種方式就失效了，我們無法得知第 N 個數所處的記憶體位置。要取下標為 200 的值，難道只能線性查詢 200 次嗎？顯然這樣的效率是相當低的，其時間複雜度就由 O(1) 下降為了 O(n)。有沒有更好的辦法呢？當然是有的。我們可以建立索引（如下圖），即：

• 將數字分為若干個桶，桶的大小可心調節（比如可以 1KB 一個桶，4KB 一個桶等）
• 我們使用另一個數組，大小為桶的數量，儲存每個桶所第一個資料所在的下標
• 在查詢時我首先使用二分查詢到對應的桶，再使用線性查詢到對應的資料

帶索引可提升壓縮後下標獲取速度

由於只是在桶內是線性查詢，而其一般不會太大，為 1KB 或者 4 KB（不能太少，因為每個桶的陣列指標也是需要佔用 16B 的）。由於第一次的索引二分查詢解決了大部分問題，查詢速度提升明顯，接近 O(logN)。使用這套方式，經測試，在 4 億隨機資料的情況佔用的空間可以縮減 30% 左右。經過簡單測試 TPS 可以達到 100w 級別，單次存取肯定是夠了。

壓縮效果

2 偏移量儲存

利用桶內是順序查詢的性質，我們可以只在桶內第一個元素存原數字，後面的都存一個偏移量，因為當資料不會明顯離散（即一會兒是十幾，一會是幾十億那種），可以很好地縮減資料大小，比如兩個數都佔用了 3 個位元組，存偏移量後，第二個數字就可以使用 1~2 個位元組來表示了。當然如果你對陣列本身的順序沒有要求的話，還可以先對陣列進行排序，這種偏移量的效果就可以暴表了。多數情況下，可以縮減 70% 以上。

二 存取優化

上述方案在線上某應用效能壓測時我們發現：單次隨機獲取沒有受到影響，但是批量順序獲取介面下降高達 97%，從 2800 多下降到了 72。經過研究發現，批量介面之所以下降明顯是由於觸及到了隨機獲取的 TPS 上限，在 ”圖：壓縮效果“ 中顯示，隨機獲取的極限 TPS 為 100w 左右，但是批量順序場景中每次批量操作會執行 1\~3w 次取操作，每次取操作走的是隨機獲取介面，所以只能是 72 這種兩位數的 TPS 了，因此我們需要深度優化壓縮資料的存取效率，為此採取瞭如下手段。

1 變固定長度桶為變長桶

之前採用二分查詢是因數我們採用定長的桶（即每個桶儲存的位元組數是相等的），每個桶儲存的數字數量不定，但如果我們採用變長桶，讓每個桶儲存 N 個數，那麼，便可以直接通過 “整除+求餘” 的方式快速打到數所在的桶，這樣在尋找桶下標的時候變可以以 O(1) 的複雜度找到，相比之前二分的 O(logn) 快了很多。經過測試批量介面的 TPS 增加為 320 左右，提升 4 倍以上，效果明顯。

非固定桶長度可以使索引塊長度固定，可快速查詢

2 編寫專用迭代器

批量其實也就是遍例操作，在之前遍例都是單獨一個一個取的，即每次都通過 get 介面。這個介面每次都會計算一遍桶的位置，然後是偏移量，再從桶開始處依據偏移量挨個查詢，在大量請求下效能開銷當然大。為此我們可以根據其順序取的特點專門設計一個迭代器，這個迭代器第一次初始化會記錄下桶的位置的，下一次就可以真接偏移一個數的長度 n 而直接找到一下個數據了，其時間複雜度為 O(1)。經過測試批量介面的 TPS 可以提升至 680 左右。

3 減少中間資料，使用棧直傳遞共用

在原來的解壓流程中，我們將資料從桶中讀取出來，然後傳遞給解決方法進行解壓，這裡會在堆在產生大量的中間資料，並且之前使用許多 ByteBuffer wrap 操作，wrap 每次都會新建一個 ByteBuffer 物件，相當的耗時。由於這均為只讀操作並且目前不支援資料刪除，我們可以直接引用桶內的資料，通過棧傳遞給解壓函式，這樣會快很多。

修改前的程式碼如下，其主要邏輯是

1. 計算數字所在的桶與偏移量，然後將其包裝成 ByteBuffer
2. 使用包裝好的 ByteBuffer 線性分析位元組陣列，通過偏移量查詢桶內數字
3. 依據數字的長度資訊（即前三個位）將對應的位元組複製至一個臨時陣列中
4. 將臨時陣列傳入 inflate 方法進行解壓

public long get(int ix) {
        // 首先尋找 shard, 由於每個桶儲存固定數量的數字，因此可以直接對映
        int i = ix / SHARD_SIZE;

        // 剩下的為需要線性查詢的偏移量
        ix %= SHARD_SIZE;

        ByteBuffer buf = ByteBuffer.wrap(shards[i]);

        // 找到對應資料的偏移量
        long offset = 0;
        if (ix > 0) {
            int len = (Byte.toUnsignedInt(buf.get(0)) >>> 5);

            byte[] bag = new byte[len];
            buf.get(bag, 0, len);
            offset = inflate(bag);
        }

        // 重置位置
        buf.position(0);

        int numPos = 0;
        while (ix > 0) {
            int len = (Byte.toUnsignedInt(buf.get(numPos)) >>> 5);

            numPos += len;
            ix -= 1;
        }
        buf.position(numPos);

        int len = (Byte.toUnsignedInt(buf.get(numPos)) >>> 5);

        byte[] bag = new byte[len];
        buf.get(bag, 0, len);

        return offset + inflate(bag);
    }
}

private static long inflate(byte[] bag) {
        byte[] num = {0, 0, 0 ,0 ,0 ,0, 0, 0};

        int n = bag.length - 1;
        int i;
        for (i = 7; n >= 0; i--) {
            num[i] = bag[n--];
        }

        int negative = num[i+1] & 0x10;

        num[i + 1] &= 0x0f;
        num[i + 1] |= negative << 63;

        return negative > 0 ? -ByteBuffer.wrap(num).getLong() : ByteBuffer.wrap(num).getLong();
    }
}

修改後的程式碼：

public long get(int ix) {
        // 首先尋找 shard, 由於每個桶儲存固定數量的數字，因此可以直接對映
        int i = ix / SHARD_SIZE;

        // 剩下的為需要線性查詢的偏移量
        ix %= SHARD_SIZE;

        byte[] shard = shards[i];

        // 找到對應資料的偏移量
        long offset = 0;
        if (ix > 0) {
            int len = (Byte.toUnsignedInt(shard[0]) >>> 5);
            offset = inflate(shards[i], 0, len);
        }

        int numPos = 0;
        while (ix > 0) {
            int len = (Byte.toUnsignedInt(shard[numPos]) >>> 5);

            numPos += len;
            ix -= 1;
        }

        int len = (Byte.toUnsignedInt(shard[numPos]) >>> 5);

        return offset + inflate(shards[i], numPos, len);
}

private static long inflate(byte[] shard, int numPos, int len) {
        byte[] num = {0, 0, 0 ,0 ,0 ,0, 0, 0};

        System.arraycopy(shard, numPos, num, num.length - len, len);

        int i = num.length - len;
        int negative = num[i] & 0x10;

        num[i] &= 0x0f;
        num[i] |= negative << 63;

        return negative > 0 ? -longFrom8Bytes(num) : longFrom8Bytes(num);
}

對比可以看出，這裡主要是去除了 bag 陣列這個中間變數，通過引用原 shard 中的資料直接去獲取資料對應的 byte 陣列，之前都是通過 ByteBuffer 去獲取桶中的位元組資料，現在我們都通過 shard[i] 直接查詢，效率高了很多。經過測試，這一優化可以提升 45% 左右的效能，直接將 TPS 拉昇至 910 多。

4 將堆資料變為棧資料

這個改造點有些難度的，對於中間變數來說，有些是可以避免的，我們可以使用上述的方式解決，但是有些是不能避免的，比如我們最後在解壓資料的時候，對於需要返回的數字，我們肯定需要一個臨時儲存的地方，這就是 inflate 第一行為什麼有個 byte[] num = {0, 0, 0 ,0 ,0 ,0, 0, 0}; 語句的原因。但是思考下，這個陣列只是為了儲存 long 的 8 個位元組資料，如果直接使用 long 那麼相當於是在棧上初始化了一個 8 位元組大小的陣列了，這裡需要解決的僅僅是針對 long 如何操作指定的位元組。其實這裡很簡單，我們只需要將對應位元組左移至相應的位置即可，例如我們需要對 long 的第二個位元組修改為 0x02 只需要如下操作：

    longData = (longData & ˜(0xff << 2 * 8)) | (0x02 << 2 * 8)

還有一個細節，就是我們直接從 byte[] 資料中取出的值是以有符號數表示的，直接合用上述上式位移會受符號位的影響，因此我們需要使用 0xff & byteAry[i] 的方式將其轉換成無符號的數。最後優化後的 inflate 方法如下：

private static long inflate(byte[] shard, int numPos, int len) {
-        byte[] num = {0, 0, 0 ,0 ,0 ,0, 0, 0};
+       long data = 0;

-       System.arraycopy(shard, numPos, num, num.length - len, len);
+       for (int i = 0; i < len; i++) {
+           data |= (long) Byte.toUnsignedInt(shard[numPos + i]) << (len - i - 1) * 8;
+       }

-       int i = num.length - len;
-       int negative = num[i] & 0x10;
+       // 檢視符號位
+       long negative = data & (0x10L << (len - 1) * 8);

-       num[i] &= 0x0f;
-       num[i] |= negative << 63;
+       // 將佔用位資料清零
+       data &= ~(0xf0L << (len - 1) * 8);

-       return negative > 0 ? -longFrom8Bytes(num) : longFrom8Bytes(num);
+       return negative > 0 ? -data : data;
}

在這裡優化後所有的堆資料申明都移除掉了，而且這裡還有個附帶優化，即之前採用臨時陣列的方式我們還需要將陣列轉換為 long，即 longFrom8Bytes 方法所起的作用，現在我們可以直接返回了，進一步的優化了效果，經過測試效能再次提升 35%， TPS 至 1250 左右。

5 內聯短函式

每次函式呼叫都需要進行一次進棧退棧操作，也是費時的，在日常程式中這些損耗都可以忽略不計，但是在本次批量情況下就被放大了，通過前面的程式碼我們可以發現 get 方法中有一個 updateOffset 的函式，這個功能其實很簡單，可以直接內聯，也就多了一行程式碼，如下：

private void updateOffset() {
            byte[] shard = shards[shardIndex];

            // 找到對應資料的偏移量
            int len = (0xff & shard[0]) >>> 5;

            curOffset = inflate(shard, 0, len);
}

我們將之內聯後表示如下：

if (numPos >= shard.length) {
                shardIndex++;
                numPos = 0;

-              updateOffset();
                // 找到對應資料的偏移量
+              shard = shards[shardIndex];
+              curOffset = inflate(shard, 0, (0xff & shard[0]) >>> 5);
}

還有一些例如 Byte.toUnsignedInt(int) 也就是簡單的一行程式碼，這種都可以直接複製出來去掉方法呼叫。

三 效能

最後，我們批量介面的 TPS 升級至了 1380 左右，相比於最開始 72 已經提升了近 20 倍。 雖然相比於原陣列還有些效能差距，但也是在同一個數量級上了。按照批量是按 5w 放大的計算，順序單次獲取的 TPS 已經達到 6500w，隨機單次 get 也達到了 260w TPS，完全足夠滿足生產需要了。

四 優化總結

從上面的優化我們可以得出：

1. Java 基本型別資料結構比物件結構快很多，越面向底層，越快
2. 堆上分配資料很慢，高頻呼叫還會頻繁觸發 Yong GC，對執行速度影響相當大，所以能棧絕不用堆
3. 物件呼叫慢於直接操作，因為需要進退棧，所以如果是幾行簡單呼叫，直接將邏輯複製調出會快很多，例如 Byte.toUnsignedInt()——當然，這是在極致效能下
4. 減少中間資料、減少臨時變數
5. 任何細小的效能損失在巨大的呼叫量在都會成倍擴大，所以對於批量介面要倍加小心

原文連結
本文為阿里雲原創內容，未經允許不得轉載。