import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
public HttpClient(Collection<String> nodes) {
    // searchShard
    // search template

    List<io.reactivex.netty.protocol.http.client.HttpClient<ByteBuf, ByteBuf>> clientsTemp = new ArrayList<>();
    // expect something like "http://%s:%d"
    for (String node : nodes) {
        String[] next = node.split(":");
        // indices admin
        String host = next[1].substring(2); // remove the // of http://
        int port = Integer.parseInt(next[2]);
        HttpClientBuilder<ByteBuf, ByteBuf> clientBuilder = RxNetty.newHttpClientBuilder(host, port);
        clientBuilder.config(new RxClient.ClientConfig.Builder().readTimeout(timeOut, MILLISECONDS).build());
        clientBuilder.withMaxConnections(maxConnections);
        clientsTemp.add(clientBuilder.build());
        logger.info("adding host {}:{}", host, port);
    }
    this.clients = new SnapshotableCopyOnWriteArray<>(clientsTemp);

    clientSupplier = new RoundRobinSupplier<>(clients);

    this.httpAdminClient = new HttpAdminClient(clientSupplier);
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
private HttpClientBuilder<ByteBuf, ByteBuf> httpClientBuilder() {
    HttpClientBuilder<ByteBuf, ByteBuf> builder = RxNetty
            .<ByteBuf, ByteBuf>newHttpClientBuilder(this.serviceEndpoint.getHost(), this.serviceEndpoint.getPort())
            .withSslEngineFactory(DefaultFactories.trustAll()).withMaxConnections(connectionPolicy.getMaxPoolSize())
            .withIdleConnectionsTimeoutMillis(this.connectionPolicy.getIdleConnectionTimeout() * 1000);

    ClientConfig config = new ClientConfig.Builder()
            .readTimeout(connectionPolicy.getRequestTimeout(), TimeUnit.SECONDS).build();
    return builder.config(config);
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
public ZmonMonitoringSystem(ZmonConfiguration configuration)
{
    HttpClientBuilder<ByteBuf, ByteBuf> builder = RxNetty.newHttpClientBuilder(configuration.getHost(), configuration.getPort());
    if (configuration.isSecure())
    {
        builder.withSslEngineFactory(DefaultFactories.trustAll());
    }
    rxNetty = builder.build();
    this.configuration = configuration;
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
/**
 * Creates a new hystrix reader.
 *
 * @param configuration The configuration to use.
 * @param cluster       The cluster to read from.
 */
public HystrixReader(Configuration configuration, String cluster)
{
    this.configuration = configuration;
    this.cluster = cluster;
    HttpClientBuilder<ByteBuf, ServerSentEvent> builder = RxNetty.newHttpClientBuilder(configuration.getTurbineHost(), configuration.getTurbinePort());
    builder.pipelineConfigurator(PipelineConfigurators.clientSseConfigurator());
    if (configuration.isSecure())
    {
        builder.withSslEngineFactory(DefaultFactories.trustAll());
    }
    rxNetty = builder.build();
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
@Override
protected HttpClient<I, ServerSentEvent> getOrCreateRxClient(Server server) {
    HttpClientBuilder<I, ServerSentEvent> clientBuilder =
            new HttpClientBuilder<I, ServerSentEvent>(server.getHost(), server.getPort()).pipelineConfigurator(pipelineConfigurator);
    int requestConnectTimeout = getProperty(IClientConfigKey.Keys.ConnectTimeout, null, DefaultClientConfigImpl.DEFAULT_CONNECT_TIMEOUT);
    RxClient.ClientConfig rxClientConfig = new HttpClientConfig.Builder().build();
    
    HttpClient<I, ServerSentEvent> client = clientBuilder.channelOption(
            ChannelOption.CONNECT_TIMEOUT_MILLIS, requestConnectTimeout).config(rxClientConfig).build();
    return client;
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
public WSClient(String host, int port, int firstStep, int stepSize, int stepDuration, String query) {
    this.host = host;
    this.port = port;
    this.firstStep = firstStep;
    this.stepSize = stepSize;
    this.stepDuration = stepDuration;
    this.query = query;

    System.out.println("Starting client with hostname: " + host + "  port: " + port + "  first-step: " + firstStep + "  step-size: " + stepSize + "  step-duration: " + stepDuration + "s  query: " + query);

    httpClient = new HttpClientBuilder<ByteBuf, ByteBuf>(this.host, this.port)
            .withMaxConnections(15000)
            .config(new HttpClient.HttpClientConfig.Builder().readTimeout(1, TimeUnit.MINUTES).build())
            .build();
    stats = new ConnectionPoolMetricListener();
    httpClient.subscribe(stats);

    client = httpClient.submit(HttpClientRequest.createGet(this.query))
            .flatMap(response -> {
                if (response.getStatus().code() == 200) {
                    counter.increment(CounterEvent.SUCCESS);
                } else {
                    counter.increment(CounterEvent.HTTP_ERROR);
                }
                return response.getContent().doOnNext(bb -> {
                    counter.add(CounterEvent.BYTES, bb.readableBytes());
                });
            }).doOnError((t) -> {
                if (t instanceof PoolExhaustedException) {
                    counter.increment(CounterEvent.POOL_EXHAUSTED);
                } else {
                    counter.increment(CounterEvent.NETTY_ERROR);
                }
            });
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
public TestRouteBasic(String backendHost, int backendPort) {
    host = backendHost;
    port = backendPort;
    client = new HttpClientBuilder<ByteBuf, ByteBuf>(host, port)
            .withMaxConnections(10000)
            .config(new HttpClient.HttpClientConfig.Builder().readTimeout(1, TimeUnit.MINUTES).build())
            .build();
    stats = new ConnectionPoolMetricListener();
    client.subscribe(stats);
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
/**
 * Execute an HTTP request.
 *
 * @param server Server to send the request to.
 * @param req    Request to execute.
 * @return Observable with the response of the request.
 */
private static Observable<HttpClientResponse<ByteBuf>>
executeSingle(Server server, HttpClientRequest<ByteBuf> req) {
    HttpClient.HttpClientConfig config = new HttpClient.HttpClientConfig.Builder()
            .readTimeout(READ_TIMEOUT_MS, TimeUnit.MILLISECONDS)
            .userAgent(USER_AGENT)
            .build();


    HttpClientBuilder<ByteBuf, ByteBuf> builder =
            RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder(server.host(), server.port())
                    .pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpClientConfigurator())
                    .config(config)
                    .channelOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIMEOUT_MS);

    if (server.isSecure()) {
        builder.withSslEngineFactory(DefaultFactories.trustAll());
    }

    final HttpClient<ByteBuf, ByteBuf> client = builder.build();
    return client.submit(req)
            .doOnNext(new Action1<HttpClientResponse<ByteBuf>>() {
                @Override
                public void call(HttpClientResponse<ByteBuf> res) {
                    LOGGER.debug("Got response: {}", res.getStatus().code());
                }
            })
            .doOnError(new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    LOGGER.info("Error sending metrics: {}/{}",
                            throwable.getClass().getSimpleName(),
                            throwable.getMessage());
                }
            })
            .doOnTerminate(new Action0() {
                @Override
                public void call() {
                    client.shutdown();
                }
            });
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
@Override
protected HttpClient<I, O> createRxClient(Server server) {
    HttpClientBuilder<I, O> clientBuilder;
    if (requestIdProvider != null) {
        clientBuilder = RxContexts.<I, O>newHttpClientBuilder(server.getHost(), server.getPort(), 
                requestIdProvider, RxContexts.DEFAULT_CORRELATOR, pipelineConfigurator);
    } else {
        clientBuilder = RxContexts.<I, O>newHttpClientBuilder(server.getHost(), server.getPort(), 
                RxContexts.DEFAULT_CORRELATOR, pipelineConfigurator);
    }
    Integer connectTimeout = getProperty(IClientConfigKey.Keys.ConnectTimeout,  null, DefaultClientConfigImpl.DEFAULT_CONNECT_TIMEOUT);
    Integer readTimeout    = getProperty(IClientConfigKey.Keys.ReadTimeout,     null, DefaultClientConfigImpl.DEFAULT_READ_TIMEOUT);
    Boolean followRedirect = getProperty(IClientConfigKey.Keys.FollowRedirects, null, null);
    HttpClientConfig.Builder builder = new HttpClientConfig.Builder().readTimeout(readTimeout, TimeUnit.MILLISECONDS);
    if (followRedirect != null) {
        builder.setFollowRedirect(followRedirect);
    }
    clientBuilder
            .channelOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout)
            .config(builder.build());
    if (isPoolEnabled()) {
        clientBuilder
            .withConnectionPoolLimitStrategy(poolStrategy)
            .withIdleConnectionsTimeoutMillis(idleConnectionEvictionMills)
            .withPoolIdleCleanupScheduler(poolCleanerScheduler);
    } 
    else {
        clientBuilder
            .withNoConnectionPooling();
    }
    
    if (sslContextFactory != null) {
        try {
            SSLEngineFactory myFactory = new DefaultFactories.SSLContextBasedFactory(sslContextFactory.getSSLContext()) {
                @Override
                public SSLEngine createSSLEngine(ByteBufAllocator allocator) {
                    SSLEngine myEngine = super.createSSLEngine(allocator);
                    myEngine.setUseClientMode(true);
                    return myEngine;
                }
            };

            clientBuilder.withSslEngineFactory(myFactory);
        } catch (ClientSslSocketFactoryException e) {
            throw new RuntimeException(e);
        }
    }
    return clientBuilder.build();
}
 

import io.reactivex.netty.protocol.http.client.HttpClientBuilder; //导入依赖的package包/类
public static HttpClient<ByteBuf, ByteBuf> createHttpClient(String host, int port) {
    return new HttpClientBuilder<ByteBuf, ByteBuf>(host, port).build();
}