zookeeper典型应用场景总结

分布式一致性配置

　　在集群环境下，挨个更改配置是比较繁琐的，使用zookeeper可以实现同步配置。

1、配置信息

package com.zk;

import java.io.Serializable;

/**
 * 模拟公共配置类
 * 
 * @author Zomi
 */
public class DbConfig implements Serializable{
    private static final long serialVersionUID = -4483388642208582886L;
    
    // 数据库配置,有默认值
    private String url = "jdbc:mysql://127.0.0.1:3306/mydata?useUnicode=true&characterEncoding=utf-8";
    private String username = "root";
    private String password = "root";
    private String driverClass = "com.mysql.jdbc.Driver";


    public DbConfig(String url, String username, String password, String driverClass) {
        this.url = url;
        this.username = username;
        this.password = password;
        this.driverClass = driverClass;
    }

    public DbConfig() {}
    
    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getDriverClass() {
        return driverClass;
    }

    public void setDriverClass(String driverClass) {
        this.driverClass = driverClass;
    }

    @Override
    public String toString() {
        return "CommConfig [url=" + url + ", username=" + username + ", password=" + password + ", driverClass="
                + driverClass + "]";
    }

}

2、配置管理服务

package com.zk;

import org.I0Itec.zkclient.ZkClient;

/**
 * zk配置管理服务器，用于将配置信息的修改同步到zk上
 * @author Zomi
 */
public class ZkConfigMng {
    private String nodePath = "/dbConfig";
    private DbConfig dbConfig;
    private ZkClient zkClient;
    
    public ZkConfigMng() {
        this.zkClient = new ZkClient("192.168.31.130:2181");
    }
    //更新配置
    public DbConfig update(DbConfig dbConfig) {
        this.dbConfig = dbConfig;
        syncConfigToZookeeper();//将配置变更同步给zk
        return dbConfig;
    }
    private void syncConfigToZookeeper() {
        if(!zkClient.exists(nodePath)) {
            zkClient.createPersistent(nodePath);
        }
        zkClient.writeData(nodePath, dbConfig);
    }
}

3、模拟应用服务集群，具备监听配置变更的功能

package com.zk;

import java.util.concurrent.TimeUnit;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

/**
 * 模拟多服务器
 * 
 * @author Zomi
 */
public class ZkConfigClient implements Runnable {

    private String nodePath = "/dbConfig";
    private DbConfig dbConfig;
    
    @Override
    public void run() {
        ZkClient zkClient = new ZkClient(new ZkConnection("192.168.31.130:2181", 5000));
        while (!zkClient.exists(nodePath)) {
            System.out.println("配置节点不存在!");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        dbConfig = (DbConfig)zkClient.readData(nodePath);
        System.out.println(Thread.currentThread().toString() +"原数据为=="+ dbConfig);
        
        //监听配置数据的改变
        zkClient.subscribeDataChanges(nodePath, new IZkDataListener() {
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println(Thread.currentThread().toString() + "监听到节点：" + dataPath + "被删除了！");
            }
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                System.out.println(Thread.currentThread().toString() + "监听到节点：" + dataPath + ", 数据：" + data + " - 更新");
                dbConfig = (DbConfig) data;
            }
        });
    }
}

package com.zk;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BusinessServers {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        // 模拟多个服务器获取配置
        executorService.submit(new ZkConfigClient());
        executorService.submit(new ZkConfigClient());
        executorService.submit(new ZkConfigClient());
    }
}

4、更新配置信息进行测试

package com.zk;

import java.util.concurrent.TimeUnit;

public class MngServer {

    public static void main(String[] args) throws InterruptedException {
        
        DbConfig dbConfig = new DbConfig();
        ZkConfigMng zkConfigMng = new ZkConfigMng();
        DbConfig xx = zkConfigMng.update(dbConfig);//初始化节点及数据
        System.out.println(xx);
        
        TimeUnit.SECONDS.sleep(10);
        
        // 修改值        
        zkConfigMng.update(new DbConfig("jdbc:mysql://192.168.6.6:3306/mydata?"
                + "useUnicode=true&characterEncoding=utf-8","admin", "admin", "com.mysql.jdbc.Driver"));
        
    }

}

5、测试结果

 分布式锁

通常实现分布式锁有如下方式：

1、基于数据库锁。（for update悲观锁；或者版本号乐观锁）

2、基于redis方式。（可参考此文https://www.cnblogs.com/zomicc/p/12468324.html）

3、基于zookeeper方式。

Zookeeper中的节点分为四种类型：

• 持久节点（PERSISTENT）
• 持久顺序节点（PERSISTENT_SEQUENTIAL）
• 临时节点（EPHEMERAL）
• 临时顺序节点（EPHEMERAL_SEQUENTIAL）

zookeeper分布式锁基于第四种临时顺序节点实现分布式锁。原理图如下：

        

下面的代码是根据Zookeeper的开源客户端Curator实现分布式锁。采用zk的原生API实现会比较繁琐，所以这里就直接用Curator这个轮子，采用Curator的acquire和release两个方法就能实现分布式锁。

1、依赖包

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.2.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.2.0</version>
</dependency>

2、测试代码

