压缩十进制数据的一次实践

简介

在ERP应用中，经常要将大批量的数据下载到客户端本地，例如查询、报表或导出等功能，这将消耗大量的网络资源来传输数据，特别是在移动办公时，较大的数据量将造成等待时间太长。本文以压缩十进制数据为例描述了压缩这些数据的实践。

分析案例

在传输的数据中，有各种类型的数据，有字符串、时间、十进制或bool等，本文将分析十进制这种类型。

在.net中，decimal是一个可以描述较大数据和小数，且保证计算准确的数据类型，比较适合ERP应用，但是他占用的空间比较大，反编译其申明可以看到其占用4个int32，即高达16个字节。

private int flags;
private int hi;
private int lo;
private int mid;

直接调用GetBits(Decimal) : Int32[]存储显然不合算。

参考实现

微软内部是如何存储的呢？System.IO.BinaryWriter的默认实现还真的是这么干的，请看：

public virtual void Write(decimal value)
{
decimal.GetBytes(value, this._buffer);
this.OutStream.Write(this._buffer, 0, 0x10);
}

注:不过他有点耍赖,调用了internal的GetBytes减少了byte[]数组的不断创建。

在二进制序列化的实现System.Runtime.Serialization.Formatters.Binary.__BinaryWriter中,使用了较为巧妙的方法,他将其转换为字符串存储:

internal void WriteDecimal(decimal value)
{
this.WriteString(value.ToString(CultureInfo.InvariantCulture));
}

通常情况下,一列的数据都是很小的数字，例如数量或者单价,可能0,也可能正整数,也可能是0.17这样的小数.根据WriteString的实现可知，需要至少一个字节描述字符串的长度，然后一个字符占用一个字节（默认utf8），也就是说，数字0占用2个字节，0.17占用5个字节。

还不错啊，至少比16个字节好多了。

有没有更好的压缩

观察

我对微软的实现还不够满意，仔细观察常见的数据，可以总结道：

1、 通常一个小数由整数部分、小数部分和小数位数组成，例如0.17整数是0，小数部分是17，小数位数是2，另外一个例子1.007，整数部分是1，小数部分是7，小数位数是3；

2、 如果小数位数是0，即整数，那么就不必描述小数部分了；

3、 如果数字就是0，这种情况非常多，是不是用1个字节而不是2个字节描述呢？

4、 像17这样的数字，用字符串描述就需要2个字节，而如果用二进制，1个字节就够了。

基本方法

所以我的方法是：第一个字节描述小数位数，后面的字节描述整数部分，如果小数位数大于0，则还包含小数部分的整数描述。

例如0.12，我会存储2，0，12三个正整数，十六进制描述就是

02

00

0C

我在上面多处提到正整数，我们可以利用7BitEncodedInt方式填写整数部分和小数部分，他是一种可变长的描述方法。

如果数字是整数，我们还可以省略小数部分，例如12

00

0C

到这里，我们发现一个问题，如果是数字0，第一个字节和上面是一样的（为0），这样即使0也还需要再存储一个字节0，用来确定整数部分是0，不然读取时，就无法区分0和12了，但这就达不到我之前提到的数字0使用一个字节来描述的要求。

而且，这里还有负数的问题，还有如果decimal的值大于int32的范围怎么办？所以我在第一个字节上做了手脚，把他再拆分成4个部分。

A

A

A

A

A

B

C

D

第一个字节的8个位中，前5位存储小数位数（A区），最大值是31，一般小数位数不会超过这个数字（事实上，下面的算法最多仅9位）。

B区一个位，如果是0，表示当前数字是0，其他任何位都无需考虑了，也没有整数部分和小数部分，如果是1，表示此值非0，通过这种方法，我们就实现了数字0仅用一个字节描述的目的。

C区一个位，如果是0表示正数，如果是1表示此值为负数；

D区一个位，如果是0表示此值使用压缩的存储方法，1表示此值太大或太小，使用了16个字节的原样输出。

现在还是0.12为例，其第一个字节其二进制实际上是：

0

0

0

1

0

1

0

0

第一个字节换算成十六进制为：0x14，所以0.12其存储的内容是：

14

00

0C

参考实现

下面是写入时的代码，我使用了比较笨拙的办法获取小数部分的整数描述值。

// Fast access for 10^n where n is 0-9
private static UInt32[] Powers10 = new UInt32[] {
10,
100,
1000,
10000,
100000,
1000000,
10000000,
100000000,
1000000000
};

