etcd使用经历

* Copyright (c) 2015, Jurriaan Mous and contributors as indicated by the &＃64;author tags.
package mousio.etcd4j;import io.netty.handler.ssl.SslContext;
import mousio.client.retry.RetryPolicy;
import mousio.client.retry.RetryWithExponentialBackOff;
import mousio.etcd4j.requests.*;
import mousio.etcd4j.responses.EtcdAuthenticationException;
import mousio.etcd4j.responses.EtcdException;
import mousio.etcd4j.responses.EtcdHealthResponse;
import mousio.etcd4j.responses.EtcdMembersResponse;
import mousio.etcd4j.responses.EtcdSelfStatsResponse;
import mousio.etcd4j.responses.EtcdStoreStatsResponse;
import mousio.etcd4j.responses.EtcdVersionResponse;
import mousio.etcd4j.responses.EtcdLeaderStatsResponse;
import mousio.etcd4j.transport.EtcdClientImpl;
import mousio.etcd4j.transport.EtcdNettyClient;import java.io.Closeable;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.TimeoutException;/*** Etcd client.*/
public class EtcdClient implements Closeable {private final EtcdClientImpl client;private RetryPolicy retryHandler;/*** Constructor** &＃64;param baseUri URI to create connection on*/public EtcdClient(URI... baseUri) {this(EtcdSecurityContext.NONE, baseUri);}/*** Constructor** &＃64;param username username* &＃64;param password password* &＃64;param baseUri URI to create connection on*/public EtcdClient(String username, String password, URI... baseUri) {this(EtcdSecurityContext.withCredential(username, password), baseUri);}/*** Constructor** &＃64;param sslContext context for Ssl connections* &＃64;param username username* &＃64;param password password* &＃64;param baseUri URI to create connection on*/public EtcdClient(SslContext sslContext, String username, String password, URI... baseUri) {this(new EtcdSecurityContext(sslContext, username, password), baseUri);}/*** Constructor** &＃64;param sslContext context for Ssl connections* &＃64;param baseUri URI to create connection on*/public EtcdClient(SslContext sslContext, URI... baseUri) {this(EtcdSecurityContext.withSslContext(sslContext), baseUri);}/*** Constructor** &＃64;param securityContext context for security* &＃64;param baseUri URI to create connection on*/public EtcdClient(EtcdSecurityContext securityContext, URI... baseUri) {this(new EtcdNettyClient(securityContext,(baseUri.length &＃61;&＃61; 0)? new URI[] { URI.create("https://127.0.0.1:4001") }: baseUri));}/*** Create a client with a custom implementation** &＃64;param etcdClientImpl to create client with.*/public EtcdClient(EtcdClientImpl etcdClientImpl) {this.client &＃61; etcdClientImpl;this.retryHandler &＃61; RetryWithExponentialBackOff.DEFAULT;}/*** Get the version of the Etcd server** &＃64;return version as String* &＃64;deprecated use version() when using etcd 2.1&＃43;.*/&＃64;Deprecatedpublic String getVersion() {try {return new EtcdOldVersionRequest(this.client, retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Get the version of the Etcd server** &＃64;return version*/public EtcdVersionResponse version() {try {return new EtcdVersionRequest(this.client, retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Get the Self Statistics of Etcd** &＃64;return EtcdSelfStatsResponse*/public EtcdSelfStatsResponse getSelfStats() {try {return new EtcdSelfStatsRequest(this.client, retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Get the Leader Statistics of Etcd** &＃64;return EtcdLeaderStatsResponse*/public EtcdLeaderStatsResponse getLeaderStats() {try {return new EtcdLeaderStatsRequest(this.client, retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Get the Store Statistics of Etcd** &＃64;return vEtcdStoreStatsResponse*/public EtcdStoreStatsResponse getStoreStats() {try {return new EtcdStoreStatsRequest(this.client, retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Get the Members of Etcd** &＃64;return vEtcdMembersResponse*/public EtcdMembersResponse getMembers() {try {return new EtcdMembersRequest(this.client,retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Get the Members of Etcd** &＃64;return vEtcdMembersResponse*/public EtcdHealthResponse getHealth() {try {return new EtcdHealthRequest(this.client,retryHandler).send().get();} catch (IOException | EtcdException | EtcdAuthenticationException | TimeoutException e) {return null;}}/*** Put a key with a value** &＃64;param key to put* &＃64;param value to put on key* &＃64;return EtcdKeysRequest*/public EtcdKeyPutRequest put(String key, String value) {return new EtcdKeyPutRequest(client, key, retryHandler).value(value);}/*** Refresh a key with new ttl* (without notifying watchers when using etcd 2.3&＃43;)** &＃64;param key to refresh* &＃64;param ttl to update key with* &＃64;return EtcdKeysRequest*/public EtcdKeyPutRequest refresh(String key, Integer ttl) {return new EtcdKeyPutRequest(client, key, retryHandler).refresh(ttl);}/*** Create a dir** &＃64;param dir to create* &＃64;return EtcdKeysRequest*/public EtcdKeyPutRequest putDir(String dir) {return new EtcdKeyPutRequest(client, dir, retryHandler).isDir();}/*** Post a value to a key for in-order keys.** &＃64;param key to post to* &＃64;param value to post* &＃64;return EtcdKeysRequest*/public EtcdKeyPostRequest post(String key, String value) {return new EtcdKeyPostRequest(client, key, retryHandler).value(value);}/*** Deletes a key** &＃64;param key to delete* &＃64;return EtcdKeysRequest*/public EtcdKeyDeleteRequest delete(String key) {return new EtcdKeyDeleteRequest(client, key, retryHandler);}/*** Deletes a directory** &＃64;param dir to delete* &＃64;return EtcdKeysRequest*/public EtcdKeyDeleteRequest deleteDir(String dir) {return new EtcdKeyDeleteRequest(client, dir, retryHandler).dir();}/*** Get by key** &＃64;param key to get* &＃64;return EtcdKeysRequest*/public EtcdKeyGetRequest get(String key) {return new EtcdKeyGetRequest(client, key, retryHandler);}/*** Get directory** &＃64;param dir to get* &＃64;return EtcdKeysGetRequest*/public EtcdKeyGetRequest getDir(String dir) {return new EtcdKeyGetRequest(client, dir, retryHandler).dir();}/*** Get all keys** &＃64;return EtcdKeysRequest*/public EtcdKeyGetRequest getAll() {return new EtcdKeyGetRequest(client, retryHandler);}&＃64;Overridepublic void close() throws IOException {if (client !&＃61; null) {client.close();}}/*** Set the retry handler. Default is an exponential back-off with start of 20ms.** &＃64;param retryHandler to set* &＃64;return this instance*/public EtcdClient setRetryHandler(RetryPolicy retryHandler) {this.retryHandler &＃61; retryHandler;return this;}
}

* Copyright (c) 2015, Jurriaan Mous and contributors as indicated by the &＃64;author tags.
package mousio.etcd4j.transport;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.codec.base64.Base64;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.multipart.HttpPostRequestEncoder;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.resolver.dns.DnsAddressResolverGroup;
import io.netty.resolver.dns.DnsServerAddresses;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import mousio.client.ConnectionState;
import mousio.client.retry.RetryHandler;
import mousio.etcd4j.EtcdSecurityContext;
import mousio.etcd4j.promises.EtcdResponsePromise;
import mousio.etcd4j.requests.EtcdRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.channels.ClosedChannelException;
import java.util.Map;
import java.util.concurrent.CancellationException;/*** &＃64;author Jurriaan Mous* &＃64;author Luca Burgazzoli** Netty client for the requests and responses*/
public class EtcdNettyClient implements EtcdClientImpl {private static final Logger logger &＃61; LoggerFactory.getLogger(EtcdNettyClient.class);// default etcd portprivate static final int DEFAULT_PORT &＃61; 2379;private static final String ENV_ETCD4J_ENDPOINT &＃61; "ETCD4J_ENDPOINT";private final EventLoopGroup eventLoopGroup;private final URI[] uris;private final Bootstrap bootstrap;//private final String hostName;private final EtcdNettyConfig config;private final EtcdSecurityContext securityContext;protected volatile int lastWorkingUriIndex;/*** Constructor** &＃64;param sslContext SSL context if connecting with SSL. Null if not connecting with SSL.* &＃64;param uri to connect to*/public EtcdNettyClient(final SslContext sslContext, final URI... uri) {this(new EtcdNettyConfig(), sslContext, uri);}/*** Constructor** &＃64;param securityContext security context.* &＃64;param uri to connect to*/public EtcdNettyClient(final EtcdSecurityContext securityContext, final URI... uri) {this(new EtcdNettyConfig(), securityContext, uri);}/*** Constructor with custom eventloop group and timeout** &＃64;param config for netty* &＃64;param sslContext SSL context if connecting with SSL. Null if not connecting with SSL.* &＃64;param uris to connect to*/public EtcdNettyClient(final EtcdNettyConfig config,final SslContext sslContext, final URI... uris) {this(config, new EtcdSecurityContext(sslContext), uris);}/*** Constructor with custom eventloop group and timeout** &＃64;param config for netty* &＃64;param uris to connect to*/public EtcdNettyClient(final EtcdNettyConfig config, final URI... uris) {this(config, EtcdSecurityContext.NONE, uris);}/*** Constructor with custom eventloop group and timeout** &＃64;param config for netty* &＃64;param securityContext security context (ssl, authentication)* &＃64;param uris to connect to*/public EtcdNettyClient(final EtcdNettyConfig config,final EtcdSecurityContext securityContext, final URI... uris) {logger.info("Setting up Etcd4j Netty client");this.lastWorkingUriIndex &＃61; 0;this.config &＃61; config.clone();this.securityContext &＃61; securityContext.clone();this.uris &＃61; uris;this.eventLoopGroup &＃61; config.getEventLoopGroup();this.bootstrap &＃61; new Bootstrap().group(eventLoopGroup).channel(config.getSocketChannelClass()).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout()).resolver(new DnsAddressResolverGroup(NioDatagramChannel.class,DnsServerAddresses.defaultAddresses())).handler(new ChannelInitializer() {&＃64;Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p &＃61; ch.pipeline();if (securityContext.hasNettySsl()) {p.addLast(securityContext.nettySslContext().newHandler(ch.alloc()));} else if (securityContext.hasSsl()) {p.addLast(new SslHandler(securityContext.sslContext().createSSLEngine()));}p.addLast("codec", new HttpClientCodec());p.addLast("auth", new HttpBasicAuthHandler());p.addLast("chunkedWriter", new ChunkedWriteHandler());p.addLast("aggregate", new HttpObjectAggregator(config.getMaxFrameSize()));}});}/*** For tests** &＃64;return the current bootstrap*/protected Bootstrap getBootstrap() {return bootstrap;}/*** Send a request and get a future.** &＃64;param etcdRequest Etcd Request to send* &＃64;return Promise for the request.*/public EtcdResponsePromise send(final EtcdRequest etcdRequest) throws IOException {ConnectionState connectionState &＃61; new ConnectionState(uris, lastWorkingUriIndex);if (etcdRequest.getPromise() &＃61;&＃61; null) {etcdRequest.setPromise(new EtcdResponsePromise(etcdRequest.getRetryPolicy(),connectionState,new RetryHandler() {&＃64;Overridepublic void doRetry(ConnectionState connectionState) throws IOException {connect(etcdRequest, connectionState);}}));}connect(etcdRequest, connectionState);return etcdRequest.getPromise();}/*** Connect to server** &＃64;param etcdRequest to request with* &＃64;param Type of response* &＃64;throws IOException if request could not be sent.*/&＃64;SuppressWarnings("unchecked")protected void connect(final EtcdRequest etcdRequest) throws IOException {this.connect(etcdRequest, etcdRequest.getPromise().getConnectionState());}/*** Connect to server** &＃64;param etcdRequest to request with* &＃64;param connectionState for retries* &＃64;param Type of response* &＃64;throws IOException if request could not be sent.*/&＃64;SuppressWarnings("unchecked")protected void connect(final EtcdRequest etcdRequest, final ConnectionState connectionState) throws IOException {if(eventLoopGroup.isShuttingDown() || eventLoopGroup.isShutdown() || eventLoopGroup.isTerminated()){etcdRequest.getPromise().getNettyPromise().cancel(true);logger.debug("Retry canceled because of closed etcd client");return;}final URI uri;// when we are called from a redirect, the url in the request may also// contain host and port!URI requestUri &＃61; URI.create(etcdRequest.getUrl());if (requestUri.getHost() !&＃61; null && requestUri.getPort() > -1) {uri &＃61; requestUri;} else if (connectionState.uris.length &＃61;&＃61; 0 && System.getenv(ENV_ETCD4J_ENDPOINT) !&＃61; null) {// read uri from environment variableString endpoint_uri &＃61; System.getenv(ENV_ETCD4J_ENDPOINT);if(logger.isDebugEnabled()) {logger.debug("Will use environment variable {} as uri with value {}", ENV_ETCD4J_ENDPOINT, endpoint_uri);}uri &＃61; URI.create(endpoint_uri);} else {uri &＃61; connectionState.uris[connectionState.uriIndex];}// Start the connection attempt.final ChannelFuture connectFuture &＃61; bootstrap.connect(connectAddress(uri));etcdRequest.getPromise().getConnectionState().loop &＃61; connectFuture.channel().eventLoop();etcdRequest.getPromise().attachNettyPromise(connectFuture.channel().eventLoop().newPromise());connectFuture.addListener(new GenericFutureListener() {&＃64;Overridepublic void operationComplete(final ChannelFuture f) throws Exception {if (!f.isSuccess()) {final Throwable cause &＃61; f.cause();if (logger.isDebugEnabled()) {logger.debug("Connection failed to {}, cause {}", connectionState.uris[connectionState.uriIndex], cause);}if (cause instanceof ClosedChannelException || cause instanceof IllegalStateException) {etcdRequest.getPromise().cancel(new CancellationException("Channel closed"));} else {etcdRequest.getPromise().handleRetry(f.cause());}return;}// Handle already cancelled promisesif (etcdRequest.getPromise().getNettyPromise().isCancelled()) {f.channel().close();etcdRequest.getPromise().getNettyPromise().setFailure(new CancellationException());return;}final Promise listenedToPromise &＃61; etcdRequest.getPromise().getNettyPromise();// Close channel when promise is satisfied or cancelled laterlistenedToPromise.addListener(new GenericFutureListener>() {&＃64;Overridepublic void operationComplete(Future future) throws Exception {// Only close if it was not redirected to new promiseif (etcdRequest.getPromise().getNettyPromise() &＃61;&＃61; listenedToPromise) {f.channel().close();}}});if (logger.isDebugEnabled()) {logger.debug("Connected to {} ({})", f.channel().remoteAddress().toString(), connectionState.uriIndex);}lastWorkingUriIndex &＃61; connectionState.uriIndex;modifyPipeLine(etcdRequest, f.channel().pipeline());createAndSendHttpRequest(uri, etcdRequest.getUrl(), etcdRequest, f.channel()).addListener(new ChannelFutureListener() {&＃64;Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (!future.isSuccess()) {etcdRequest.getPromise().setException(future.cause());if (!f.channel().eventLoop().inEventLoop()) {f.channel().eventLoop().shutdownGracefully();}f.channel().close();}}});f.channel().closeFuture().addListener(new ChannelFutureListener() {&＃64;Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (logger.isDebugEnabled()) {logger.debug("Connection closed for request {} on uri {} ",etcdRequest.getMethod().name(),etcdRequest.getUri());}}});}});}/*** Modify the pipeline for the request** &＃64;param req to process* &＃64;param pipeline to modify* &＃64;param Type of Response*/private void modifyPipeLine(final EtcdRequest req, final ChannelPipeline pipeline) {final EtcdResponseHandler handler &＃61; new EtcdResponseHandler<>(this, req);if (req.hasTimeout()) {pipeline.addFirst(new ReadTimeoutHandler(req.getTimeout(), req.getTimeoutUnit()));}pipeline.addLast(handler);pipeline.addLast(new ChannelHandlerAdapter() {&＃64;Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {handler.retried(true);req.getPromise().handleRetry(cause);}});}/*** Get HttpRequest belonging to etcdRequest** &＃64;param server server for http request* &＃64;param uri to send request to* &＃64;param etcdRequest to send* &＃64;param channel to send request on* &＃64;param Response type* &＃64;return HttpRequest* &＃64;throws Exception when creating or sending HTTP request fails*/private ChannelFuture createAndSendHttpRequest(URI server, String uri, EtcdRequest etcdRequest, Channel channel) throws Exception {HttpRequest httpRequest &＃61; new DefaultHttpRequest(HttpVersion.HTTP_1_1, etcdRequest.getMethod(), uri);httpRequest.headers().add(HttpHeaderNames.CONNECTION, "keep-alive");if(!this.config.hasHostName()) {httpRequest.headers().add(HttpHeaderNames.HOST, server.getHost() &＃43; ":" &＃43; server.getPort());} else {httpRequest.headers().add(HttpHeaderNames.HOST, this.config.getHostName());}HttpPostRequestEncoder bodyRequestEncoder &＃61; null;Map keyValuePairs &＃61; etcdRequest.getRequestParams();if (keyValuePairs !&＃61; null && !keyValuePairs.isEmpty()) {HttpMethod etcdRequestMethod &＃61; etcdRequest.getMethod();if (etcdRequestMethod &＃61;&＃61; HttpMethod.POST || etcdRequestMethod &＃61;&＃61; HttpMethod.PUT) {bodyRequestEncoder &＃61; new HttpPostRequestEncoder(httpRequest, false);for (Map.Entry entry : keyValuePairs.entrySet()) {bodyRequestEncoder.addBodyAttribute(entry.getKey(), entry.getValue());}httpRequest &＃61; bodyRequestEncoder.finalizeRequest();} else {QueryStringEncoder encoder &＃61; new QueryStringEncoder(uri);for (Map.Entry entry : keyValuePairs.entrySet()) {encoder.addParam(entry.getKey() , entry.getValue());}httpRequest.setUri(encoder.toString());}}etcdRequest.setHttpRequest(httpRequest);ChannelFuture future &＃61; channel.write(httpRequest);if (bodyRequestEncoder !&＃61; null && bodyRequestEncoder.isChunked()) {future &＃61; channel.write(bodyRequestEncoder);}channel.flush();return future;}/*** Close netty*/&＃64;Overridepublic void close() {logger.info("Shutting down Etcd4j Netty client");if (config.isManagedEventLoopGroup()) {logger.debug("Shutting down Netty Loop");eventLoopGroup.shutdownGracefully();}}private InetSocketAddress connectAddress(URI uri) {return InetSocketAddress.createUnresolved(uri.getHost(), uri.getPort() &＃61;&＃61; -1 ? DEFAULT_PORT : uri.getPort());}private class HttpBasicAuthHandler extends ChannelOutboundHandlerAdapter {&＃64;Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {if (securityContext.hasCredentials() && msg instanceof HttpRequest) {addBasicAuthHeader((HttpRequest)msg);}ctx.write(msg, promise);}private void addBasicAuthHeader(HttpRequest request) {final String auth &＃61; Base64.encode(Unpooled.copiedBuffer(securityContext.username() &＃43; ":" &＃43; securityContext.password(),CharsetUtil.UTF_8)).toString(CharsetUtil.UTF_8);request.headers().add(HttpHeaderNames.AUTHORIZATION, "Basic " &＃43; auth);}}
}