Service Discovery with Apache Curator

 

Curator的介绍

  Curator就是Zookeeper的一个客户端工具(不知道Zookeeper的同窗能够到http://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/学习下),封装ZooKeeper client与ZooKeeper server之间的链接处理以及zookeeper的经常使用操做,提供ZooKeeper各类应用场景(recipe, 好比共享锁服务, 集群领导选举机制)的抽象封装。固然还有他看起来很是舒服的Fluent风格的API。 Curator主要从如下几个方面下降了zk使用的复杂性: html

  • 重试机制:提供可插拔的重试机制, 它将给捕获全部可恢复的异常配置一个重试策略, 而且内部也提供了几种标准的重试策略(好比指数补偿).
  • 链接状态监控: Curator初始化以后会一直的对zk链接进行监听, 一旦发现链接状态发生变化, 将做出相应的处理.
  • zk客户端实例管理:Curator对zk客户端到server集群链接进行管理. 并在须要的状况, 重建zk实例, 保证与zk集群的可靠链接 
  • 各类使用场景支持:Curator实现zk支持的大部分使用场景支持(甚至包括zk自身不支持的场景), 这些实现都遵循了zk的最佳实践, 并考虑了各类极端状况.

  Curator经过以上的处理, 让用户专一于自身的业务自己, 而无需花费更多的精力在zk自己.这里咱们介绍的是Curator的Service Discovery模块java

Service Discovery 

  咱们一般在调用服务的时候,须要知道服务的地址,端口,或者其余一些信息,一般状况下,咱们是把他们写到程序里面,可是随着服务愈来愈多,维护起来也愈来愈费劲,更重要的是,因为地址都是在程序中配置的,咱们根本不知道远程的服务是否可用,当咱们增长或者删除服务,咱们又须要到配置文件中配置么? 这时候,Zookeeper帮大忙了,咱们能够把咱们的服务注册到Zookeeper中,建立一个临时节点(当链接断开以后,节点将被删除),存放咱们的服务信息(url,ip,port等信息),把这些临时节点都存放在以serviceName命名的节点下面,这样咱们要获取某个服务的地址,只须要到Zookeeper中找到这个path,而后就能够读取到里面存放的服务信息,这时候咱们就能够根据这些信息调用咱们的服务。这样,经过Zookeeper咱们就作到了动态的添加和删除服务,作到了一旦一个服务时效,就会自动从Zookeeper中移除,基本上Curator中的Service Discovery就是作的这点事。
  下面咱们用两张图片来比较一下,通常状况下的服务调用,和使用 Dynamic Service Registry 的区别web

    

 

  使用zookeeper作服务注册以后:apache

 

关于Apache curator的service discovery的一些介绍能够参考官方文档:http://curator.apache.org/curator-x-discovery/index.htmldom

 

Service Discovery 的使用

  通常而言,分为 Service Registry 和 Service Discovery,对应服务端和客户端。也就是由服务提供者,讲自身的信息注册到Zookeeper,而后,客户端经过到Zookeeper中查找服务信息,而后根据信息就行调用(见上图)。说了这么多,上代码了。ide

  首先咱们定义个payload,咱们这一在里面存储一些服务信息。这个信息会被保存在Zookeeper,这里只是举个例子,你还能够写入更多你想要的信息。工具

  

package discovery;

import org.codehaus.jackson.map.annotate.JsonRootName;



/**
 * Created by hupeng on 2014/9/16.
 */
@JsonRootName("details")
public class InstanceDetails {

    private String id;

    private String listenAddress;

    private int listenPort;

    private String interfaceName;

    public InstanceDetails(String id, String listenAddress, int listenPort,String interfaceName) {
        this.id = id;
        this.listenAddress = listenAddress;
        this.listenPort = listenPort;
        this.interfaceName = interfaceName;
    }

    public InstanceDetails() {
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getListenAddress() {
        return listenAddress;
    }

    public void setListenAddress(String listenAddress) {
        this.listenAddress = listenAddress;
    }

    public int getListenPort() {
        return listenPort;
    }

    public void setListenPort(int listenPort) {
        this.listenPort = listenPort;
    }

    public String getInterfaceName() {
        return interfaceName;
    }

    public void setInterfaceName(String interfaceName) {
        this.interfaceName = interfaceName;
    }

    @Override
    public String toString() {
        return "InstanceDetails{" +
                "id='" + id + '\'' +
                ", listenAddress='" + listenAddress + '\'' +
                ", listenPort=" + listenPort +
                ", interfaceName='" + interfaceName + '\'' +
                '}';
    }
}

 

咱们先写服务注册,也就是服务端那边作的事情。学习

package discovery;

import org.apache.curator.framework.CuratorFramework;
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.details.JsonInstanceSerializer;

import java.io.IOException;
/**
 * Created by hupeng on 2014/9/16.
 */
public class ServiceRegistrar{

    private ServiceDiscovery<InstanceDetails> serviceDiscovery;
    private final CuratorFramework client;


    public ServiceRegistrar(CuratorFramework client,String basePath) throws Exception {
        this.client = client;
        JsonInstanceSerializer<InstanceDetails> serializer = new JsonInstanceSerializer<InstanceDetails>(InstanceDetails.class);
        serviceDiscovery = ServiceDiscoveryBuilder.builder(InstanceDetails.class)
                .client(client)
                .serializer(serializer)
                .basePath(basePath)
                .build();
        serviceDiscovery.start();
    }

    public void registerService(ServiceInstance<InstanceDetails> serviceInstance) throws Exception {
        serviceDiscovery.registerService(serviceInstance);
    }

    public void unregisterService(ServiceInstance<InstanceDetails> serviceInstance) throws Exception {
        serviceDiscovery.unregisterService(serviceInstance);

    }

    public void updateService(ServiceInstance<InstanceDetails> serviceInstance) throws Exception {
        serviceDiscovery.updateService(serviceInstance);

    }

    public void close() throws IOException {
        serviceDiscovery.close();
    }
}

通常状况下,会在咱们服务启动的时候就将服务信息注册,好比咱们是web项目的话能够写一个Servlet Listener进行注册,这里为了方便,写一个Main方法进行测试,若是咱们把咱们的信息存储在payload中的话,UriSpec是能够不定义的。测试

package discovery;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.UriSpec;


import java.util.UUID;

/**
 * User: hupeng
 * Date: 14-9-16
 * Time: 下午8:05
 */
public class ServerApp {

    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));
        client.start();
        ServiceRegistrar serviceRegistrar = new ServiceRegistrar(client,"services");
        ServiceInstance<InstanceDetails> instance1 = ServiceInstance.<InstanceDetails>builder()
                .name("service1")
                .port(12345)
                .address("192.168.1.100")   //address不写的话,会取本地ip
                .payload(new InstanceDetails(UUID.randomUUID().toString(),"192.168.1.100",12345,"Test.Service1"))
                .uriSpec(new UriSpec("{scheme}://{address}:{port}"))
                .build();
        ServiceInstance<InstanceDetails> instance2 = ServiceInstance.<InstanceDetails>builder()
                .name("service2")
                .port(12345)
                .address("192.168.1.100")
                .payload(new InstanceDetails(UUID.randomUUID().toString(),"192.168.1.100",12345,"Test.Service2"))
                .uriSpec(new UriSpec("{scheme}://{address}:{port}"))
                .build();
        serviceRegistrar.registerService(instance1);
        serviceRegistrar.registerService(instance2);


        Thread.sleep(Integer.MAX_VALUE);
    }
}

