Java实现zookeeper数据的备份以及恢复

zookeper学习日记

1、背景

项目为多团队共同开发，java管理端、flink分析端、C++协议端，项目目前为单机版本，使用zookeeper(后面简称zk)来完成配置的管理与共享，以供多个开发端的人员共享一些配置数据。
由于项目应用场景等原因，只考虑zk中持久节点数据的备份和恢复，并无权限的备份和恢复等。

2、zookeeper的目录存放和存储结构

本人菜鸟一枚，今年项目刚接触zk，所以底层存储过程以及方式也不太了解，下面有错误的地方欢迎指正。

2.1

服务器安装完zk后一般需要配置数据目录(dataDir)和日志目录(dataLogDir),本人配置如下：配置文件为conf下的zoo.cfg

dataDir=/root/zookeeper/data
dataLogDir=/root/zookeeper/logs

快照文件：

日志文件:

2.2 zk的存储方式和过程

zookeeper的存储过程简单可以这么理解：
zk的存储是为内存存储和磁盘存储结合起来的，内存中的数据会根据一定的规则定时刷新到磁盘中。
内存中的数据你可以理解为有一个内存数据库，存储了zk的会话信息、操作信息、节点信息、节点数据(DataTree结构)、事务日志等。然后磁盘文件则在 zoo.cfg配置文件中的数据目录(dataDir)和日志目录(dataLogDir)下，dataDir 对应 snapshot文件，dataLogDir 对应 log文件。随着时间的推移和zk数据的增删改，当到达一定次数时，zk会把内存中某一时刻的全量数据持久化到dataDir目录下。以snapshot文件在磁盘中存储，后面dataDir下会有多个snapshot文件,每一个snapshot都是某一时间点zk的全量数据快照。
于此类似，dataLogDir中也会有多个log文件，log文件是按操作顺序记录的一些事务日志，包括增删改等写操作，当然还有其他的一些信息。可以这么理解，snapshot记录的是某个时间点的所有数据信息，而log则是记录了这些数据在一段时间内的所有变化过程。
啰嗦了这么多想必对zk的存储应该有些概念了，既然知道zk的存储方式，那就可以想到方式一是如何来备份和恢复了。原则上可以备份数据目录下最新的快照文件和日志目录下最新的日志文件就可以了。

3、数据备份和恢复

	网上搜了一下找到了3种方案，简单说一下:
1.通过复制zk数据文件和日志文件来完成备份
2.zk-shell工具
3.自己写脚本或代码通过读写文件的方式

1.第一种方式为zk本身存储特性所支持。

就是把最新的快照文件和最新的日志文件拷贝到另一台服务器，清空另一台服务器的快照文件和日志文件，将拷贝的两份文件放入相应的数据目录和日志目录下，重启zk服务。这是zk会将磁盘上的数据加载到内存中，即可以通过zk的shell命令或工具查看恢复的数据。

假如要把A服务器上的zk数据备份下来，然后再服务器B上恢复，则大概步骤如下：

1.登录A系统，进入zk数据目录和日志目录。进行备份
[root@master version-2]# ls -alh
总用量 4.0K
drwxr-xr-x. 2 root root 23 9月 1 10:27 .
drwxr-xr-x. 3 root root 49 9月 1 10:27 …
-rw-r–r--. 1 root root 457 9月 1 10:27 snapshot.0

[root@master version-2]# ls -alh
总用量 20K
drwxr-xr-x. 2 root root 18 9月 1 10:43 .
drwxr-xr-x. 3 root root 22 9月 1 10:27 …
-rw-r–r--. 1 root root 65M 9月 1 13:56 log.1

如上，实际情况目录下可能有多个文件，将最新的snapshot和最新的log文件拷贝出来。
2.登录服务器B，将B的zk的数据目录和日志目录下的文件清空，将刚刚拷贝出的文件放到相应的目录下
3.启动B的zk服务(此时可通过工具或命令查看数据是否恢复正常)

注意:这种方式有些注意点，本人也只是听说，未实际测试过。
1.网上有人说不建议拷贝当天最新的文件，因为恢复时可能会发生错乱，或者数据不完整或者数据有损坏。
2.因为恢复时要清空服务器B上的zk历史数据文件和日志文件，万一拷贝的文件由于种种原因损坏不能恢复，则机器B上的数据也已经清空了。如果机器B上本身已有数据，建议备份一下。(当然这概率很小且只是猜测)