package com.zk.distributeLock;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorDistributeLock {

    public static void main(String[] args) throws IOException {
        //参数一：表示获取不到锁，则1000ms后重试 
        //参数二：最多重试3次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        
        ExecutorService excutor = Executors.newFixedThreadPool(5);
        for(int i=0;i<5;i++) {//使用线程池模拟同时5个请求
            excutor.submit(new Runnable() {
                InterProcessMutex mutex;
                CuratorFramework client;
                @Override
                public void run() {
                    try {
                        client = CuratorFrameworkFactory.newClient("192.168.31.130:2181", retryPolicy);
                        client.start();
                        mutex = new InterProcessMutex(client, "/curator/lock");
                        try {
                            mutex.acquire();
                            System.out.println(Thread.currentThread().toString() + " 获取到了锁");
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    } finally {
                        System.out.println(Thread.currentThread().toString() + " 释放了锁");
                        try {
                            mutex.release();
                            client.close();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
        excutor.shutdown();
    }
}

3、测试结果

发现不会如线程抢占式那样交替执行，而是顺序执行。只有获取锁的线程才会执行，其他线程等待。

集群负载均衡

　　概念这里不多讲，直接看代码。

1、服务提供者

package com.zk.colony;

import java.util.concurrent.TimeUnit;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.zookeeper.CreateMode;

/**
   *      服务提供者
 * @author Zomi
 *
 */
public class ServiceProvider {
    static String ZOOKEEPER_STR = "192.168.31.130:2181";
    static String NODE_PATH = "/service";
    static String SERVICE_NAME = "/myService";
    
    private ZkClient zkClient;
    
    public ServiceProvider() {
        zkClient = new ZkClient(new ZkConnection(ZOOKEEPER_STR));
        System.out.println("provider sucess connected to zookeeper server!");
        if(!zkClient.exists(NODE_PATH)) {
            zkClient.create(NODE_PATH, "my test service", CreateMode.PERSISTENT);
        }
    }
    
    public void registryService(String serviceIp, Object obj) {
        if(!zkClient.exists(NODE_PATH + SERVICE_NAME)) {            
            zkClient.create(NODE_PATH + SERVICE_NAME, "provider services list", CreateMode.PERSISTENT);        
        }
        //对本机服务进行注册
        zkClient.createEphemeral(NODE_PATH + SERVICE_NAME + "/" + serviceIp, obj);
        System.out.println("注册成功！[" + serviceIp + "]");
    }
    /**
     * 服务启动、注册服务
     * @param args
     * @throws InterruptedException 
     */
    public static void main(String[] args) throws InterruptedException {
        System.out.println("服务器开始运行了.......");
        ServiceProvider provider = new ServiceProvider();
        provider.registryService("1.1.1.1", "1.1.1.1 data");
        TimeUnit.SECONDS.sleep(20);
        provider.registryService("2.2.2.2", "2.2.2.2 data");
        TimeUnit.SECONDS.sleep(20);
        provider.registryService("3.3.3.3", "3.3.3.3 data");
        System.out.println("服务器异常宕机了.......");
    }
}

2、服务消费者

package com.zk.colony;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.zookeeper.CreateMode;

/**
 * 消费者，通过某种负载均衡算法选择一个提供者进行消费
 * @author Zomi
 *
 */
public class ServiceConsumer {
    static String ZOOKEEPER_STR = "192.168.31.130:2181";
    static String NODE_PATH = "/service";
    static String SERVICE_NAME = "/myService";
    
    private ZkClient zkClient;
    
    private List<String> serviceList = new ArrayList<String>();
    
    public ServiceConsumer() {
        zkClient = new ZkClient(new ZkConnection(ZOOKEEPER_STR));
        System.out.println("consumer sucess connected to zookeeper server!");
        if(!zkClient.exists(NODE_PATH)) {
            zkClient.create(NODE_PATH, "my test service", CreateMode.PERSISTENT);
        }
    }
    
    /**
     * 订阅服务
     */
    public void subscribeSerivce() {
        serviceList = zkClient.getChildren(NODE_PATH + SERVICE_NAME);//下载注册列表
        zkClient.subscribeChildChanges(NODE_PATH + SERVICE_NAME, new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                //注意：此处没有考虑空
                if(serviceList.size() != currentChilds.size() || !serviceList.containsAll(currentChilds)) {
                    System.out.println("服务列表变化了，我要更新了:" + currentChilds);
                    serviceList = currentChilds;
                }
            }
        });
    }
    /**
     * 模拟调用服务
     */
    public void consume() {
        if(serviceList!=null && serviceList.size()>0) {
            //这里使用随机访问某台服务
            int index = new Random().nextInt(serviceList.size());
            System.out.println("调用[" + NODE_PATH + SERVICE_NAME + "]服务：" + serviceList.get(index));
        }else {
            System.out.println("没有服务提供者");
        }
    }
    
    /**
     * 客户端调用启动
     * @param args
     */
    public static void main(String[] args) {
        ServiceConsumer consumer = new ServiceConsumer();
        new Thread() {
            public void run() {
                while(true) {
                    try {
                        consumer.subscribeSerivce();
                        TimeUnit.SECONDS.sleep(5);
                        consumer.consume();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
        }.start();
        
    }
}

3、先启动消费者，再启动服务者。运行结果如下：

      

服务高可用

原理很简单，都是基于zk临时节点的特性，及监听机制。下面的图，一看就懂：

备服务监听到主服务挂掉了，立马当先，对外提供服务。

 分布式协调

库存、订单的例子，保证库存先扣减成功，再订单处理成功（然后才能通知快递员过来揽件之类的，否则没有货订单却创建成功了，快递小哥就白跑了，整个工作就显得很不协调了）。