Service Discovery with Apache Curator

23858 ワード

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の使用の複雑さを低減した.

再試行メカニズム:すべての回復可能な異常をキャプチャするために再試行ポリシーを構成し、内部にもいくつかの標準的な再試行ポリシー(例えば指数補償)を提供する挿抜可能な再試行メカニズムを提供する.

接続状態モニタリング:Curatorは初期化後もzk接続を常に傍受し、接続状態が変化することを発見すると、対応する処理を行う.

zkクライアントインスタンス管理:Curatorはzkクライアントからserverクラスタへの接続を管理する.必要に応じてzkインスタンスを再構築し、zkクラスタとの信頼性の高い接続を保証する

各種使用シーンサポート:Curatorがzkサポートを実現する大部分の使用シーンサポート(zk自身がサポートしていないシーンも含む)は、zkのベストプラクティスに従い、様々な極端な状況を考慮する.

Curatorは以上の処理により、zk自体に多くの精力を費やすことなく、ユーザー自身の業務そのものに専念させる.ここで紹介するのは、CuratorのService Discoveryモジュールです
Service Discovery
私たちは通常、サービスを呼び出すときに、サービスのアドレス、ポート、または他の情報を知る必要があります.通常、私たちは彼らをプログラムに書きますが、サービスが多くなるにつれて、メンテナンスもますます苦労しています.さらに重要なのは、アドレスがプログラムに構成されているため、リモートのサービスが利用できるかどうか分かりません.サービスを追加または削除する場合は、プロファイルに構成する必要がありますか?この時、Zookeeperは助かりました.私たちのサービスをZookeeperに登録することができます.接続が切断されると削除される一時ノードを作成し、サービス情報を保存します.(url,ip,portなどの情報)、これらの一時ノードをserviceNameで命名されたノードの下に格納します.これにより、あるサービスのアドレスを取得するには、Zookeeperでこのpathを見つけて、そこに格納されているサービス情報を読み取るだけで、これらの情報に基づいてサービスを呼び出すことができます.これにより、Zookeeperで私たちは動的にサービスを追加・削除すると、サービスの時効が1つになると自動的にZookeeperから削除されるようになり、基本的にはCuratorのService Discoveryが行うことになります.一般的なサービスコールとDynamic Service Registryの違いを2枚の画像で比較してみましょう
　　

Service Discovery with Apache Curator_第1张图片

zookeeperを使用してサービス登録を行った後:

Service Discovery with Apache Curator_第2张图片

Apache curatorのservice discoveryについてのいくつかの紹介は、公式ドキュメント:http://curator.apache.org/curator-x-discovery/index.htmlを参照してください.

Service Discoveryの使用
一般的には、サービス側とクライアントに対応するサービスレジストリとサービス側Discoveryに分けられます.つまり、サービスプロバイダが、自身の情報をZookeeperに登録し、クライアントがZookeeperでサービス情報を検索し、情報に基づいて呼び出す(上図参照).これだけ言って、コードをつけた.
まず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プロジェクトであればサーブレット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 discoveryを書きます

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);
    }
}

はい、コードはここまでです.何か問題があれば、指摘してください.

【CentOS7＆AWS】EBSを新規に追加し、/var配下を新規EBSへマウントし直す手順

Oracle Database Cloud 各種ログのメンテナンス