1.前言

之前自己寫了一些關(guān)于Zookeeper的基礎(chǔ)知識，Zookeeper作為一種協(xié)調(diào)分布式應(yīng)用高性能的調(diào)度服務(wù)，實(shí)際的應(yīng)用場景也非常的廣泛，這里主要通過幾個例子來具體的說明Zookeeper在特定場景下的使用方式（下面的這些功能估計consul和etcd也能實(shí)現(xiàn)，以后學(xué)到了再說吧）。

2.具體應(yīng)用

2.1.一致性配置管理

我們在開發(fā)的時候，有時候需要獲取一些公共的配置，比如數(shù)據(jù)庫連接信息等，并且偶然可能需要更新配置。如果我們的服務(wù)器有N多臺的話，那修改起來會特別的麻煩，并且還需要重新啟動。這里Zookeeper就可以很方便的實(shí)現(xiàn)類似的功能。

2.1.1.思路

將公共的配置存放在Zookeeper的節(jié)點(diǎn)中

應(yīng)用程序可以連接到Zookeeper中并對Zookeeper中配置節(jié)點(diǎn)進(jìn)行讀取或者修改（對于寫操作可以進(jìn)行權(quán)限驗(yàn)證設(shè)置），下面是具體的流程圖：

2.1.2.事例

數(shù)據(jù)庫配置信息一致性的維護(hù)

配置類：

public class CommonConfig implements Serializable{
 // 數(shù)據(jù)庫連接配置
 private String dbUrl;
 private String username;
 private String password;
 private String driverClass;
 
 public CommonConfig() {}
 
 public CommonConfig(String dbUrl, String username, String password, String driverClass) {
  super();
  this.dbUrl = dbUrl;
  this.username = username;
  this.password = password;
  this.driverClass = driverClass;
 }
 
 public String getDbUrl() {
  return dbUrl;
 }
 
 public void setDbUrl(String dbUrl) {
  this.dbUrl = dbUrl;
 }
 
 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 "CommonConfig:{dbUrl:" + this.dbUrl +
    ", username:" + this.username + 
    ", password:" + this.password + 
    ", driverClass:" + this.driverClass + "}";
 }
}

配置管理中心

• 獲取本地配置信息
• 修改配置，并同步

同步配置信息到Zookeeper服務(wù)器

public class ZkConfigMng {
 private String nodePath = "/commConfig";
 private CommonConfig commonConfig;
 private ZkClient zkClient;
 
 public CommonConfig initConfig(CommonConfig commonConfig) {
  if(commonConfig == null) {
   this.commonConfig = new CommonConfig("jdbc:mysql://127.0.0.1:3306/mydata?useUnicode=true&characterEncoding=utf-8",
     "root", "root", "com.mysql.jdbc.Driver"); 
  } else {
   this.commonConfig = commonConfig;
  }
  return this.commonConfig;
 }
 
 /**
  * 更新配置
  * 
  * @param commonConfig
  * @return
  */
 public CommonConfig update(CommonConfig commonConfig) {
  if(commonConfig != null) {
   this.commonConfig = commonConfig;
  }
  syncConfigToZookeeper();
  return this.commonConfig;
 }
 
 public void syncConfigToZookeeper() {
  if(zkClient == null) {
   zkClient = new ZkClient("127.0.0.1:2181");
  }
  if(!zkClient.exists(nodePath)) {
   zkClient.createPersistent(nodePath);
  }
  zkClient.writeData(nodePath, commonConfig);
 }
}

以上是提供者，下面我們需要一個客戶端獲取這些配置

public class ZkConfigClient implements Runnable {
 
 private String nodePath = "/commConfig";
 
 private CommonConfig commonConfig;
 
