经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Zookeeper » 查看文章
zookeeper【3】服务发现
来源：cnblogs　　作者：撸码识途　　时间：2018/11/28 9:58:27　　对本文有异议

服务发现：指对集群中的服务上下线做统一管理，每个工作服务器都可以作为数据的发布方，向集群注册自己的基本信息，而让某些监控服务器作为订阅方，订阅工作服务器的基本信息。当工作服务器的基本信息改变时，如服务上下线、服务器的角色或服务范围变更，那么监控服务器可以得到通知并响应这些变化。

实现代码如下：

  1. import com.alibaba.fastjson.JSON;
  2. import org.I0Itec.zkclient.IZkDataListener;
  3. import org.I0Itec.zkclient.ZkClient;
  4. import org.I0Itec.zkclient.exception.ZkNoNodeException;
  6. /**
  7.  * 代表工作服务器
  8.  */
  9. public class WorkServer {
  11.     private ZkClient zkClient;
  12.     // ZooKeeper
  13.     private String configPath;
  14.     // ZooKeeper集群中servers节点的路径
  15.     private String serversPath;
  16.     // 当前工作服务器的基本信息
  17.     private ServerData serverData;
  18.     // 当前工作服务器的配置信息
  19.     private ServerConfig serverConfig;
  20.     private IZkDataListener dataListener;
  22.     public WorkServer(String configPath, String serversPath,
  23.                       ServerData serverData, ZkClient zkClient, ServerConfig initConfig) {
  24.         this.zkClient = zkClient;
  25.         this.serversPath = serversPath;
  26.         this.configPath = configPath;
  27.         this.serverConfig = initConfig;
  28.         this.serverData = serverData;
  30.         this.dataListener = new IZkDataListener() {
  32.             public void handleDataDeleted(String dataPath) throws Exception {
  34.             }
  36.             public void handleDataChange(String dataPath, Object data)
  37.                     throws Exception {
  38.                 String retJson = new String((byte[])data);
  39.                 ServerConfig serverConfigLocal = (ServerConfig) JSON.parseObject(retJson,ServerConfig.class);
  40.                 updateConfig(serverConfigLocal);
  41.                 System.out.println("new Work server config is:"+serverConfig.toString());
  43.             }
  44.         };
  46.     }
  48.     // 启动服务器
  49.     public void start() {
  50.         System.out.println("work server start...");
  51.         initRunning();
  52.     }
  54.     // 停止服务器
  55.     public void stop() {
  56.         System.out.println("work server stop...");
  57.         zkClient.unsubscribeDataChanges(configPath, dataListener); // 取消监听config节点
  58.     }
  60.     // 服务器初始化
  61.     private void initRunning() {
  62.         registMe(); // 注册自己
  63.         zkClient.subscribeDataChanges(configPath, dataListener); // 订阅config节点的改变事件
  64.     }
  66.     // 启动时向zookeeper注册自己的注册函数
  67.     private void registMe() {
  68.         String mePath = serversPath.concat("/").concat(serverData.getAddress());
  70.         try {
  71.             zkClient.createEphemeral(mePath, JSON.toJSONString(serverData)
  72.                     .getBytes());
  73.         } catch (ZkNoNodeException e) {
  74.             zkClient.createPersistent(serversPath, true);
  75.             registMe();
  76.         }
  77.     }
  79.     // 更新自己的配置信息
  80.     private void updateConfig(ServerConfig serverConfig) {
  81.         this.serverConfig = serverConfig;
  82.     }
  84. }
  1. /**
  2.  * 调度类
  3.  */
  4. public class SubscribeZkClient {
  6.     private static final int  CLIENT_QTY = 5; // Work Server数量
  7.  
  8.     private static final String  ZOOKEEPER_SERVER = "192.168.1.105:2181";
  10.     private static final String  CONFIG_PATH = "/config";
  11.     private static final String  COMMAND_PATH = "/command";
  12.     private static final String  SERVERS_PATH = "/servers";
  14.     public static void main(String[] args) throws Exception {
  16.         List<ZkClient> clients = new ArrayList<ZkClient>();
  17.         List<WorkServer>  workServers = new ArrayList<WorkServer>();
  18.         ManageServer manageServer = null;
  20.         try {
  22.             // 创建一个默认的配置
  23.             ServerConfig initConfig = new ServerConfig();
  24.             initConfig.setDbPwd("123456");
  25.             initConfig.setDbUrl("jdbc:mysql://localhost:3306/mydb");
  26.             initConfig.setDbUser("root");
  28.             // 实例化一个Manage Server
  29.             ZkClient clientManage = new ZkClient(ZOOKEEPER_SERVER, 5000, 5000, new BytesPushThroughSerializer());
  30.             manageServer = new ManageServer(SERVERS_PATH, COMMAND_PATH,CONFIG_PATH,clientManage,initConfig);
  31.             manageServer.start(); // 启动Manage Server
  33.             // 创建指定个数的工作服务器
  34.             for ( int i = 0; i < CLIENT_QTY; ++i ) {
  35.                 ZkClient client = new ZkClient(ZOOKEEPER_SERVER, 5000, 5000, new BytesPushThroughSerializer());
  36.                 clients.add(client);
  37.                 ServerData serverData = new ServerData();
  38.                 serverData.setId(i);
  39.                 serverData.setName("WorkServer#"+i);
  40.                 serverData.setAddress("192.168.1."+i);
  42.                 WorkServer  workServer = new WorkServer(CONFIG_PATH, SERVERS_PATH, serverData, client, initConfig);
  43.                 workServers.add(workServer);
  44.                 workServer.start();    // 启动工作服务器
  46.             }
  48.             System.out.println("敲回车键退出！\n");
  49.             new BufferedReader(new InputStreamReader(System.in)).readLine();
  51.         } finally {
  52.             System.out.println("Shutting down...");
  54.             for ( WorkServer workServer : workServers ) {
  55.                 try {
  56.                     workServer.stop();
  57.                 } catch (Exception e) {
  58.                     e.printStackTrace();
  59.                 }
  60.             }
  62.             for ( ZkClient client : clients ) {
  63.                 try {
  64.                     client.close();
  65.                 } catch (Exception e) {
  66.                     e.printStackTrace();
  67.                 }
  69.             }
  70.         }
  71.     }
  73. }
  1. /**
  2.  * 服务器基本信息
  3.  */
  4. public class ServerData {
  6.     private String address;
  7.     private Integer id;
  8.     private String name;
  10.     public String getAddress() {
  11.         return address;
  12.     }
  13.     public void setAddress(String address) {
  14.         this.address = address;
  15.     }
  16.     public Integer getId() {
  17.         return id;
  18.     }
  19.     public void setId(Integer id) {
  20.         this.id = id;
  21.     }
  22.     public String getName() {
  23.         return name;
  24.     }
  25.     public void setName(String name) {
  26.         this.name = name;
  27.     }
  29.     @Override
  30.     public String toString() {
  31.         return "ServerData [address=" + address + ", id=" + id + ", name="
  32.                 + name + "]";
  33.     }
  35. }
  1. /**
  2.  * 配置信息
  3.  */
  4. public class ServerConfig {
  6.     private String dbUrl;
  7.     private String dbPwd;
  8.     private String dbUser;
  9.     public String getDbUrl() {
  10.         return dbUrl;
  11.     }
  12.     public void setDbUrl(String dbUrl) {
  13.         this.dbUrl = dbUrl;
  14.     }
  15.     public String getDbPwd() {
  16.         return dbPwd;
  17.     }
  18.     public void setDbPwd(String dbPwd) {
  19.         this.dbPwd = dbPwd;
  20.     }
  21.     public String getDbUser() {
  22.         return dbUser;
  23.     }
  24.     public void setDbUser(String dbUser) {
  25.         this.dbUser = dbUser;
  26.     }
  28.     @Override
  29.     public String toString() {
  30.         return "ServerConfig [dbUrl=" + dbUrl + ", dbPwd=" + dbPwd
  31.                 + ", dbUser=" + dbUser + "]";
  32.     }
  34. }
  1. import com.alibaba.fastjson.JSON;
  2. import org.I0Itec.zkclient.IZkChildListener;
  3. import org.I0Itec.zkclient.IZkDataListener;
  4. import org.I0Itec.zkclient.ZkClient;
  5. import org.I0Itec.zkclient.exception.ZkNoNodeException;
  6. import org.I0Itec.zkclient.exception.ZkNodeExistsException;
  8. import java.util.List;
  10. public class ManageServer {
  12.     // zookeeper的servers节点路径
  13.     private String serversPath;
  14.     // zookeeper的command节点路径
  15.     private String commandPath;
  16.     // zookeeper的config节点路径
  17.     private String configPath;
  18.     private ZkClient zkClient;
  19.     private ServerConfig config;
  20.     // 用于监听servers节点的子节点列表的变化
  21.     private IZkChildListener childListener;
  22.     // 用于监听command节点数据内容的变化
  23.     private IZkDataListener dataListener;
  24.     // 工作服务器的列表
  25.     private List<String> workServerList;
  27.     public ManageServer(String serversPath, String commandPath,
  28.                         String configPath, ZkClient zkClient, ServerConfig config) {
  29.         this.serversPath = serversPath;
  30.         this.commandPath = commandPath;
  31.         this.zkClient = zkClient;
  32.         this.config = config;
  33.         this.configPath = configPath;
  34.         this.childListener = new IZkChildListener() {
  36.             public void handleChildChange(String parentPath,
  37.                                           List<String> currentChilds) throws Exception {
  38.                 // TODO Auto-generated method stub
  39.                 workServerList = currentChilds; // 更新内存中工作服务器列表
  41.                 System.out.println("work server list changed, new list is ");
  42.                 execList();
  44.             }
  45.         };
  46.         this.dataListener = new IZkDataListener() {
  48.             public void handleDataDeleted(String dataPath) throws Exception {
  49.                 // TODO Auto-generated method stub
  50.                 // ignore;
  51.             }
  53.             public void handleDataChange(String dataPath, Object data)
  54.                     throws Exception {
  55.                 // TODO Auto-generated method stub
  56.                 String cmd = new String((byte[]) data);
  57.                 System.out.println("cmd:"+cmd);
  58.                 exeCmd(cmd); // 执行命令
  60.             }
  61.         };
  63.     }
  65.     private void initRunning() {
  66.         zkClient.subscribeDataChanges(commandPath, dataListener);
  67.         zkClient.subscribeChildChanges(serversPath, childListener);
  68.     }
  70.     /*
  71.      * 1: list 2: create 3: modify
  72.      */
  73.     private void exeCmd(String cmdType) {
  74.         if ("list".equals(cmdType)) {
  75.             execList();
  77.         } else if ("create".equals(cmdType)) {
  78.             execCreate();
  79.         } else if ("modify".equals(cmdType)) {
  80.             execModify();
  81.         } else {
  82.             System.out.println("error command!" + cmdType);
  83.         }
  85.     }
  87.     // 列出工作服务器列表
  88.     private void execList() {
  89.         System.out.println(workServerList.toString());
  90.     }
  92.     // 创建config节点
  93.     private void execCreate() {
  94.         if (!zkClient.exists(configPath)) {
  95.             try {
  96.                 zkClient.createPersistent(configPath, JSON.toJSONString(config)
  97.                         .getBytes());
  98.             } catch (ZkNodeExistsException e) {
  99.                 zkClient.writeData(configPath, JSON.toJSONString(config)
  100.                         .getBytes()); // config节点已经存在，则写入内容就可以了
  101.             } catch (ZkNoNodeException e) {
  102.                 String parentDir = configPath.substring(0,
  103.                         configPath.lastIndexOf('/'));
  104.                 zkClient.createPersistent(parentDir, true);
  105.                 execCreate();
  106.             }
  107.         }
  108.     }
  110.     // 修改config节点内容
  111.     private void execModify() {
  112.         // 我们随意修改config的一个属性就可以了
  113.         config.setDbUser(config.getDbUser() + "_modify");
  115.         try {
  116.             zkClient.writeData(configPath, JSON.toJSONString(config).getBytes());
  117.         } catch (ZkNoNodeException e) {
  118.             execCreate(); // 写入时config节点还未存在，则创建它
  119.         }
  120.     }
  122.     // 启动工作服务器
  123.     public void start() {
  124.         initRunning();
  125.     }
  127.     // 停止工作服务器
  128.     public void stop() {
  129.         zkClient.unsubscribeChildChanges(serversPath, childListener);
  130.         zkClient.unsubscribeDataChanges(commandPath, dataListener);
  131.     }
  133. }

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Zookeeper热门文章

  • java Zookeeper简述
  • zookeeper-操作与应用场景-《每日五分钟搞定大数据》
  • ZooInspector 连接不到 Zookeeper 的解决方法
  • IDEA zookeeper插件的使用
  • 解决dubbo错误ip及ip乱入问题的方法
  • zookeeper集群搭建
  • 大数据开发phoenix连接hbase流程详解
  • 快来看看 Dubbo 是如何实现服务的动态发现的
  • zookeeper特性与节点说明
  • zookeeper【1】配置管理

    • Zookeeper推荐文章

  • Zookeeper安装部署
  • 得物 ZooKeeper SLA 也可以 99.99%
  • zookeeper源码(06)ZooKeeperServer及子类
  • Dubbo 我手写几行代码，就把通信模式给你解释清楚！
  • Rpc-实现Zookeeper注册中心
  • Dubbo 中 Zookeeper 注册中心原理分析
  • Java&nbsp;zookeeper服务的使用详解
  • 【入门】ZooKeeper 相关概念总结
  • zookeeper特性与节点说明
  • 快来看看 Dubbo 是如何实现服务的动态发现的

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号