Zookeeper-序列化

读完这篇文章你将会收获到

• 在 Zookeeper 源码项目中新建模块，使用 Jute 进行序列化和反序列化
• 修改 Jute 中的 buffer size 来序列化 / 反序列化大对象

序言

从 前面的文章 我们得知、ZK 的客户端和服务端会通过网络进行一系列的数据交互 (节点中的数据内容、ACL 信息)，而我们知道从一个内存对象到网络传输，那么就会涉及到序列化和反序列化操作。ZK 使用到是一个叫 Jute 的序列化组件 (对不起，我真的没听过，尴尬了)

Jute 介绍

Jute 是 ZK 中序列化的组件，前身是 Hadoop Record IO 中的序列化组件。

ZK 从第一个正式对外的版本开始，就一直使用 Jute 组件来进行网络数据传输和本地磁盘数据存储的序列化和反序列化工作。并不是 Jute 优秀到不被其他序列化框架所超越、而是替换这种基层组件、老版本的兼容性问题很难处理，并且 Jute 的序列化能力并不是 ZK 性能上的瓶颈，so 现在还是这个序列化组件。

Jute 使用

Talk is cheap. Show me the code

在 zk 源代码中新建一个模块

加入依赖

dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper-jute</artifactId>
            <version>3.7.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.7.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

创建 POJO , 并实现 Record 接口

@Data
@Accessors(chain = true)
@AllArgsConstructor
@NoArgsConstructor
public class Person implements Record {
    private String name;
    private int age;
    @Override
    public void serialize(OutputArchive archive, String tag) throws IOException {archive.startRecord(this, tag);
        archive.writeInt(age, "age");
        archive.writeString(name, "name");
        archive.endRecord(this, tag);
    }

    @Override
    public void deserialize(InputArchive archive, String tag) throws IOException {archive.startRecord(tag);
        age = archive.readInt("age");
        name = archive.readString("name");
        archive.endRecord(tag);
    }
}

创建单元测试

    @Test
    public void serializeTest() throws IOException {
        // 开始序列化
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        BinaryOutputArchive binaryOutputArchive = BinaryOutputArchive.getArchive(byteArrayOutputStream);
        new Person("coderLi", 100).serialize(binaryOutputArchive, "person");
        // 通常是 TCP 网络传输对象
        ByteBuffer byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
        // 反序列化
        ByteBufferInputStream byteBufferInputStream = new ByteBufferInputStream(byteBuffer);
        BinaryInputArchive binaryInputArchive = BinaryInputArchive.getArchive(byteBufferInputStream);
        Person person = new Person();
        person.deserialize(binaryInputArchive, "person");
        System.out.println(person.toString());

        byteArrayOutputStream.close();
        byteBufferInputStream.close();}

运行

Person(name=coderLi, age=100)

可能出现的问题

java.io.IOException: Unreasonable length = 1668244581
    at org.apache.jute.BinaryInputArchive.checkLength(BinaryInputArchive.java:166)
    at org.apache.jute.BinaryInputArchive.readString(BinaryInputArchive.java:116)
    at com.coder.li.data.Person.deserialize(Person.java:36)

根据报错信息找到 BinaryInputArchive

public class BinaryInputArchive implements InputArchive {

    public static final String UNREASONBLE_LENGTH = "Unreasonable length =";

    // CHECKSTYLE.OFF: ConstantName - for backward compatibility
    public static final int maxBuffer = Integer.getInteger("jute.maxbuffer", 0xfffff);
    // CHECKSTYLE.ON:
    private static final int extraMaxBuffer;

    static {
        final Integer configuredExtraMaxBuffer =
            Integer.getInteger("zookeeper.jute.maxbuffer.extrasize", maxBuffer);
        if (configuredExtraMaxBuffer < 1024) {
            // Earlier hard coded value was 1024, So the value should not be less than that value
            extraMaxBuffer = 1024;
        } else {extraMaxBuffer = configuredExtraMaxBuffer;}
    }
    ....
    ...
    private void checkLength(int len) throws IOException {if (len < 0 || len > maxBufferSize + extraMaxBufferSize) {throw new IOException(UNREASONBLE_LENGTH + len);
        }
    }

我们系统中并没有配置这两个参数，所以这俩个 buffer size 的和应该是 1024

我们在启动参数中配置

-Djute.maxbuffer=0 -Dzookeeper.jute.maxbuffer.extrasize=1668244581

只要 maxBufferSize + extraMaxBufferSize 和大于等于 1668244581 即可 (注意不要设置过大、导致相加结果溢出变为负数)

后来排查问题发现原来导致这个问题的出现是在反序列化的时候、自己写代码出错导致，但是借此来认识到一个坑还是不错的

或者你在实体类中有一个非常大的对象需要被序列化和反序列化、也会抛出这个异常

private String name;
    private int age;
    private byte[] bigData;

    public Person() {}

    public Person(String name, int age) {
        this.age = age;
        this.name = name;
        bigData = new byte[2048 * 2048];
    }

Jute 使用流程

• 实体类实现接口 Record 的 serialize 和 deserialize
• 构造 BinaryOutputArchive
• 序列化
• 反序列化

相关组件

Record

public interface Record {void serialize(OutputArchive archive, String tag) throws IOException;
    void deserialize(InputArchive archive, String tag) throws IOException;
}

在 Zookeeper Watcher 流程分析 文章中我们分析的 WatcherEvent 这个实体类就是实现了 Record 接口

InputArchive/OutputArchive

这两个接口分别是 Jute 底层序列化器和反序列化器接口定义，实现类主要有 BinaryOutputArchive 和 BinaryInputArchive

public interface InputArchive {byte readByte(String tag) throws IOException;

boolean readBool(String tag) throws IOException;

int readInt(String tag) throws IOException;

long readLong(String tag) throws IOException;

float readFloat(String tag) throws IOException;

double readDouble(String tag) throws IOException;

String readString(String tag) throws IOException;

byte[] readBuffer(String tag) throws IOException;

void readRecord(Record r, String tag) throws IOException;
..........
}

都是定义了一些非常基本的方法

其实都是依赖于 Java 的 InputStream 和 OutputStream 进行操作的

相关文章

编译运行 Zookeeper 源码

ZooKeeper 数据模型：节点的特性

Zookeeper-Access Control List

Zookeeper Watcher 流程分析