 @Override
 public void run() {
  ZkClient zkClient = new ZkClient(new ZkConnection("127.0.0.1:2181", 5000));
  while (!zkClient.exists(nodePath)) {
   System.out.println("配置節(jié)點(diǎn)不存在!");
   try {
    TimeUnit.SECONDS.sleep(1);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
  // 獲取節(jié)點(diǎn)
  commonConfig = (CommonConfig)zkClient.readData(nodePath);
  System.out.println(commonConfig.toString());
  zkClient.subscribeDataChanges(nodePath, new IZkDataListener() {
   
   @Override
   public void handleDataDeleted(String dataPath) throws Exception {
    if(dataPath.equals(nodePath)) {
     System.out.println("節(jié)點(diǎn)：" + dataPath + "被刪除了！");
    }
   }
   
   @Override
   public void handleDataChange(String dataPath, Object data) throws Exception {
    if(dataPath.equals(nodePath)) {
     System.out.println("節(jié)點(diǎn)：" + dataPath + ", 數(shù)據(jù)：" + data + " - 更新");
     commonConfig = (CommonConfig) data;
    }
   }
  });
 }
 
}

下面啟動Main函數(shù)

配置管理服務(wù)啟動

public static void main(String[] args) throws InterruptedException {
  SpringApplication.run(ZookeeperApiDemoApplication.class, args);
  
  ZkConfigMng zkConfigMng = new ZkConfigMng();
  zkConfigMng.initConfig(null);
  zkConfigMng.syncConfigToZookeeper();
  TimeUnit.SECONDS.sleep(10);
  
  // 修改值
  zkConfigMng.update(new CommonConfig("jdbc:mysql://192.168.1.122:3306/mydata?useUnicode=true&characterEncoding=utf-8",
    "root", "wxh", "com.mysql.jdbc.Driver"));
 }
}

客戶端啟動：

public static void main(String[] args) throws InterruptedException {
  SpringApplication.run(ZookeeperApiDemoApplication.class, args);
 
  ExecutorService executorService = Executors.newFixedThreadPool(3);
  // 模擬多個客戶端獲取配置
  executorService.submit(new ZkConfigClient());
  executorService.submit(new ZkConfigClient());
  executorService.submit(new ZkConfigClient());
 }
}

2.2.分布式鎖

在我們?nèi)粘５拈_發(fā)中，如果是單個進(jìn)程中對共享資源的訪問，我們只需要用synchronized或者lock就能實(shí)現(xiàn)互斥操作。但是對于跨進(jìn)程、跨主機(jī)、跨網(wǎng)絡(luò)的共享資源似乎就無能為力了。

另外，關(guān)注Java知音公眾號，回復(fù)“后端面試”，送你一份面試題寶典！

2.1.1.思路

• 首先zookeeper中我們可以創(chuàng)建一個/distributed_lock持久化節(jié)點(diǎn)
• 然后再在/distributed_lock節(jié)點(diǎn)下創(chuàng)建自己的臨時順序節(jié)點(diǎn)，比如：/distributed_lock/task_00000000008
• 獲取所有的/distributed_lock下的所有子節(jié)點(diǎn)，并排序
• 判讀自己創(chuàng)建的節(jié)點(diǎn)是否最小值（第一位）
• 如果是，則獲取得到鎖，執(zhí)行自己的業(yè)務(wù)邏輯，最后刪除這個臨時節(jié)點(diǎn)。
• 如果不是最小值，則需要監(jiān)聽自己創(chuàng)建節(jié)點(diǎn)前一位節(jié)點(diǎn)的數(shù)據(jù)變化，并阻塞。
• 當(dāng)前一位節(jié)點(diǎn)被刪除時，我們需要通過遞歸來判斷自己創(chuàng)建的節(jié)點(diǎn)是否在是最小的，如果是則執(zhí)行5）；如果不是則執(zhí)行6）（就是遞歸循環(huán)的判斷）

下面是具體的流程圖：

2.1.3.事例

public class DistributedLock {
 
 // 常亮
 static class Constant {
  private static final int SESSION_TIMEOUT = 10000;
  private static final String CONNECTION_STRING = "127.0.0.1:2181";
  private static final String LOCK_NODE = "/distributed_lock";
  private static final String CHILDREN_NODE = "/task_";
 }
 