由于这种方式怀疑有风险并且需要重启zk(本人所在项目在运行时不允许重启zk服务)，故没采用这种备份方案。

2. 使用zk-shell工具

安装zk-shell工具
yum install python2-pip
pip install zk_shell

git地址:https://github.com/rgs1/zk_shell

此工具可以把zk数据写入一个json文件中。
假如想备份zk的根节点/，备份文件路径为
/tmp/backup.json
执行命令
zk-shell localhost:2181 --run-once ‘mirror / json://!tmp!zookeeper-backup.json/’
由于本人比较懒，懒得去git上去看使用方法，我只是想用一个备份和恢复的功能，不想为了这一点功能再去熟悉这个，所以也没采用这个方案。如想了解这个方案可以自行去git上看源码和使用文档。

3. 使用zk的一些javaApi来实现备份和恢复

api使用的为org.I0Itec.zkclient.ZkClient
maven依赖


            com.101tec
            zkclient
            0.11

3.1 实现简要思想

1.使用javaapi ZkClient，ZkClient为封装好的一个客户端api，可以对zkServer进行一些原子操作，节点及路径的增删改查等操作。
2.备份操作
ZkClient提供了路径及节点的crud方法，自己写一个递归方法，获取某个路径下所有的叶子节点的路径和数据，保存k-v形式写入磁盘文件。
3.恢复操作
先清空备份的节点下所有的节点，然后读取备份的文件，根据k-v，使用ZkClient 提供的写入方法，path-data写入备份的节点。

3.2 主要代码片段

1.UserTO.java，用来当做序列化对象

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class UserTO {
    private Long id;
    private String name;
    
    private Integer sex;
}

2.ZkConfig.java，对ZkClient的再一次封装，增加少量基础业务

import com.alibaba.fastjson.JSON;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.CreateMode;
import to.UserTO;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ZkConfig {
    private String zkUrl = "172.16.11.253:2181";
    // 父亲路径
    private String fatherPath = "/root/father";
    // 儿子路径
    private String childPath = "/root/father/child";
    // 叔叔路径
    private String unclePath = "/root/uncle";
    private ZkClient zkClient;

    
    public void initZkClient() {
        zkClient = new ZkClient(zkUrl);
        zkClient.setZkSerializer(new ApiZkSerializer());
        createPath(fatherPath);
        createPath(childPath);
    }

    
    ZkConfig() {
        zkClient = new ZkClient(zkUrl);
        zkClient.setZkSerializer(new ApiZkSerializer());
    }

    
    public void saveFatherNode(UserTO to) {
        String path = fatherPath + "/" + to.getId();
        zkRecursionChgNode(path, JSON.toJSONString(to));
    }

    
    public void saveChildNode(UserTO to) {
        String path = childPath + "/" + to.getId();
        zkRecursionChgNode(path, JSON.toJSONString(to));
    }

    
    public void saveUncleNode(UserTO to) {
        String path = unclePath + "/" + to.getId();
        zkRecursionChgNode(path, JSON.toJSONString(to));
    }

    
    private void zkChgNode(String path, String value) {
        // 如存在先删除节点再创建
        if (zkClient.exists(path)) {
            zkClient.delete(path);
        }
        try {
            Thread.sleep(200L);
        } catch (Exception e) {
            e.printStackTrace();
        }
        // 创建持久节点
        zkClient.create(path, value, CreateMode.PERSISTENT);
    }

    
    public Map getLeafNodeMap(String nodePath) {
        Map resultMap = new HashMap<>();
        List pathList = getLeafNodePath(nodePath);
        for (String path : pathList) {
            if (!zkClient.exists(path)) {
                continue;
            }
            String nodeData = zkClient.readData(path);
            resultMap.put(path, nodeData);
        }
        return resultMap;
    }

    
    public void zkRecursionChgNode(String path, String value) {
        // 如不存在则创建路径
        if (!zkClient.exists(path)) {
            zkClient.createPersistent(path, true);
        }
        zkClient.writeData(path, value);
    }

    
    private List getLeafNodePath(String parentPath) {
        List result = new ArrayList<>();
        List childPaths = zkClient.getChildren(parentPath);
        if (childPaths == null || childPaths.size() == 0) {
            result.add(parentPath);
            return result;
        }
        for (String childPath : childPaths) {
            String path = parentPath + "/" + childPath;
            result.addAll(getLeafNodePath(path));
        }
        return result;
    }

    
    public void createPath(String path) {
        if (zkClient.exists(path)) {
            return;
        }
        zkClient.createPersistent(path, true);
    }

    
    public String getDataByPath(String path) {
        if (!zkClient.exists(path)) {
            return null;
        }
        return zkClient.readData(path);
    }

    
    public void deleteNode(String nodePath) {
        if (zkClient.exists(nodePath)) {
            zkClient.delete(nodePath);
        }
    }

    
    public void deleteRecursive(String nodePath) {
        if (zkClient.exists(nodePath)) {
            zkClient.deleteRecursive(nodePath);
        }
    }
}

