ZooKeeper編程01--RMI服務的多服務器管理

服務器端與客戶端都要用到：

public interface ZkInfo {String ZK_CONNECTION_STRING = "192.168.1.201:2181,192.168.1.202:2181,192.168.1.203:2181";int ZK_SESSION_TIMEOUT = 5000;String ZK_REGISTRY_PATH = "/registry";String ZK_PROVIDER_PATH = ZK_REGISTRY_PATH + "/provider";
}

import java.rmi.Remote;
import java.rmi.RemoteException;public interface MyService extends Remote {String showInfo(String name) throws RemoteException;
}

服務器端

import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.Calendar;public class MyServiceImpl extends UnicastRemoteObject implements MyService{/*** */private static final long serialVersionUID = 8329425460319055273L;protected MyServiceImpl() throws RemoteException {}public String showInfo(String name) throws RemoteException {String h = "";try {InetAddress host = Inet4Address.getLocalHost();h = host.getHostAddress();} catch (UnknownHostException e) {// TODO Auto-generated catch blocke.printStackTrace();}Calendar c = Calendar.getInstance();return h + " " + c.getTime().toString() + name;}}


import java.io.IOException;
import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.util.concurrent.CountDownLatch;import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class ServiceProvider {private static final Logger LOGGER = LoggerFactory.getLogger(ServiceProvider.class);//用於等待SyncConnected事件觸發後繼續執行當前進程private CountDownLatch latch = new CountDownLatch(1);//發佈RMI服務並註冊RMI地址到ZooKeeper中public void publish(Remote remote, String host, int port){String url = publishService(remote, host, port);//發佈RMI服務並返回RMI地址if (url != null){ZooKeeper zk = connectServer();//連接ZooKeeper服務器並獲取ZooKeeper對象if (zk != null){createNode(zk, url); //創建ZNode並將RMI地址放入ZNode上}}}//發佈RMI服務private String publishService(Remote remote, String host, int port){String url = null;try{url = String.format("rmi://%s:%d/%s", host, port, remote.getClass().getName());LocateRegistry.createRegistry(port);Naming.rebind(url, remote);LOGGER.debug("publish rmi service(url: {})", url);} catch(RemoteException | MalformedURLException e) {LOGGER.error("", e);}return url;}//連接ZooKeeper服務器private ZooKeeper connectServer(){ZooKeeper zk = null;try{zk = new ZooKeeper(ZkInfo.ZK_CONNECTION_STRING, ZkInfo.ZK_SESSION_TIMEOUT, new Watcher(){@Overridepublic void process(WatchedEvent event){if (event.getState() == Event.KeeperState.SyncConnected){latch.countDown();//喚醒當前正在執行的線程}}});if (zk.exists(ZkInfo.ZK_REGISTRY_PATH, false) == null)zk.create(ZkInfo.ZK_REGISTRY_PATH, " ".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);latch.await(); //使當前進程處於等待狀態} catch (IOException | InterruptedException | KeeperException e){LOGGER.error("", e);}return zk;}//創建ZNodeprivate void createNode(ZooKeeper zk, String url){try{byte[] data = url.getBytes();//創建一個臨時性且有序的ZNodeString path = zk.create(ZkInfo.ZK_PROVIDER_PATH, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);LOGGER.debug("create zookeeper node ({} => {})", path, url);} catch (KeeperException | InterruptedException e){LOGGER.error("", e);}}
}

//ZooKeeper版本
public class Server {public static void main(String[] args) throws Exception{if (args.length != 2){System.err.println("please using command: java Server <rmi_host> <rmi_port>");System.exit(-1);}String host = args[0];int port = Integer.parseInt(args[1]);ServiceProvider provider = new ServiceProvider();MyServiceImpl myService = new MyServiceImpl();provider.publish(myService, host, port);Thread.sleep(Long.MAX_VALUE);}
}

import java.rmi.Naming;
import java.rmi.registry.LocateRegistry;//無ZooKeeper版本
public class RmiServer {public static void main(String[] args) throws Exception{int port = 1099;String url = "rmi://localhost:1099/MyServiceImpl";LocateRegistry.createRegistry(port);Naming.rebind(url, new MyServiceImpl());}
}

客戶端

import java.io.IOException;
import java.net.MalformedURLException;
import java.rmi.ConnectException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class ServiceConsumer {private static final Logger LOGGER = LoggerFactory.getLogger(ServiceConsumer.class);//用於等待SyncConnected事件觸發後繼續執行當前進程private CountDownLatch latch = new CountDownLatch(1);// 定义一个 volatile 成员变量，用于保存最新的 RMI 地址（考虑到该变量或许会被其它线程所修改，一旦修改后，该变量的值会影响到所有线程）// volatile让变量每次在使用的时候，都从主存中取。而不是从各个线程的“工作内存”。// volatile具有synchronized关键字的“可见性”，但是没有synchronized关键字的“并发正确性”，也就是说不保证线程执行的有序性。// 也就是说，volatile变量对于每次使用，线程都能得到当前volatile变量的最新值。但是volatile变量并不保证并发的正确性。private volatile List<String> urlList = new ArrayList<>();//連接ZooKeeper服務器private ZooKeeper connectServer(){ZooKeeper zk = null;try {zk = new ZooKeeper(ZkInfo.ZK_CONNECTION_STRING, ZkInfo.ZK_SESSION_TIMEOUT, new Watcher(){@Overridepublic void process(WatchedEvent event){if (event.getState() == Event.KeeperState.SyncConnected){latch.countDown();//喚醒當前正在執行的線程}}});latch.await();//使當前線程處於等待狀態} catch (IOException | InterruptedException e){LOGGER.error("", e);}return zk;}//觀察/registry節點下所有子節點是否有變化private void watchNode(final ZooKeeper zk){try{List<String> nodeList = zk.getChildren(ZkInfo.ZK_REGISTRY_PATH, new Watcher(){@Overridepublic void process(WatchedEvent event){if (event.getType() == Event.EventType.NodeChildrenChanged){watchNode(zk);//若子節點有變化，則重新調用該方法（爲了獲取最新子節點中的數據）}}});List<String> dataList = new ArrayList<>();//用於存放/registry所有子節點中的 數據for (String node : nodeList){byte[] data = zk.getData(ZkInfo.ZK_REGISTRY_PATH + "/" + node, false, null); //獲取/registry的子節點中的數據dataList.add(new String(data));}LOGGER.debug("node data: {}", dataList);urlList = dataList; // 更新最新的 RMI 地址} catch (KeeperException | InterruptedException e){LOGGER.error("", e);}}//構造器public ServiceConsumer(){ZooKeeper zk = connectServer();//連接ZooKeeper服務器並獲取ZooKeeper對象if (zk != null){watchNode(zk);//觀察/registry節點的所有子節點並更新urlList成員變量}}//在JNDI中查找RMI遠程服務對象@SuppressWarnings("unchecked")private <T> T lookupService(String url){T remote = null;try{remote = (T) Naming.lookup(url);} catch (NotBoundException | MalformedURLException | RemoteException e){if ( e instanceof ConnectException){//若連接中斷，則使用urlList中第一個RMI地址來查找(這是一種簡單的重試方式，確保不會拋出異常)LOGGER.error("ConnectException -> url: {}", url);if (urlList.size() != 0){url = urlList.get(0);return lookupService(url);}}LOGGER.error("", e);}return remote;}//查找RMI服務public <T extends Remote> T lookup(){T service = null;int size = urlList.size();if (size > 0){String url;if (size == 1){url = urlList.get(0); //若urlList中只有一個元素，則直接獲取該元素LOGGER.debug("using only url: {}", url);} else {url = urlList.get(ThreadLocalRandom.current().nextInt(size));//若urlList中存在多個元素，則隨機獲取一個元素LOGGER.debug("using random url: {}", url);System.out.println(url);}service = lookupService(url);}return service;}
}

//含ZooKeeper版本
public class Client {public static void main(String[] args) throws Exception{ServiceConsumer consumer = new ServiceConsumer();//zookeeper測試while (true){MyService myService = consumer.lookup();String result = myService.showInfo("test");System.out.println(result);Thread.sleep(3000);}}
}

import java.rmi.Naming;//無ZooKeeper版本
public class RmiClient {public static void main(String[] args) throws Exception{String url = "rmi://localhost:1099/MyServiceImpl";MyService myService = (MyService)Naming.lookup(url);String result = myService.showInfo("jinzhao");System.out.println(result);}
}

log4j.properties

# Define some default values that can be overridden by system properties
zookeeper.root.logger=INFO, CONSOLE
zookeeper.console.threshold=INFO
zookeeper.log.dir=.
zookeeper.log.file=zookeeper.log
zookeeper.log.threshold=DEBUG
zookeeper.tracelog.dir=.
zookeeper.tracelog.file=zookeeper_trace.log#
# ZooKeeper Logging Configuration
## Format is "<default threshold> (, <appender>)+# DEFAULT: console appender only
log4j.rootLogger=${zookeeper.root.logger}# Example with rolling log file
#log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE# Example with rolling log file and tracing
#log4j.rootLogger=TRACE, CONSOLE, ROLLINGFILE, TRACEFILE#
# Log INFO level and above messages to the console
#
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Threshold=${zookeeper.console.threshold}
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n#
# Add ROLLINGFILE to rootLogger to get log file output
#    Log DEBUG level and above messages to a log file
log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold}
log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file}# Max log file size of 10MB
log4j.appender.ROLLINGFILE.MaxFileSize=10MB
# uncomment the next line to limit number of backup files
#log4j.appender.ROLLINGFILE.MaxBackupIndex=10log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n#
# Add TRACEFILE to rootLogger to get log file output
#    Log DEBUG level and above messages to a log file
log4j.appender.TRACEFILE=org.apache.log4j.FileAppender
log4j.appender.TRACEFILE.Threshold=TRACE
log4j.appender.TRACEFILE.File=${zookeeper.tracelog.dir}/${zookeeper.tracelog.file}log4j.appender.TRACEFILE.layout=org.apache.log4j.PatternLayout
### Notice we are including log4j's NDC here (%x)
log4j.appender.TRACEFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L][%x] - %m%n

ZooKeeper編程01--RMI服務的多服務器管理相关推荐

ZooKeeper編程02--多線程的分佈式鎖
面向過程版: package distributedLockProcess;import java.util.Collections; import java.util.List; import ja ...
php 服務器連接,cocos2d-x網絡編程連接php服務器筆記4
VS工程部分----網絡編程本節會把最終實現代碼和資源放在文章最未提供各位下載學習. 本節我們開始重頭戲聯網功能的開發,我用的是cocos2d-x綁定的curl庫,這個curl據說很火,雖然我本人了 ...
iOS 並行編程初步
原文链接 : iOS Concurrency: Getting Started with NSOperation and Dispatch Queues 原文作者 : Ghareeb Hossam 译 ...
[轉載]Google C++編程風格指南（四）：智能指針和其他C++特性
[轉載]Google C++編程風格指南(四):智能指針和其他C++特性 @ 複雜的人生,複雜的自己. :: 痞客邦 PIXNET :: [轉載]Google C++編程風格指南(四):智能指針和其他 ...
Linux下的多线程編程
Linux下的多線程編程本文出自:http://www.china-pub.com 引言線程(thread)技術早在60年代就被提出,但真正應用多線程到操作系統中去,是在80年代中期,solari ...
中國IT從業人員如此之多，為什麼沒有流行世界的核心技術呢？例如，操作系統，編程語言，數據庫等...
工业革命与文艺复兴 18世紀中葉,英國人瓦特改良蒸汽機之後,一系列技術革命引起了從手工勞動向動力機器生產轉變的重大飛躍.隨後傳播到英格蘭到整個歐洲大陸,19世紀傳播到北美地區.工業革命的基礎,是物理和 ...
java实现鼠标宏编程_我應該如何編程高級java遊戲中的鼠標/鍵輸入？
我是一名自學成才的程序員,所以我不知道正確的做事方式.我製作了諸如小行星和蛇之類的簡單遊戲,但在這些遊戲中,您可以輕鬆修改鍵事件功能中的變量.這裏是我在我的簡單的小行星遊戲做到了:我應該如何編程高級j ...
java中break，continue，標籤實現goto效果（編程思想）
goto 編程語言中一開始就有goto關鍵詞了.事實上,goto起源於彙編語言的進程控制:"若條件A成立,則調到這裏:否則跳到那裏". goto語句時在源碼級別上的跳轉,這導致了其 ...
linux多线程编程和linux 2.6下的nptl,Linux多線程編程和Linux 2.6下的NPTL
這幾天由於工作需要,琢磨了一下Linux下的多線程的相關資料.Linux下最常用的多線程支持庫為Pthread庫,它是glibc庫的組成部分.但是關於Pthread的說明文檔非常缺乏,特別是對POSI ...

ZooKeeper編程01--RMI服務的多服務器管理

ZooKeeper編程01--RMI服務的多服務器管理相关推荐

最新文章

热门文章