 private ZkClient zkClient;
 
 public DistributedLock() {
  // 連接到Zookeeper
  zkClient = new ZkClient(new ZkConnection(Constant.CONNECTION_STRING));
  if(!zkClient.exists(Constant.LOCK_NODE)) {
   zkClient.create(Constant.LOCK_NODE, "分布式鎖節(jié)點(diǎn)", CreateMode.PERSISTENT);
  }
 }
 
 public String getLock() {
  try {
   // 1。在Zookeeper指定節(jié)點(diǎn)下創(chuàng)建臨時順序節(jié)點(diǎn)
   String lockName = zkClient.createEphemeralSequential(Constant.LOCK_NODE + Constant.CHILDREN_NODE, "");
   // 嘗試獲取鎖
   acquireLock(lockName);
   return lockName;
  } catch(Exception e) {
   e.printStackTrace();
  }
  
  return null;
 }
 
 /**
  * 獲取鎖
  * @throws InterruptedException 
  */
 public Boolean acquireLock(String lockName) throws InterruptedException {
  // 2.獲取lock節(jié)點(diǎn)下的所有子節(jié)點(diǎn)
  List<String> childrenList = zkClient.getChildren(Constant.LOCK_NODE);
  // 3.對子節(jié)點(diǎn)進(jìn)行排序，獲取最小值
  Collections.sort(childrenList, new Comparator<String>() {
   @Override
   public int compare(String o1, String o2) {
    return Integer.parseInt(o1.split("_")[1]) - Integer.parseInt(o2.split("_")[1]);
   }
   
  });
  // 4.判斷當(dāng)前創(chuàng)建的節(jié)點(diǎn)是否在第一位
  int lockPostion = childrenList.indexOf(lockName.split("/")[lockName.split("/").length - 1]);
  if(lockPostion < 0) {
   // 不存在該節(jié)點(diǎn)
   throw new ZkNodeExistsException("不存在的節(jié)點(diǎn)：" + lockName);
  } else if (lockPostion == 0) {
   // 獲取到鎖
   System.out.println("獲取到鎖：" + lockName);
   return true;
  } else if (lockPostion > 0) {
   // 未獲取到鎖，阻塞
   System.out.println("...... 未獲取到鎖，阻塞等待 。。。。。。");
   // 5.如果未獲取得到鎖，監(jiān)聽當(dāng)前創(chuàng)建的節(jié)點(diǎn)前一位的節(jié)點(diǎn)
   final CountDownLatch latch = new CountDownLatch(1);
   IZkDataListener listener = new IZkDataListener() {
    
    @Override
    public void handleDataDeleted(String dataPath) throws Exception {
     // 6.前一個節(jié)點(diǎn)被刪除,當(dāng)不保證輪到自己
     System.out.println("。。。。。。前一個節(jié)點(diǎn)被刪除  。。。。。。");
     acquireLock(lockName);
     latch.countDown();
    }
    
    @Override
    public void handleDataChange(String dataPath, Object data) throws Exception {
     // 不用理會
    }
   };
   try {
    zkClient.subscribeDataChanges(Constant.LOCK_NODE + "/" + childrenList.get(lockPostion - 1), listener);
    latch.await();
   } finally {
    zkClient.unsubscribeDataChanges(Constant.LOCK_NODE + "/" + childrenList.get(lockPostion - 1), listener);
   }
  }
  return false;
 }
 
 /**
  * 釋放鎖（刪除節(jié)點(diǎn)）
  * 
  * @param lockName
  */
 public void releaseLock(String lockName) {
  zkClient.delete(lockName);
 }
 
 public void closeZkClient() {
  zkClient.close();
 }
}
 
@SpringBootApplication
public class ZookeeperDemoApplication {
 
 public static void main(String[] args) throws InterruptedException {
  SpringApplication.run(ZookeeperDemoApplication.class, args);
  
  DistributedLock lock = new DistributedLock();
  String lockName = lock.getLock();
  /** 
   * 執(zhí)行我們的業(yè)務(wù)邏輯
   */
  if(lockName != null) {
   lock.releaseLock(lockName);
  }
  
  lock.closeZkClient();
 }
}

