色婷婷阁,天天操天天爽天天射,免费看AAA片,高潮毛片无遮挡免费看,啊～我是sao货快cao我,校花让我她我爽了好久动漫,成人免费视频观看,狼友视频在线观看

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

2.1.一致性配置管理

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

2.1.1.思路

1. 將公共的配置存放在Zookeeper的節(jié)點中

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

2.1.2.事例

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

配置類：

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 + "}";
??}
}

配置管理中心

1. 獲取本地配置信息

2. 修改配置，并同步

同步配置信息到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é)點不存在!");
??????try?{
????????TimeUnit.SECONDS.sleep(1);
??????} catch?(InterruptedException e) {
????????e.printStackTrace();
??????}
????}
????// 獲取節(jié)點
????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é)點："?+ dataPath + "被刪除了！");
????????}
??????}
??????
??????@Override
??????public?void?handleDataChange(String dataPath, Object data)?throws?Exception {
????????if(dataPath.equals(nodePath)) {
??????????System.out.println("節(jié)點："?+ 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ā)中，如果是單個進程中對共享資源的訪問，我們只需要用synchronized或者lock就能實現(xiàn)互斥操作。但是對于跨進程、跨主機、跨網(wǎng)絡(luò)的共享資源似乎就無能為力了。

2.1.1.思路

1. 首先zookeeper中我們可以創(chuàng)建一個/distributed_lock持久化節(jié)點

2. 然后再在/distributed_lock節(jié)點下創(chuàng)建自己的臨時順序節(jié)點，比如：/distributed_lock/task_00000000008

3. 獲取所有的/distributed_lock下的所有子節(jié)點，并排序

4. 判讀自己創(chuàng)建的節(jié)點是否最小值（第一位）

5. 如果是，則獲取得到鎖，執(zhí)行自己的業(yè)務(wù)邏輯，最后刪除這個臨時節(jié)點。

6. 如果不是最小值，則需要監(jiān)聽自己創(chuàng)建節(jié)點前一位節(jié)點的數(shù)據(jù)變化，并阻塞。

7. 當前一位節(jié)點被刪除時，我們需要通過遞歸來判斷自己創(chuàng)建的節(jié)點是否在是最小的，如果是則執(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é)點", CreateMode.PERSISTENT);
????}
??}
??
??public?String getLock()?{
????try?{
??????// 1。在Zookeeper指定節(jié)點下創(chuàng)建臨時順序節(jié)點
??????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é)點下的所有子節(jié)點
????List childrenList = zkClient.getChildren(Constant.LOCK_NODE);
????// 3.對子節(jié)點進行排序，獲取最小值
????Collections.sort(childrenList, new?Comparator() {
??????@Override
??????public?int?compare(String o1, String o2)?{
????????return?Integer.parseInt(o1.split("_")[1]) - Integer.parseInt(o2.split("_")[1]);
??????}
??????
????});
????// 4.判斷當前創(chuàng)建的節(jié)點是否在第一位
????int?lockPostion = childrenList.indexOf(lockName.split("/")[lockName.split("/").length - 1]);
????if(lockPostion < 0) {
??????// 不存在該節(jié)點
??????throw?new?ZkNodeExistsException("不存在的節(jié)點："?+ lockName);
????} else?if?(lockPostion == 0) {
??????// 獲取到鎖
??????System.out.println("獲取到鎖："?+ lockName);
??????return?true;
????} else?if?(lockPostion > 0) {
??????// 未獲取到鎖，阻塞
??????System.out.println("...... 未獲取到鎖，阻塞等待 。。。。。。");
??????// 5.如果未獲取得到鎖，監(jiān)聽當前創(chuàng)建的節(jié)點前一位的節(jié)點
??????final?CountDownLatch latch = new?CountDownLatch(1);
??????IZkDataListener listener = new?IZkDataListener() {
????????
????????@Override
????????public?void?handleDataDeleted(String dataPath)?throws?Exception {
??????????// 6.前一個節(jié)點被刪除,當不保證輪到自己
??????????System.out.println("。。。。。。前一個節(jié)點被刪除 。。。。。。");
??????????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é)點）
???* 
???* @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)者消費者模式中，經(jīng)常會使用BlockingQueue來充當緩沖區(qū)的角色。但是在分布式系統(tǒng)中這種方式就不能使用BlockingQueue來實現(xiàn)了，但是Zookeeper可以實現(xiàn)。

2.1.1.思路

1. 首先利用Zookeeper中臨時順序節(jié)點的特點

2. 當生產(chǎn)者創(chuàng)建節(jié)點生產(chǎn)時，需要判斷父節(jié)點下臨時順序子節(jié)點的個數(shù)，如果達到了上限，則阻塞等待；如果沒有達到，就創(chuàng)建節(jié)點。

3. 當消費者獲取節(jié)點時，如果父節(jié)點中不存在臨時順序子節(jié)點，則阻塞等待；如果有子節(jié)點，則獲取執(zhí)行自己的業(yè)務(wù)，執(zhí)行完畢后刪除該節(jié)點即可。

4. 獲取時獲取最小值，保證FIFO特性。

2.1.2.事例

這個是一個消費者對一個生產(chǎn)者，如果是多個消費者對多個生產(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é)點
????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 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é)點的改變，不用放用while循環(huán)中，監(jiān)聽一次就行了，不需要重復(fù)綁定
????zkClient.subscribeChildChanges(NODE_PATH, listener);
????try?{
??????//循環(huán)隨機發(fā)送郵件模擬真是情況
??????while(true) {
????????// 判斷是否可以發(fā)送郵件
????????checkMailReceive();
????????// 接受郵件
????????List mailList = zkClient.getChildren(NODE_PATH);
????????// 如果mailsize==0,也沒有關(guān)系；可以直接循環(huán)獲取就行了
????????if(mailList.size() > 0) {
??????????Collections.sort(mailList, new?Comparator() {
????????????@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 mailList = zkClient.getChildren(NODE_PATH);
??????System.out.println("mailbox size: "?+ mailList.size());
??????if(mailList.size() == 0) {
????????// 郵箱為空，阻塞消費者，直到郵箱有郵件
????????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é)點
????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 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é)點的改變，不用放用while循環(huán)中，監(jiān)聽一次就行了，不需要重復(fù)綁定
????zkClient.subscribeChildChanges(NODE_PATH, listener);
????try?{
??????//循環(huán)隨機發(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);
????????// 模擬隨機間隔的發(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 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.均衡負載

首先我們需要簡單的理解分布式和集群，通俗點說：分布式就是將一個系統(tǒng)拆分到多個獨立運行的應(yīng)用中（有可能在同一臺主機也有可能在不同的主機上），集群就是將單個獨立的應(yīng)用復(fù)制多分放在不同的主機上來減輕服務(wù)器的壓力。而Zookeeper不僅僅可以作為分布式集群的服務(wù)注冊調(diào)度中心（例如dubbo），也可以實現(xiàn)集群的負載均衡。

2.4.1.思路

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

2. 如上的結(jié)構(gòu)，當服務(wù)調(diào)用方調(diào)用服務(wù)時，就可以根據(jù)特定的均衡負載算法來實現(xiàn)對服務(wù)的調(diào)用（調(diào)用前需要監(jiān)聽/service/serviceXXX節(jié)點，以更新列表數(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é)點
????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ù)進行注冊
????zkClient.createEphemeral(NODE_PATH + SERIVCE_NAME + "/"?+ localIp, obj);
????System.out.println("注冊成功！["?+ localIp + "]");
??} 
}
?
/**
?* 消費者，通過某種均衡負載算法選擇某一個提供者
?* 
?* @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 serviceList = new?ArrayList();
??
??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é)點
????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 currentChilds) throws Exception {
????????serviceList = currentChilds;
??????}
????});
??}
??
??/**
???* 模擬調(diào)用服務(wù)
???*/
??public?void?consume() {
????//負載均衡算法獲取某臺機器調(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是一個功能非常強大的應(yīng)用，除了上面幾種應(yīng)用外，還有命名服務(wù)、分布式協(xié)調(diào)通知等也是常用的場景。

【END】

歡迎關(guān)注Java公眾號

Zookeeper 典型應(yīng)用場景場景

2.1.一致性配置管理

2.1.1.思路

2.1.2.事例