再来写Service discoveryui

package discovery;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.curator.framework.CuratorFramework;
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.details.JsonInstanceSerializer;
import org.apache.curator.x.discovery.strategies.RandomStrategy;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.Map;
/**
 * Created by hupeng on 2014/9/16.
 */
public class ServiceDiscoverer {
    private ServiceDiscovery<InstanceDetails> serviceDiscovery;
    private Map<String, ServiceProvider<InstanceDetails>> providers = Maps.newHashMap();
    private List<Closeable> closeableList = Lists.newArrayList();
    private Object lock = new Object();


    public ServiceDiscoverer(CuratorFramework client ,String basePath) throws Exception {
        JsonInstanceSerializer<InstanceDetails> serializer = new JsonInstanceSerializer<InstanceDetails>(InstanceDetails.class);
        serviceDiscovery = ServiceDiscoveryBuilder.builder(InstanceDetails.class)
                .client(client)
                .basePath(basePath)
                .serializer(serializer)
                .build();

        serviceDiscovery.start();
    }


    public ServiceInstance<InstanceDetails> getInstanceByName(String serviceName) throws Exception {
        ServiceProvider<InstanceDetails> provider = providers.get(serviceName);
        if (provider == null) {
            synchronized (lock) {
                provider = providers.get(serviceName);
                if (provider == null) {
                    provider = serviceDiscovery.serviceProviderBuilder().
                            serviceName(serviceName).
                            providerStrategy(new RandomStrategy<InstanceDetails>())
                            .build();
                    provider.start();
                    closeableList.add(provider);
                    providers.put(serviceName, provider);
                }
            }
        }


        return provider.getInstance();
    }


    public synchronized void close(){
       for (Closeable closeable : closeableList) {
           CloseableUtils.closeQuietly(closeable);
       }
    }


}

客户端测试程序:

package discovery;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.x.discovery.ServiceInstance;

/**
 * User: hupeng
 * Date: 14-9-16
 * Time: 下午8:16
 */
public class ClientApp {

    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));
        client.start();
        ServiceDiscoverer serviceDiscoverer = new ServiceDiscoverer(client,"services");

        ServiceInstance<InstanceDetails> instance1 = serviceDiscoverer.getInstanceByName("service1");

        System.out.println(instance1.buildUriSpec());
        System.out.println(instance1.getPayload());

        ServiceInstance<InstanceDetails> instance2 = serviceDiscoverer.getInstanceByName("service1");

        System.out.println(instance2.buildUriSpec());
        System.out.println(instance2.getPayload());

        serviceDiscoverer.close();
        CloseableUtils.closeQuietly(client);
    }
}

好了,代码就到这里,若是有什么问题的话,请指正。

相关文章
相关标签/搜索