2.3.分布式隊列

在日常使用中，特別是像生產(chǎn)者消費(fèi)者模式中，經(jīng)常會使用BlockingQueue來充當(dāng)緩沖區(qū)的角色。但是在分布式系統(tǒng)中這種方式就不能使用BlockingQueue來實(shí)現(xiàn)了，但是Zookeeper可以實(shí)現(xiàn)。

2.1.1.思路

• 首先利用Zookeeper中臨時順序節(jié)點(diǎn)的特點(diǎn)
• 當(dāng)生產(chǎn)者創(chuàng)建節(jié)點(diǎn)生產(chǎn)時，需要判斷父節(jié)點(diǎn)下臨時順序子節(jié)點(diǎn)的個數(shù)，如果達(dá)到了上限，則阻塞等待；如果沒有達(dá)到，就創(chuàng)建節(jié)點(diǎn)。
• 當(dāng)消費(fèi)者獲取節(jié)點(diǎn)時，如果父節(jié)點(diǎn)中不存在臨時順序子節(jié)點(diǎn)，則阻塞等待；如果有子節(jié)點(diǎn)，則獲取執(zhí)行自己的業(yè)務(wù)，執(zhí)行完畢后刪除該節(jié)點(diǎn)即可。
• 獲取時獲取最小值，保證FIFO特性。

2.1.2.事例

這個是一個消費(fèi)者對一個生產(chǎn)者，如果是多個消費(fèi)者對多個生產(chǎn)者，對代碼需要調(diào)整。

public interface AppConstant {
 static String ZK_CONNECT_STR = "127.0.0.1:2181";
 static String NODE_PATH = "/mailbox";
 static String CHILD_NODE_PATH = "/mail_";
 static int MAILBOX_SIZE = 10;
}
 
public class MailConsumer implements Runnable, AppConstant{
 
 private ZkClient zkClient;
 private Lock lock;
 private Condition condition;
 
 public MailConsumer() {
  lock = new ReentrantLock();
  condition = lock.newCondition();
  zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
  System.out.println("sucess connected to zookeeper server!");
  // 不存在就創(chuàng)建mailbox節(jié)點(diǎn)
  if(!zkClient.exists(NODE_PATH)) {
   zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
  }
 }
 
 @Override
 public void run() {
  IZkChildListener listener = new IZkChildListener() {  
   @Override
   public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
    System.out.println("Znode["+parentPath + "] size:" + currentChilds.size());
    // 還是要判斷郵箱是否為空
    if(currentChilds.size() > 0) {
     // 喚醒等待的線程
     try {
      lock.lock();
      condition.signal();
     } catch (Exception e) {
      e.printStackTrace();
     } finally {
      lock.unlock();
     }
    }
   }
  };
  // 監(jiān)視子節(jié)點(diǎn)的改變，不用放用while循環(huán)中，監(jiān)聽一次就行了，不需要重復(fù)綁定
  zkClient.subscribeChildChanges(NODE_PATH, listener);
  try {
   //循環(huán)隨機(jī)發(fā)送郵件模擬真是情況
   while(true) {
    // 判斷是否可以發(fā)送郵件
    checkMailReceive();
    // 接受郵件
    List<String> mailList = zkClient.getChildren(NODE_PATH);
    // 如果mailsize==0,也沒有關(guān)系；可以直接循環(huán)獲取就行了
    if(mailList.size() > 0) {
     Collections.sort(mailList, new Comparator<String>() {
      @Override
      public int compare(String o1, String o2) {
       return Integer.parseInt(o1.split("_")[1]) - Integer.parseInt(o2.split("_")[1]);
      }
     });
     // 模擬郵件處理(0-1S)
     TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
     zkClient.delete(NODE_PATH + "/" + mailList.get(0));
     System.out.println("mail has been received:" + NODE_PATH + "/" + mailList.get(0));
    }
   } 
  }catch (Exception e) {
   e.printStackTrace();
  } finally {
   zkClient.unsubscribeChildChanges(NODE_PATH, listener);
  }
 }
 
 private void checkMailReceive() {
  try {
   lock.lock();
   // 判斷郵箱是為空
   List<String> mailList = zkClient.getChildren(NODE_PATH);
   System.out.println("mailbox size: " + mailList.size());
   if(mailList.size() == 0) {
    // 郵箱為空，阻塞消費(fèi)者，直到郵箱有郵件
    System.out.println("mailbox is empty, please wait 。。。");
    condition.await();
    // checkMailReceive();
   }
  } catch (Exception e) {
   e.printStackTrace();
  } finally {
   lock.unlock();
  }
 }
}
 
