import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
/**
 * Constructs a finagle java memcached client for the list of endpoints..
 *
 * @param endpoints list of {@code InetSocketAddress} for all the memcached servers.
 * @return {@link Client} to read/write to the hash ring of the servers..
 */
static Client makeMemcachedClient(Options opts, List<InetSocketAddress> endpoints)
    throws UnknownHostException {
  com.twitter.finagle.memcached.Client client =
      KetamaClientBuilder.get()
          .nodes(getHostPortWeightTuples(endpoints))
          .clientBuilder(ClientBuilder.get()
                             .codec(new Memcached())
                             .tcpConnectTimeout(new Duration(TimeUnit.MILLISECONDS.toNanos(opts.connectTimeoutMillis)))
                             .requestTimeout(new Duration(TimeUnit.MILLISECONDS.toNanos(opts.requestTimeoutMillis)))
                             .timeout(new Duration(TimeUnit.MILLISECONDS.toNanos(opts.e2eTimeoutMillis)))
                             .hostConnectionLimit(opts.hostConnectionLimit)
                             .hostConnectionMaxWaiters(opts.maxWaiters)
                             .retries(opts.requestRetries))
          .build();

  return new ClientBase(client);
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
@Test(timeout = 60000)
public void testBuildClientsFromSameBuilder() throws Exception {
    DistributedLogClientBuilder builder = DistributedLogClientBuilder.newBuilder()
            .name("build-clients-from-same-builder")
            .clientId(ClientId$.MODULE$.apply("test-builder"))
            .finagleNameStr("inet!127.0.0.1:7001")
            .streamNameRegex(".*")
            .handshakeWithClientInfo(true)
            .clientBuilder(ClientBuilder.get()
                .hostConnectionLimit(1)
                .connectTimeout(Duration.fromSeconds(1))
                .tcpConnectTimeout(Duration.fromSeconds(1))
                .requestTimeout(Duration.fromSeconds(10)));
    DistributedLogClient client1 = builder.build();
    DistributedLogClient client2 = builder.build();
    assertFalse(client1 == client2);
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
/**
 * Sanity check to make sure both checksum flag values work.
 */
@Test(timeout = 60000)
public void testChecksumFlag() throws Exception {
    String name = "dlserver-basic-write";
    LocalRoutingService routingService = LocalRoutingService.newBuilder().build();
    routingService.addHost(name, dlServer.getAddress());
    DistributedLogClientBuilder dlClientBuilder = DistributedLogClientBuilder.newBuilder()
        .name(name)
        .clientId(ClientId$.MODULE$.apply("test"))
        .routingService(routingService)
        .handshakeWithClientInfo(true)
        .clientBuilder(ClientBuilder.get()
            .hostConnectionLimit(1)
            .connectionTimeout(Duration.fromSeconds(1))
            .requestTimeout(Duration.fromSeconds(60)))
        .checksum(false);
    DistributedLogClient dlClient = (DistributedLogClientImpl) dlClientBuilder.build();
    Await.result(dlClient.write(name, ByteBuffer.wrap(("1").getBytes())));
    dlClient.close();

    dlClient = dlClientBuilder.checksum(true).build();
    Await.result(dlClient.write(name, ByteBuffer.wrap(("2").getBytes())));
    dlClient.close();
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
protected TwoRegionDLClient(String name, Map<SocketAddress, String> regionMap) {
    localRoutingService = new LocalRoutingService();
    remoteRoutingService = new LocalRoutingService();
    RegionsRoutingService regionsRoutingService =
            RegionsRoutingService.of(new DefaultRegionResolver(regionMap),
                    localRoutingService, remoteRoutingService);
    dlClientBuilder = DistributedLogClientBuilder.newBuilder()
                .name(name)
                .clientId(ClientId$.MODULE$.apply(name))
                .routingService(regionsRoutingService)
                .streamNameRegex(".*")
                .handshakeWithClientInfo(true)
                .maxRedirects(2)
                .clientBuilder(ClientBuilder.get()
                    .hostConnectionLimit(1)
                    .connectionTimeout(Duration.fromSeconds(1))
                    .requestTimeout(Duration.fromSeconds(10)));
    dlClient = (DistributedLogClientImpl) dlClientBuilder.build();
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
public static Builder newBuilder(String clientName,
                                 ClientId clientId,
                                 ClientBuilder clientBuilder,
                                 ClientConfig clientConfig,
                                 ClientStats clientStats) {
    return new DefaultBuilder(clientName, clientId, clientBuilder, clientConfig, clientStats);
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
private DefaultBuilder(String clientName,
                       ClientId clientId,
                       ClientBuilder clientBuilder,
                       ClientConfig clientConfig,
                       ClientStats clientStats) {
    this.clientName = clientName;
    this.clientId = clientId;
    this.clientStats = clientStats;
    // client builder
    ClientBuilder builder = setDefaultSettings(null == clientBuilder ? getDefaultClientBuilder() : clientBuilder);
    if (clientConfig.getThriftMux()) {
        builder = enableThriftMux(builder, clientId);
    }
    this.clientBuilder = builder;
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
private ClientBuilder getDefaultClientBuilder() {
    return ClientBuilder.get()
        .hostConnectionLimit(1)
        .tcpConnectTimeout(Duration.fromMilliseconds(200))
        .connectTimeout(Duration.fromMilliseconds(200))
        .requestTimeout(Duration.fromSeconds(1));
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private ClientBuilder setDefaultSettings(ClientBuilder builder) {
    return builder.name(clientName)
           .codec(ThriftClientFramedCodec.apply(Option.apply(clientId)))
           .failFast(false)
           .noFailureAccrual()
           // disable retries on finagle client builder, as there is only one host per finagle client
           // we should throw exception immediately on first failure, so DL client could quickly detect
           // failures and retry other proxies.
           .retries(1)
           .keepAlive(true);
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public ProxyClient build(SocketAddress address) {
    Service<ThriftClientRequest, byte[]> client =
        ClientBuilder.safeBuildFactory(
                clientBuilder
                        .hosts((InetSocketAddress) address)
                        .reportTo(clientStats.getFinagleStatsReceiver(address))
        ).toService();
    DistributedLogService.ServiceIface service =
            new DistributedLogService.ServiceToClient(client, new TBinaryProtocol.Factory());
    return new ProxyClient(address, client, service);
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
static DistributedLogClientBuilder createDistributedLogClientBuilder(ServerSet serverSet) {
    return DistributedLogClientBuilder.newBuilder()
                    .name("rebalancer_tool")
                    .clientId(ClientId$.MODULE$.apply("rebalancer_tool"))
                    .maxRedirects(2)
                    .serverSet(serverSet)
                    .clientBuilder(ClientBuilder.get()
                            .connectionTimeout(Duration.fromSeconds(2))
                            .tcpConnectTimeout(Duration.fromSeconds(2))
                            .requestTimeout(Duration.fromSeconds(10))
                            .hostConnectionLimit(1)
                            .hostConnectionCoresize(1)
                            .keepAlive(true)
                            .failFast(false));
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
@Override
protected int runCmd(CommandLine commandLine) throws Exception {
    try {
        parseCommandLine(commandLine);
    } catch (ParseException pe) {
        System.err.println("ERROR: failed to parse commandline : '" + pe.getMessage() + "'");
        printUsage();
        return -1;
    }

    DLZkServerSet serverSet = DLZkServerSet.of(uri, 60000);
    logger.info("Created serverset for {}", uri);
    try {
        DistributedLogClient client = DistributedLogClientBuilder.newBuilder()
                .name("proxy_tool")
                .clientId(ClientId$.MODULE$.apply("proxy_tool"))
                .maxRedirects(2)
                .serverSet(serverSet.getServerSet())
                .clientBuilder(ClientBuilder.get()
                    .connectionTimeout(Duration.fromSeconds(2))
                    .tcpConnectTimeout(Duration.fromSeconds(2))
                    .requestTimeout(Duration.fromSeconds(10))
                    .hostConnectionLimit(1)
                    .hostConnectionCoresize(1)
                    .keepAlive(true)
                    .failFast(false))
                .build();
        try {
            return runCmd(client);
        } finally {
            client.close();
        }
    } finally {
        serverSet.close();
    }
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
protected DLClient(String name, String streamNameRegex) {
    routingService = LocalRoutingService.newBuilder().build();
    dlClientBuilder = DistributedLogClientBuilder.newBuilder()
                .name(name)
                .clientId(ClientId$.MODULE$.apply(name))
                .routingService(routingService)
                .streamNameRegex(streamNameRegex)
                .handshakeWithClientInfo(true)
                .clientBuilder(ClientBuilder.get()
                    .hostConnectionLimit(1)
                    .connectionTimeout(Duration.fromSeconds(1))
                    .requestTimeout(Duration.fromSeconds(60)));
    dlClient = (DistributedLogClientImpl) dlClientBuilder.build();
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
public PinLaterClient(String host, int port, int concurrency) {
  this.service = ClientBuilder.safeBuild(
      ClientBuilder.get()
          .hosts(new InetSocketAddress(host, port))
          .codec(ThriftClientFramedCodec.apply(Option.apply(new ClientId("pinlaterclient"))))
          .hostConnectionLimit(concurrency)
          .tcpConnectTimeout(Duration.apply(2, TimeUnit.SECONDS))
          .requestTimeout(Duration.apply(10, TimeUnit.SECONDS))
          .retries(1));
  this.iface = new PinLater.ServiceToClient(service, new TBinaryProtocol.Factory());
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
protected Future<TerrapinServerInternal.ServiceIface> getClientFuture(final String hostName) {
  return connectionfuturePool.apply(new Function0<TerrapinServerInternal.ServiceIface>() {
    @Override
    public TerrapinServerInternal.ServiceIface apply() {
      Pair<Service<ThriftClientRequest, byte[]>, TerrapinServerInternal.ServiceIface> client =
          thriftClientCache.getIfPresent(hostName);
      if (client == null) {
        Service<ThriftClientRequest, byte[]> service = ClientBuilder.safeBuild(
            ClientBuilder.get()
                .hosts(new InetSocketAddress(hostName, targetPort))
                .codec(new ThriftClientFramedCodecFactory(Option.<ClientId>empty()))
                .retries(1)
                .connectTimeout(Duration.fromMilliseconds(connectTimeoutMs))
                .requestTimeout(Duration.fromMilliseconds(timeoutMs))
                .hostConnectionLimit(100)
                .failFast(false));

        client = new ImmutablePair(service, new TerrapinServerInternal.ServiceToClient(
            service, new TBinaryProtocol.Factory()));
        // A release is automatically called when an element is kicked out as part of the
        // removal listener. Doing a double release causes other issues.
        thriftClientCache.asMap().putIfAbsent(hostName, client);
      }
      return client.getRight();
    }
  });
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
public GeneratorMaster() {
    redisCache = new RedisCache("localhost", 7000);

    String propertiesPath = "/videogenmaster.properties";
    PropertyPlaceholder propsHolder = new PropertyPlaceholder(propertiesPath);
    propsHolder.generatePropertyMap();
    Map<String, String> props = propsHolder.getPropertyMap();

    videoCacheUpdatePool = Executors.newFixedThreadPool(Integer.parseInt(props.get("maxThreads")));
    videoCacheUpdateTimer = new Timer();
    /*videoCacheUpdateTimer.schedule(new VideoCache(videoCacheUpdatePool, propsHolder, client), 
            5000, 5000);*/

    stats = new Statistics();

    RetryPolicy<Try<Nothing$>> retryPolicy = new SimpleRetryPolicy<Try<Nothing$>>() {

        @Override
        public Duration backoffAt(int retryCount) {
            if (retryCount > 3) {
                return Duration.fromSeconds(16);
            }
            return Duration.fromSeconds((int) Math.pow(2.0, retryCount));
        }

        @Override
        public boolean shouldRetry(Try<Nothing$> arg) {
            return true;
        }
    };

    ClientBuilder clientBuilder = ClientBuilder.get()
            .codec(com.twitter.finagle.http.Http.get())
            .retryPolicy(retryPolicy) // Retry forever, with exponential backoff <= 16
            .hostConnectionLimit(500)
            .hosts("localhost:" + SLAVE_PORT);
    client = ClientBuilder.safeBuild(clientBuilder);
    // client = Http.newService("localhost:8001");
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
public GeneratorSlave() {
    ExecutorService pool = Executors.newFixedThreadPool(20);
    futurePool = new ExecutorServiceFuturePool(pool);
    stats = new SlaveStatistics();
    //client = Http.newService("localhost:8000");
    client = ClientBuilder
            .safeBuild(ClientBuilder.get().codec(com.twitter.finagle.http.Http.get())
                    .hosts(":8000").hostConnectionLimit(500));

}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
/**
 * initialise a {@link Client} instance using {@link ServiceFactory} from a
 * {@link ClientBuilder}
 *
 * @param addr
 */
public SimpleKestrelClient(InetSocketAddress addr) {
	final ClientBuilder<Command, Response, Yes, Yes, Yes> builder = ClientBuilder.get()
			.codec(Kestrel.get())
			.hosts(addr)
			.hostConnectionLimit(1);
	final ServiceFactory<Command, Response> kestrelClientBuilder = ClientBuilder.safeBuildFactory(builder);
	client = Client.newInstance(kestrelClientBuilder);
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private ClientBuilder enableThriftMux(ClientBuilder builder, ClientId clientId) {
    return builder.stack(ThriftMux.client().withClientId(clientId));
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
public DistributedLogClientImpl(String name,
                                ClientId clientId,
                                RoutingService routingService,
                                ClientBuilder clientBuilder,
                                ClientConfig clientConfig,
                                StatsReceiver statsReceiver,
                                StatsReceiver streamStatsReceiver,
                                RegionResolver regionResolver,
                                boolean enableRegionStats) {
    this.clientName = name;
    this.clientId = clientId;
    this.routingService = routingService;
    this.clientConfig = clientConfig;
    this.streamFailfast = clientConfig.getStreamFailfast();
    this.streamNameRegexPattern = Pattern.compile(clientConfig.getStreamNameRegex());
    this.regionResolver = regionResolver;
    // Build the timer
    this.dlTimer = new HashedWheelTimer(
            new ThreadFactoryBuilder().setNameFormat("DLClient-" + name + "-timer-%d").build(),
            this.clientConfig.getRedirectBackoffStartMs(),
            TimeUnit.MILLISECONDS);
    // register routing listener
    this.routingService.registerListener(this);
    // build the ownership cache
    this.ownershipCache = new OwnershipCache(this.clientConfig, this.dlTimer, statsReceiver, streamStatsReceiver);
    // Client Stats
    this.clientStats = new ClientStats(statsReceiver, enableRegionStats, regionResolver);
    // Client Manager
    this.clientBuilder = ProxyClient.newBuilder(clientName, clientId, clientBuilder, clientConfig, clientStats);
    this.clientManager = new ProxyClientManager(
            this.clientConfig,  // client config
            this.clientBuilder, // client builder
            this.dlTimer,       // timer
            this,               // host provider
            clientStats);       // client stats
    this.clientManager.registerProxyListener(this);

    // Cache Stats
    StatsReceiver cacheStatReceiver = statsReceiver.scope("cache");
    Seq<String> numCachedStreamsGaugeName =
            scala.collection.JavaConversions.asScalaBuffer(Arrays.asList("num_streams")).toList();
    cacheStatReceiver.provideGauge(numCachedStreamsGaugeName, new Function0<Object>() {
        @Override
        public Object apply() {
            return (float) ownershipCache.getNumCachedStreams();
        }
    });
    Seq<String> numCachedHostsGaugeName =
            scala.collection.JavaConversions.asScalaBuffer(Arrays.asList("num_hosts")).toList();
    cacheStatReceiver.provideGauge(numCachedHostsGaugeName, new Function0<Object>() {
        @Override
        public Object apply() {
            return (float) clientManager.getNumProxies();
        }
    });

    logger.info("Build distributedlog client : name = {}, client_id = {}, routing_service = {}, stats_receiver = {}, thriftmux = {}",
                new Object[] { name, clientId, routingService.getClass(), statsReceiver.getClass(), clientConfig.getThriftMux() });
}
 

import com.twitter.finagle.builder.ClientBuilder; //导入依赖的package包/类
/**
 * Attempt to load data (already in HDFS on a correct directory) into an already locked fileset.
 * The data is assumed to already have been placed in the correct directory on the terrapin
 * cluster. This is being called by the Terrapin loader jobs. The @fsInfo object is the same
 * as the locked fsInfo object.
 */
public static void loadFileSetData(ZooKeeperManager zkManager, FileSetInfo fsInfo, Options options)
    throws Exception {
    InetSocketAddress controllerSockAddress = zkManager.getControllerLeader();
    LOG.info("Connecting to controller at " +
            controllerSockAddress.getHostName() + ":" + controllerSockAddress.getPort());
    LOG.info("Load timeout " + Constants.LOAD_TIMEOUT_SECONDS + " seconds.");

    Service<ThriftClientRequest, byte[]> service = ClientBuilder.safeBuild(ClientBuilder.get()
            .hosts(controllerSockAddress)
            .codec(new ThriftClientFramedCodecFactory(Option.<ClientId>empty()))
            .retries(1)
            .connectTimeout(Duration.fromMilliseconds(1000))
            .requestTimeout(Duration.fromSeconds(Constants.LOAD_TIMEOUT_SECONDS))
            .hostConnectionLimit(100)
            .failFast(false));
    TerrapinController.ServiceIface iface = new TerrapinController.ServiceToClient(
            service, new TBinaryProtocol.Factory());
    TerrapinLoadRequest request = new TerrapinLoadRequest();
    request.setHdfsDirectory(fsInfo.servingInfo.hdfsPath);
    request.setOptions(options);
    request.setFileSet(fsInfo.fileSetName);
    request.setExpectedNumPartitions(fsInfo.servingInfo.numPartitions);

    LOG.info("Loading file set " + fsInfo.fileSetName + " at " + fsInfo.servingInfo.hdfsPath);
    long startTimeSeconds = System.currentTimeMillis() / 1000;
    int numTriesLeft = 5;
    boolean done = false;
    Exception e = null;
    while (numTriesLeft > 0) {
        try {
            iface.loadFileSet(request).get();
            done = true;
            break;
        } catch (Throwable t) {
            LOG.error("Swap failed with exception.", t);
            e = new Exception(t);
            numTriesLeft--;
        }
        LOG.info("Retrying in 10 seconds.");
        try {
            Thread.sleep(10000);
        } catch (InterruptedException ie) {
            LOG.error("Interrupted.");
            break;
        }
    }
    if (done) {
        LOG.info("Load successful. Swap took " +
                ((System.currentTimeMillis() / 1000) - startTimeSeconds) + " seconds.");
    } else {
        LOG.error("Load failed !!!.");
        throw new Exception(e);
    }
}