Java - zookeeper 服务注册发现-白红宇

Java - zookeeper 服务注册发现

阅读量：4202 次

发布时间：2019-05-26

本文共 16250 字，大约阅读时间需要 54 分钟。

网上实际一些的案例严重缺乏，真心不易出此稿~~

1、zookeeper 服务注册发现模型

流程：1）注册，2）发现：监听、负载均衡、故障检测、变更通知，3）调用。

From http://blog.cloudera.com/blog/2014/03/zookeeper-resilience-at-pinterest/

From http://www.techweb.com.cn/network/hardware/2015-12-25/2246973.shtml

From http://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.3.2/bk_hadoop-ha/content/ha-hs2-service-discovery.html

2、单机代码

虚拟机准备：1）启动zookeeper，2）启动多个服务（如ElasticSearch, Tomcat之类的），3）一个服务（如tomcat、elasticsearch）对应多个instance（如节点1、节点2）

流程：

1）准备：CuratorFramework（可选择getClient或newClient方法），ServiceInstance；

2）注册：ServiceDiscovery (注册有两种方式：registerService/thisInstance)；

3.1）负载均衡：ServiceProvider providerStrategy()

3.2）启动监听：ServiceProvider start()，一个provider对应一个服务；

4）通知：ServiceProvider getInstance()

5）调用: 有ip、port，想怎么用就怎么用吧~