public class MailProducer implements Runnable, AppConstant{
 
 private ZkClient zkClient;
 private Lock lock;
 private Condition condition;
 
 /**
  * 初始化狀態(tài)
  */
 public MailProducer() {
  lock = new ReentrantLock();
  condition = lock.newCondition();
  zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
  System.out.println("sucess connected to zookeeper server!");
  // 不存在就創(chuàng)建mailbox節(jié)點(diǎn)
  if(!zkClient.exists(NODE_PATH)) {
   zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
  }
 }
 
 @Override
 public void run() {
  IZkChildListener listener = new IZkChildListener() {  
   @Override
   public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
    System.out.println("Znode["+parentPath + "] size:" + currentChilds.size());
    // 還是要判斷郵箱是否已滿
    if(currentChilds.size() < MAILBOX_SIZE) {
     // 喚醒等待的線程
     try {
      lock.lock();
      condition.signal();
     } catch (Exception e) {
      e.printStackTrace();
     } finally {
      lock.unlock();
     }
    }
   }
  };
  // 監(jiān)視子節(jié)點(diǎn)的改變，不用放用while循環(huán)中，監(jiān)聽一次就行了，不需要重復(fù)綁定
  zkClient.subscribeChildChanges(NODE_PATH, listener);
  try {
   //循環(huán)隨機(jī)發(fā)送郵件模擬真是情況
   while(true) {
    // 判斷是否可以發(fā)送郵件
    checkMailSend();
    // 發(fā)送郵件
    String cretePath = zkClient.createEphemeralSequential(NODE_PATH + CHILD_NODE_PATH, "your mail");
    System.out.println("your mail has been send:" + cretePath);
    // 模擬隨機(jī)間隔的發(fā)送郵件(0-10S)
    TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
   } 
  }catch (Exception e) {
   e.printStackTrace();
  } finally {
   zkClient.unsubscribeChildChanges(NODE_PATH, listener);
  }
 }
 
 private void checkMailSend() {
  try {
   lock.lock();
   // 判斷郵箱是否已滿
   List<String> mailList = zkClient.getChildren(NODE_PATH);
   System.out.println("mailbox size: " + mailList.size());
   if(mailList.size() >= MAILBOX_SIZE) {
    // 郵箱已滿，阻塞生產(chǎn)者，直到郵箱有空間
    System.out.println("mailbox is full, please wait 。。。");
    condition.await();
    checkMailSend();
   }
  } catch (Exception e) {
   e.printStackTrace();
  } finally {
   lock.unlock();
  }
 }
}

2.4.均衡負(fù)載

首先我們需要簡單的理解分布式和集群，通俗點(diǎn)說：分布式就是將一個系統(tǒng)拆分到多個獨(dú)立運(yùn)行的應(yīng)用中（有可能在同一臺主機(jī)也有可能在不同的主機(jī)上），集群就是將單個獨(dú)立的應(yīng)用復(fù)制多分放在不同的主機(jī)上來減輕服務(wù)器的壓力。

而Zookeeper不僅僅可以作為分布式集群的服務(wù)注冊調(diào)度中心（例如dubbo），也可以實(shí)現(xiàn)集群的負(fù)載均衡。