3.ZkTest.java测试类，一些功能函数

import org.junit.Test;
import to.UserTO;

import java.io.*;
import java.util.HashMap;
import java.util.Map;

public class ZkTest {
    private static ZkConfig zkConfig = new ZkConfig();

    @Test
    public void zkMain() {
        String nodePath = "/";
        // 备份
//        backups(nodePath, "F:/", "zkData.txt");
        // 恢复
        recovery(nodePath, "F:/zkData.txt");
    }

    @Test
    public void zkTest() {
        // 构造测试节点数据
        UserTO father = new UserTO(1L, "PigFather", 1);
        UserTO page = new UserTO(1L, "PigPage", 1);
        UserTO george = new UserTO(2L, "PigGeorge", 1);
        UserTO uncle = new UserTO(1L, "PigUncle", 1);
        // 增加节点
        zkConfig.saveFatherNode(father);
        zkConfig.saveChildNode(page);
        zkConfig.saveChildNode(george);
        zkConfig.saveUncleNode(uncle);
    }

    
    public void backups(String nodePath, String filePath, String fileName) {
        // 1.获取节点所有叶子节点路径以及数据
        Map result = zkConfig.getLeafNodeMap(nodePath);
        // 2.构造文本格式和数据，路径与数据用:分隔,每条数据换行隔开
        StringBuffer buffer = new StringBuffer();
        for (Map.Entry entry : result.entrySet()) {
            System.out.println("构造数写入据->" + entry.getKey() + ":" + entry.getValue() + "rn");
            buffer.append(entry.getKey()).append(":").append(entry.getValue()).append("rn");
        }
        // 3.将数据写入指定文件
        writeFile(filePath, fileName, buffer.toString());
    }

    
    public void recovery(String nodePath, String filePath) {
        // 1.读取文件构造map
        Map result = buildMap(filePath);
        // 2.删除节点以及子节点
        zkConfig.deleteRecursive(nodePath);
        // 3.将map中的数据写入zk中
        for (Map.Entry entry : result.entrySet()) {
            zkConfig.zkRecursionChgNode(entry.getKey(), entry.getValue());
        }
    }

    
    private static String writeFile(String filePath, String fileName, String content) {
        File dir = new File(filePath);
        if (!dir.exists()) {
            dir.mkdir();
        }
        String newPath = filePath + fileName;
        File file = new File(newPath);
        if (!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try (FileOutputStream fos = new FileOutputStream(newPath)) {
            fos.write(content.getBytes());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return newPath;
    }

    
    private static Map buildMap(String path) {
        Map result = new HashMap<>();
        InputStream in = null;
        try {
            in = new FileInputStream(path);
            BufferedReader br = new BufferedReader(new InputStreamReader(in));
            String str = null;
            while ((str = br.readLine()) != null) {
                // 按写入时的分隔符(冒号)进行解析，0:node路径 1:node数据
                String[] lineData = str.split(":", 2);
                result.put(lineData[0], lineData[1]);
                System.out.println("文件读取-> " + "节点路径: " + lineData[0] + "    节点数据: " + lineData[1]);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }
}

Java实现zookeeper数据的备份以及恢复

大数据系统相关栏目本月热门文章