import java.io.Closeable;import java.util.Iterator;import java.util.List;import java.util.Map;import java.util.UUID;import org.apache.curator.RetryPolicy;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.curator.retry.RetryNTimes;import org.apache.curator.utils.CloseableUtils;import org.apache.curator.x.discovery.ServiceDiscovery;import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;import org.apache.curator.x.discovery.ServiceInstance;import org.apache.curator.x.discovery.ServiceProvider;import org.apache.curator.x.discovery.UriSpec;import org.apache.curator.x.discovery.details.JsonInstanceSerializer;import org.apache.curator.x.discovery.strategies.RandomStrategy;import org.apache.zookeeper.WatchedEvent;import org.apache.zookeeper.Watcher;import org.apache.zookeeper.ZooKeeper;import com.google.common.collect.Lists;import com.google.common.collect.Maps;/* * http://www.techweb.com.cn/network/hardware/2015-12-25/2246973.shtml * http://blog.csdn.net/hxpjava1/article/details/8612228 * http://www.tuicool.com/articles/F3Avue * http://www.bkjia.com/Javabc/879495.html * http://curator.apache.org/apidocs/index.html * http://blog.csdn.net/john_f_lau/article/details/50660195  * * http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.curator%22 * share one CuratorFramework per ZooKeeper cluster * By default ServiceProvider uses Round-robin ProviderStrategy */public class ZkTest01 {		private static String basePath = "/es_path";	private static List
   
     closeableList = Lists.newArrayList();		// Register worker 1	private static ServiceDiscovery
    
      registerInZookeeper(CuratorFramework client, 			ServiceInstance
     
       instance) throws Exception {		ServiceDiscovery
      
        serviceDiscovery;		JsonInstanceSerializer
       
         serializer = new JsonInstanceSerializer
        
         ( Void.class); //(3) serviceDiscovery = ServiceDiscoveryBuilder .builder(Void.class) .client(client) .serializer(serializer) .basePath(basePath) //.thisInstance(instance) //注册方式2 .build(); serviceDiscovery.start(); //(4) cd 
         
          /bin, ./zkCli.sh return serviceDiscovery; } // Register worker 2 private static ServiceDiscovery
          
            registerInZookeeper2(CuratorFramework client, ServiceInstance
           
             instance) throws Exception { ServiceDiscovery
            
              serviceDiscovery; JsonInstanceSerializer
             
               serializer = new JsonInstanceSerializer
              
               ( Void.class); //(3) serviceDiscovery = ServiceDiscoveryBuilder .builder(Void.class) .client(client) .serializer(serializer) .basePath(basePath) //.thisInstance(instance) //注册方式2 .build(); serviceDiscovery.start(); //(4) cd 
               
                /bin, ./zkCli.sh return serviceDiscovery; } public static ServiceInstance
                
                  getInstanceByName(ServiceDiscovery
                 
                   serviceDiscovery, String serviceName) throws Exception { //(5) 非注册方式，do not register in ZooKeeper, get instance from ServiceProvider ServiceProvider
                  
                    provider = serviceDiscovery.serviceProviderBuilder(). serviceName(serviceName). providerStrategy(new RandomStrategy
                   
                    ()) .build(); provider.start(); closeableList.add(provider); return provider.getInstance(); } public static synchronized void close(){ for (Closeable closeable : closeableList) { CloseableUtils.closeQuietly(closeable); } } public static CuratorFramework getClient() { RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); CuratorFramework client = CuratorFrameworkFactory.builder() .connectString("127.0.0.1:2181") .sessionTimeoutMs(5000) .connectionTimeoutMs(10000) .retryPolicy(retryPolicy) .namespace("text").build(); client.start(); return client; } public static void main(String[] args) throws Exception { // (1) CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("centos1:2182", new ExponentialBackoffRetry(1000, 1)); curatorFramework.start(); // (2) ServiceInstance
                    
                      instance1 = ServiceInstance.
                     
                      builder() .name("es1") .port(8080) .address("centos1") //address不写的话，会取本地ip //.payload(new Void(UUID.ra ndomUUID().toString(),"centos1",9200)) .uriSpec(new UriSpec("{scheme}://{address}:{port}")) .build(); ServiceDiscovery
                      
                        serviceDiscovery = registerInZookeeper2(curatorFramework, instance1); Iterator
                       
                         ItrServices = serviceDiscovery.queryForNames().iterator(); while(ItrServices.hasNext()) { System.out.println("ItrServices~~~" + ItrServices.next()); } serviceDiscovery.registerService(instance1); //注册方式1 Iterator ItrInstances = serviceDiscovery.queryForInstances("es1").iterator(); while(ItrInstances.hasNext()) { System.out.println("ItrInstances~~~" + ItrInstances.next()); } ServiceInstance
                        
                          instance_loopup = getInstanceByName(serviceDiscovery,"es1"); System.out.println("address~~~" + instance_loopup.buildUriSpec()); System.out.println("payload~~~" + instance_loopup.getPayload()); //System.out.println("listenAddr~~~" + instance_loopup.getPayload().getListenAddress()); close(); CloseableUtils.closeQuietly(curatorFramework); } /* //register ServiceDiscovery = ServiceDiscoveryBuilder.builder.thisInstance(serviceInstance) start() //lookup (by load balancer) ServiceProvider = serviceDiscovery.serviceProviderBuilder().providerStrategy start() //invoke service ------- CuratorFramework = CuratorFrameworkFactory.newClient / builder start() ServiceInstance = ServiceInstance.builder */}

3、单机负载均衡测试

多个服务是否按照负载均衡策略进行访问，需要利用REST http方式进行测试。单机测试如下：

ZkTest02.java

import java.io.BufferedReader;import java.io.IOException;import java.io.InputStream;import java.io.InputStreamReader;import java.io.OutputStream;import java.net.InetSocketAddress;import java.util.concurrent.Executors;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.curator.x.discovery.ServiceDiscovery;import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;import org.apache.curator.x.discovery.ServiceInstance;import org.apache.curator.x.discovery.ServiceProvider;import org.apache.curator.x.discovery.UriSpec;import org.apache.curator.x.discovery.details.JsonInstanceSerializer;import org.apache.curator.x.discovery.strategies.RandomStrategy;import org.apache.curator.x.discovery.strategies.RoundRobinStrategy;import org.slf4j.LoggerFactory;import com.sun.net.httpserver.HttpExchange;import com.sun.net.httpserver.HttpHandler;import com.sun.net.httpserver.HttpServer;import com.sun.net.httpserver.spi.HttpServerProvider;public class ZkTest02 {		private static String basePath = "/es_path";	private static ServiceDiscovery
   
     serviceDiscovery;	private static ServiceProvider
    
      provider;		public static void main(String[] args) throws IOException {        httpserverService();    }		public static CuratorFramework getClient() {		CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("centos1:2182", new ExponentialBackoffRetry(1000, 1));		curatorFramework.start();		return curatorFramework;	}	public static ServiceDiscovery
     
       getServiceDiscovery(CuratorFramework client) throws Exception {		JsonInstanceSerializer
      
        serializer = new JsonInstanceSerializer
       
        (				Void.class);		ServiceDiscovery
        
          serviceDiscovery = ServiceDiscoveryBuilder .builder(Void.class) .client(client) .serializer(serializer) .basePath(basePath) .build(); serviceDiscovery.start(); return serviceDiscovery; } public static ServiceInstance
         
           getInstance1() throws Exception { ServiceInstance
          
            instance1 = ServiceInstance.
           
            builder() .name("es1") .port(8080) .address("centos1") .uriSpec(new UriSpec("http://centos1:8080")) .build(); return instance1; } public static ServiceInstance
            
              getInstance2() throws Exception { ServiceInstance
             
               instance1 = ServiceInstance.
              
               builder() .name("es1") .port(9200) .address("centos1") .uriSpec(new UriSpec("http://centos1:9200")) .build(); return instance1; } public static ServiceProvider
               
                 serviceProvider(ServiceDiscovery
                
                  serviceDiscovery, String serviceName) throws Exception { ServiceProvider
                 
                   provider = serviceDiscovery.serviceProviderBuilder(). serviceName(serviceName). providerStrategy(new RandomStrategy
                  
                   ()) .build(); provider.start(); return provider; } public static void process(String param) throws Exception { if (null == serviceDiscovery) { LoggerFactory.getLogger(""); serviceDiscovery = getServiceDiscovery(getClient()); serviceDiscovery.registerService(getInstance1()); serviceDiscovery.registerService(getInstance2()); } if (null == provider) { provider = serviceProvider(serviceDiscovery, param); } if (null != provider) { System.out.println("address~~~" + provider.getInstance().getAddress() + ", port~~~" + provider.getInstance().getPort());// provider.close(); } } //启动服务，监听来自客户端的请求 public static void httpserverService() throws IOException { HttpServerProvider provider = HttpServerProvider.provider(); HttpServer httpserver =provider.createHttpServer(new InetSocketAddress(8080), 100);//监听端口6666,能同时接 受100个请求 httpserver.createContext("/sc/es", new MyHttpHandler()); //httpserver.setExecutor(null); //使用单线程 httpserver.setExecutor(Executors.newFixedThreadPool(5)); httpserver.start(); System.out.println("server started"); } //Http请求处理类 static class MyHttpHandler implements HttpHandler { public void handle(HttpExchange httpExchange) throws IOException { String responseMsg = "ok"; //响应信息 InputStream in = httpExchange.getRequestBody(); //获得输入流 BufferedReader reader = new BufferedReader(new InputStreamReader(in)); String temp = null; while((temp = reader.readLine()) != null) { System.out.println("client request:"+temp); try { process(temp); } catch (Exception e) { e.printStackTrace(); } } httpExchange.sendResponseHeaders(200, responseMsg.length()); //设置响应头属性及响应信息的长度 OutputStream out = httpExchange.getResponseBody(); //获得输出流 out.write(responseMsg.getBytes()); out.flush(); httpExchange.close(); } }}

SpringRestTestClient.java （测试类）

import java.net.URI;import org.springframework.web.client.RestClientException;import org.springframework.web.client.RestTemplate;public class SpringRestTestClient {	public static final String REST_SERVICE_URI = "http://localhost:8080/sc";//	public static final String REST_SERVICE_URI = "http://localhost:6666/myApp";		private static void listAllUsers(){		RestTemplate restTemplate = new RestTemplate();		String param = "es1";		try {//			URI uri = restTemplate.postForLocation(REST_SERVICE_URI+"/es", jsonParam, String.class);//			if (null != uri){//				System.out.println("Location : "+uri.toASCIIString());//			} else {//				System.out.println("uri is null~");//			}			String result = restTemplate.postForObject(REST_SERVICE_URI+"/es", param, String.class);			System.out.println("es-result~" + result);		} catch (RestClientException e) {			e.printStackTrace();		}	}		public static void main(String[] args) {		listAllUsers();	}}

先启动ZkTest02，然后连续运行SpringRestTestClient 几次。ZkTest02那边的结果（sys out）：

client request:es1address~~~centos1, port~~~8080client request:es1address~~~centos1, port~~~9200client request:es1address~~~centos1, port~~~9200client request:es1address~~~centos1, port~~~8080client request:es1address~~~centos1, port~~~9200

4、Spring下负载均衡测试代码

另外，也可以用Spring controller来做测试，不过遇到一些引用包的问题，应该与spring相关，由于前面的方法已经满足PoC需求，暂时没有再试下去，有兴趣也可以参考以下代码：

package com.test;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.curator.x.discovery.ServiceDiscovery;import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;import org.apache.curator.x.discovery.ServiceInstance;import org.apache.curator.x.discovery.ServiceProvider;import org.apache.curator.x.discovery.UriSpec;import org.apache.curator.x.discovery.details.JsonInstanceSerializer;import org.apache.curator.x.discovery.strategies.RoundRobinStrategy;import org.slf4j.LoggerFactory;import org.springframework.web.bind.annotation.RequestBody;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestMethod;import org.springframework.web.bind.annotation.RestController;import org.springframework.web.util.UriComponentsBuilder;@RestController@RequestMapping("/zk")public class ZkController {		private static String basePath = "/es_path";	private ServiceDiscovery
   
     serviceDiscovery;		@RequestMapping(value = "/es", method = RequestMethod.POST)    public String postUser(@RequestBody String param, UriComponentsBuilder ucBuilder) throws Exception {    	if (null == serviceDiscovery) {    		LoggerFactory.getLogger("");	    	serviceDiscovery = getServiceDiscovery(getClient());	    	serviceDiscovery.registerService(getInstance1());	    	serviceDiscovery.registerService(getInstance2());    	}    	System.out.println("param~~~" + param);    	ServiceProvider
    
      provider = serviceProvider(serviceDiscovery, param);    	if (null != provider) {	    	System.out.println("address~~~" + provider.getInstance().getAddress() + 	    			", port~~~" + provider.getInstance().getPort());	    	provider.close();    	}    	return param;    }		public CuratorFramework getClient() {		CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("centos1:2182", new ExponentialBackoffRetry(1000, 1));		curatorFramework.start();		return curatorFramework;	}	public ServiceDiscovery
     
       getServiceDiscovery(CuratorFramework client) throws Exception {		JsonInstanceSerializer
      
        serializer = new JsonInstanceSerializer
       
        (				Void.class);		ServiceDiscovery
        
          serviceDiscovery = ServiceDiscoveryBuilder .builder(Void.class) .client(client) .serializer(serializer) .basePath(basePath) .build(); serviceDiscovery.start(); return serviceDiscovery; } public ServiceInstance
         
           getInstance1() throws Exception { ServiceInstance
          
            instance1 = ServiceInstance.
           
            builder() .name("es1") .port(8080) .address("centos1") .uriSpec(new UriSpec("http://centos1:8080")) .build(); return instance1; } public ServiceInstance
            
              getInstance2() throws Exception { ServiceInstance
             
               instance1 = ServiceInstance.
              
               builder() .name("es1") .port(9200) .address("centos1") .uriSpec(new UriSpec("http://centos1:9200")) .build(); return instance1; } public ServiceProvider
               
                 serviceProvider(ServiceDiscovery
                
                  serviceDiscovery, String serviceName) throws Exception { ServiceProvider
                 
                   provider = serviceDiscovery.serviceProviderBuilder(). serviceName(serviceName). providerStrategy(new RoundRobinStrategy
                  
                   ()) .build(); provider.start(); return provider; }}

上面的代码只是基于zookeeer znode操作，对服务的监测还是需要自己实现的，比如一个provider - war程序服务，可以做一个REST接口出来进行监听。

5、Q&A

Q: curator没有现成的jar包

A: 上http://search.maven.org/

Q: Connection Timeout

A: 检查zookeeper、服务的端口是否打开（如在centos，打开端口命令为/sbin/iptables -I INPUT -p tcp --dport 2181 -j ACCEPT, /etc/init.d/iptables save, service iptables restart）

Q: base path找不到

A: 启动zookeeper, 运行./zkCli.sh，使用zookeeer命令：ls / 看有没有base_path，没有就新建，如上面的例子，put /es_path/es1，ls /es_path/es1

Q: 在浏览器http访问测试时，报不支持post错误或访问为get方式

A: post方式不能用浏览器http直接访问测试，用代码发送post请求或用postman来测试；

Q: 除了Curator，还有其它实现手段吗

A: （1）zoologist（http://npm.taobao.org/package/zoologist，基于Curator REST接口以Node.js实现的包）；（2）Spring Cloud Zookeeper（http://cloud.spring.io/spring-cloud-zookeeper/spring-cloud-zookeeper.html，基于ZK和curator，spring boot上的实现）；（3）dubbo（http://dubbo.io/User+Guide-zh.htm，阿里框架，可以基于zookeeper和curator，并支持可视化交互monitor）。

Q: 有没有其它Sample

A: （1）GitHub - curator-playground-master, curator-disco-master, disco-java；（2）Link: http://www.jianshu.com/p/c0ba6d62dd2e

你可能感兴趣的文章