另外，關(guān)注Java知音公眾號，回復(fù)“后端面試”，送你一份面試題寶典！

2.4.1.思路

首先我們要理解，如果是一個集群，那么他就會有多臺主機(jī)。所以，他在Zookeeper中信息的存在應(yīng)該是如下所示：

如上的結(jié)構(gòu)，當(dāng)服務(wù)調(diào)用方調(diào)用服務(wù)時，就可以根據(jù)特定的均衡負(fù)載算法來實(shí)現(xiàn)對服務(wù)的調(diào)用（調(diào)用前需要監(jiān)聽/service/serviceXXX節(jié)點(diǎn)，以更新列表數(shù)據(jù)）

2.4.2.事例

/**
 * 服務(wù)提供者
 * 
 * @author Administrator
 *
 */
public class ServiceProvider {
 // 靜態(tài)常量
 static String ZK_CONNECT_STR = "127.0.0.1:2181";
 static String NODE_PATH = "/service";
 static String SERIVCE_NAME = "/myService";
 
 private ZkClient zkClient;
 
 public ServiceProvider() {
  zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
  System.out.println("sucess connected to zookeeper server!");
  // 不存在就創(chuàng)建NODE_PATH節(jié)點(diǎn)
  if(!zkClient.exists(NODE_PATH)) {
   zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
  }
 }
 
 public void registryService(String localIp, Object obj) {
  if(!zkClient.exists(NODE_PATH + SERIVCE_NAME)) {
   zkClient.create(NODE_PATH + SERIVCE_NAME, "provider services list", CreateMode.PERSISTENT);
  }
  // 對自己的服務(wù)進(jìn)行注冊
  zkClient.createEphemeral(NODE_PATH + SERIVCE_NAME + "/" + localIp, obj);
  System.out.println("注冊成功！[" + localIp + "]");
 } 
}
 
/**
 * 消費(fèi)者，通過某種均衡負(fù)載算法選擇某一個提供者
 * 
 * @author Administrator
 *
 */
public class ServiceConsumer {
 // 靜態(tài)常量
 static String ZK_CONNECT_STR = "127.0.0.1:2181";
 static String NODE_PATH = "/service";
 static String SERIVCE_NAME = "/myService";
 
 private List<String> serviceList = new ArrayList<String>();
 
 private ZkClient zkClient;
 
 public ServiceConsumer() {
  zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
  System.out.println("sucess connected to zookeeper server!");
  // 不存在就創(chuàng)建NODE_PATH節(jié)點(diǎn)
  if(!zkClient.exists(NODE_PATH)) {
   zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
  }
 }
 
 /**
  * 訂閱服務(wù)
  */
 public void subscribeSerivce() {
  serviceList = zkClient.getChildren(NODE_PATH + SERIVCE_NAME);
  zkClient.subscribeChildChanges(NODE_PATH + SERIVCE_NAME, new IZkChildListener() {
   @Override
   public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
    serviceList = currentChilds;
   }
  });
 }
 
 /**
  * 模擬調(diào)用服務(wù)
  */
 public void consume() {
  //負(fù)載均衡算法獲取某臺機(jī)器調(diào)用服務(wù)
  int index = new Random().nextInt(serviceList.size());
  System.out.println("調(diào)用[" + NODE_PATH + SERIVCE_NAME + "]服務(wù)：" + serviceList.get(index));
 }
}

3.總結(jié)

Zookeeper是一個功能非常強(qiáng)大的應(yīng)用，除了上面幾種應(yīng)用外，還有命名服務(wù)、分布式協(xié)調(diào)通知等也是常用的場景。

如有文章對你有幫助，
歡迎關(guān)注??、點(diǎn)贊??、轉(zhuǎn)發(fā)??！

推薦， Java面試題庫，詳情點(diǎn)擊：
牛逼！又發(fā)現(xiàn)了一款牛逼的Java面試題庫，史上最強(qiáng)！

點(diǎn)擊文末“閱讀原文”可直達(dá)