private static void WriteToFullByte(SerializeContext context, decimal[] values) {
//整个逻辑放在一个方法中，目的是防止foreach循环中不断跳入其他方法，增加消耗，在这个性能要求较高的程序是允许的。
var stream = context.Stream;
int flag;
int a, b, c;
decimal d, e;

foreach (var value in values) {
if (value == 0m) {
//当数字是0时，第一个字节的所有位都是0，直接写0
stream.WriteByte(byte.MinValue);
}
else {
//分析出整数部分，小数位和小数部分；
//ex: v = 1.070m;

//整数部分太大，超过了int32的描述范围；
//当value == Int32.MinValue时，其变为正数后，比Int32.MaxValue大了1，超过范围，所以后面一个判断使用<=
if (value > Int32.MaxValue || value <= Int32.MinValue) {
goto FullWrite;
}

if (value < 0) {
flag = 0x4 | 0x2;
e = value * -1;
a = decimal.ToInt32(e);
d = e - a;
}
else {
flag = 0x4;
//取出整数部分 a = 1
a = decimal.ToInt32(value);
//得到小数部分 d = 0.070m;
d = value - a;
}

c = 0;
if (d == 0m) {
//没有小数部分
stream.WriteByte((byte)flag);
context.Write7BitEncodedInt(a);
goto NextFor;
}

do {
e = d * Powers10[c]; //ex: 0.7 , 7
b = decimal.ToInt32(e);
c++;

if (b == e) {//整数部分和小数部分相等，则得到小数部分的整数描述
flag = (c << 3 | flag);
stream.WriteByte((byte)flag);
context.Write7BitEncodedInt(a);
context.Write7BitEncodedInt(b);
goto NextFor;
}
} while (c < 9);

//小数部分太多，也使用原始方法写入。
FullWrite:
stream.WriteByte((byte)1);
//这里使用了BinaryWriter来填充到流，此实现bw实例级缓存了byte[]并调用decimal内部的
//的 GetBytes 方法，这比调用GetBits不断创建数组减少了开销。
context.Writer.Write(value);

NextFor:
flag = 0;
}
}
}

下面是读取时的方法：

private static decimal[] PowersDecimal = new decimal[] {
0.1m,
0.01m,
0.001m,
0.0001m,
0.00001m,
0.000001m,
0.0000001m,
0.00000001m,
0.000000001m
};

private static void ReadFromFullByte(SerializeContext context, decimal[] values, int length) {
var stream = context.Stream;
int flag;
int a, b, c;
decimal d;

for (int i = 0; i < length; i++) {
flag = stream.ReadByte();
if (flag != 0) {
if (flag == 1) {
//较大的数，系统内部完整数据
values[i] = context.Reader.ReadDecimal();
}
else {
c = flag >> 3;
a = context.Read7BitEncodedInt();

if (c > 0) {
b = context.Read7BitEncodedInt();
//乘法比除法快，在测试用例中，除法时，总时间是1分04秒，而乘法时为50秒
d = a + ((decimal)b * PowersDecimal[c - 1]);
}
else {
d = a;
}

if ((flag & 0x2) == 0x2) {
values[i] = d * -1; //负数
}
else {
values[i] = d;
}
}
}
}
}

更进一步的优化

我在以上的基础上，进一步设想，

1、 如果这一列所有的数字都是0呢？这在ERP中非常常见，因为为了满足各种需求，设计人员总是设计一堆的字段，而这些字段一般用户根本不填写，全是0；

2、 如果这一列数字全是正整数，那么我是不是可以全部不要那个标志位字节，就节省一半的大小了，这种情况其实也常见，例如数量这一列，零售单明细中都是1~5之间的正整数，除非是称重的商品。

基于上面的想法，我在整列的存储中，使用一个字节描述存储方式，注意是整列的开头而不是每个数字的开头。这第一个字节：

1、 如果是0，表示所有的数字是0；

2、 如果是1，表示所有的数字都是小于0xFF的正整数，后面都使用一个字节描述其整数部分；

3、 如果是2，表示所有的数字都是小于0xFF FF的正整数，后面都使用2个字节描述其整数部分；

4、 如果是3，表示所有的数字都是小于0xFF FF FF的正整数，后面都使用3个字节描述其整数部分；

5、 如果是大于3的数字，表示此列使用本文描述的动态压缩方法存储数据。

下面是一段简单分析数据的方法，其返回此标志位：

private static byte GetFirstFlag(decimal[] values) {
decimal maxValue = 0;
foreach (var item in values) {
if (decimal.Floor(item) == item) {
if (item > maxValue) {
maxValue = item;
if (maxValue > 0xFFFFFF) {
return 4;//大于3个byte可描述范围 使用full的算法
}
}
else if (item < 0) {
return 4;//负数 使用string的算法
}
}
else {
return 4;//存在有效小数 使用full的算法
}
}

if (maxValue > 0xFFFF) {
return 3;// 65535 < maxValue <= 16777215 3个byte
}
else if (maxValue > 0xFF) {
return 2;//255 < maxValue <= 65535 2个byte
}
else if (maxValue > 0) {//0 < maxValue <= 255 1个byte
return 1;
}
return 0;
}

总结

通过不断分析ERP中的常见数据，理顺其规律，我们就能设计出更加优化的特定压缩算法。

完整的测试代码请点击这里下载。