Java 类org.apache.http.nio.reactor.IOReactorException 实例源码

private CloseableHttpAsyncClient createHttpAsyncClient(YunpianConf conf) throws IOReactorException {
    IOReactorConfig ioReactorConfig = IOReactorConfig.custom().setIoThreadCount(Runtime.getRuntime().availableProcessors())
            .setConnectTimeout(conf.getConfInt(YunpianConf.HTTP_CONN_TIMEOUT, "10000"))
            .setSoTimeout(conf.getConfInt(YunpianConf.HTTP_SO_TIMEOUT, "30000")).build();
    ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);

    PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
    ConnectionConfig connectionConfig = ConnectionConfig.custom().setMalformedInputAction(CodingErrorAction.IGNORE)
            .setUnmappableInputAction(CodingErrorAction.IGNORE)
            .setCharset(Charset.forName(conf.getConf(YunpianConf.HTTP_CHARSET, YunpianConf.HTTP_CHARSET_DEFAULT))).build();
    connManager.setDefaultConnectionConfig(connectionConfig);
    connManager.setMaxTotal(conf.getConfInt(YunpianConf.HTTP_CONN_MAXTOTAL, "100"));
    connManager.setDefaultMaxPerRoute(conf.getConfInt(YunpianConf.HTTP_CONN_MAXPERROUTE, "10"));

    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(connManager).build();
    httpclient.start();
    return httpclient;
}

public FiberApacheHttpClientRequestExecutor(final Validator<CloseableHttpResponse> resValidator, final int maxConnections, final int timeout, final int parallelism) throws IOReactorException {
  final DefaultConnectingIOReactor ioreactor = new DefaultConnectingIOReactor(IOReactorConfig.custom().
    setConnectTimeout(timeout).
    setIoThreadCount(parallelism).
    setSoTimeout(timeout).
    build());

  final PoolingNHttpClientConnectionManager mngr = new PoolingNHttpClientConnectionManager(ioreactor);
  mngr.setDefaultMaxPerRoute(maxConnections);
  mngr.setMaxTotal(maxConnections);

  final CloseableHttpAsyncClient ahc = HttpAsyncClientBuilder.create().
    setConnectionManager(mngr).
    setDefaultRequestConfig(RequestConfig.custom().setLocalAddress(null).build()).build();

  client = new FiberHttpClient(ahc);
  validator = resValidator;
}

public AsyncConnectionManager(
        HttpHost target,
        int maxConnections,
        NHttpClientHandler handler, 
        HttpParams params) throws IOReactorException {
    super();
    this.target = target;
    this.maxConnections = maxConnections;
    this.handler = handler;
    this.params = params;
    this.lock = new Object();
    this.allConns = new HashSet<NHttpClientConnection>();
    this.availableConns = new LinkedList<NHttpClientConnection>();
    this.pendingRequests = new LinkedList<AsyncConnectionRequest>();
    this.ioreactor = new DefaultConnectingIOReactor(2, params);
}

private DashboardSetupStatus initDashboard(final String hostAddress, final int port) {
  final String dashboardURL = String.format("http://%s:%d/", hostAddress, port);
  try {
    // Create a pool of http client connection, which allow up to Integer.MAX_VALUE connections.
    final PoolingNHttpClientConnectionManager connectionManager
        = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor());
    connectionManager.setMaxTotal(Integer.MAX_VALUE);
    final CloseableHttpAsyncClient reusableHttpClient =
        HttpAsyncClients.custom().setConnectionManager(connectionManager).build();
    reusableHttpClient.start();

    // run another thread to send metrics.
    runMetricsSenderThread();

    return DashboardSetupStatus.getSuccessful(dashboardURL, reusableHttpClient);
  } catch (IOReactorException e) {
    LOG.log(Level.WARNING, "Dashboard: Fail on initializing connection to the dashboard server.", e);
    return DashboardSetupStatus.getFailed();
  }
}

/**
 * Used internally to initialize the internal HTTP client used by all
 * instances of a client.
 * <p>
 * This method can be overriden to provide a client with different options.
 * The client built gets an extra interceptor to add the credentials headers.
 *
 * @return HTTP default async client builder
 */
protected static HttpAsyncClientBuilder defaultClientBuilder() {

  try {
    DefaultConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
    connMgr = new PoolingNHttpClientConnectionManager(ioReactor);
    connMgr.setMaxTotal(maxConnections);
    connMgr.setDefaultMaxPerRoute(maxConnections);
  } catch (IOReactorException e) {
  }

  return HttpAsyncClients
      .custom()
      .addInterceptorLast(new GzipInterceptors.GzipRequestInterceptor())
      .setConnectionManager(connMgr)
      .setDefaultRequestConfig(
          RequestConfig.custom()
          .setSocketTimeout(3600 * 1000) // 1 hour
              .build())
      .setKeepAliveStrategy(keepAliveStrategy);
}

public AsyncServiceClient(ClientConfiguration config) {
    super(config);
    try {
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(config.getIoThreadCount()).build();
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(
                ioReactorConfig);
        PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(
                ioReactor);
        cm.setMaxTotal(config.getMaxConnections());
        cm.setDefaultMaxPerRoute(config.getMaxConnections());
        httpClient = new HttpFactory().createHttpAsyncClient(config, cm);

        /*
         * socketTimeout的值限制了closeIdleConnections执行的周期。
         * 如果周期相对socketTimeout的值过长，有可能一个请求分配到一个即将socketTimeout的连接，
         * 在请求发送之前即抛出SocketTimeoutException。
         * 现在让closeIdleConnections的执行周期为socketTimeout / 2.5。
         */
        long closePeriod = 5000;
        if (config.getSocketTimeoutInMillisecond() > 0) {
            closePeriod = (long) (config.getSocketTimeoutInMillisecond() / 2.5);
        }
        closePeriod = closePeriod < 5000 ? closePeriod : 5000;
        connEvictor = new IdleConnectionEvictor(cm, closePeriod);
        httpClient.start();
        connEvictor.start();
    } catch (IOReactorException ex) {
        throw new ClientException(String.format("IOReactorError: %s",
                ex.getMessage()), ex);
    }
}

@Bean
PoolingNHttpClientConnectionManager poolingNHttpClientConnectionManager() throws IOReactorException {
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(maxTotalConnections);
    connectionManager.setDefaultMaxPerRoute(maxConnectionsPerRoute);
    return connectionManager;
}

protected TopicBasedSelectorBot(String _targetHost, String _userId, String _topic, ResultsType _resultsType) throws IOReactorException {
    super(_userId, _topic);
    this.resultsType = _resultsType;
    requestAdapter = new RmendRequestAdapter(_targetHost, ENDPOINT_TEMPLATE);

    System.out.println();
    logger.info(StringUtils.repeat("-","*",30));
    logger.info("Target Host: "+_targetHost);
    logger.info("UID: "+_userId);
    logger.info("Topic: "+_topic);
    logger.info("Result type: "+resultsType);
    logger.info(StringUtils.repeat("-","*",30));
}

public static void main(final String[] args) throws SuspendExecution, InterruptedException, ExecutionException, IOReactorException, IOException {
  final IntervalGenerator intervalGenerator = new ConstantIntervalGenerator(10000000);
  try (final FiberApacheHttpClientRequestExecutor requestExecutor =
          new FiberApacheHttpClientRequestExecutor<>((res) -> {
            if (res == null) {
              throw new AssertionError("Response is null");
            }
            final int status = res.getStatusLine().getStatusCode();
            if (status != 200) {
              throw new AssertionError("Status is " + status);
            }
          }, 1000000)) {

    final Channel<HttpGet> requestCh = Channels.newChannel(1000);
    final Channel<TimingEvent<CloseableHttpResponse>> eventCh = Channels.newChannel(1000);

    // Requests generator
    new Fiber<Void>("req-gen", () -> {
      // Bench handling 1k reqs
      for (int i = 0; i < 1000; ++i) {
        requestCh.send(new HttpGet("http://localhost:8080/hello-world"));
      }

      requestCh.close();
    }).start();

    final Histogram histogram = new Histogram(3600000000L, 3);

    // Event recording, both HistHDR and logging
    record(eventCh, new HdrHistogramRecorder(histogram, 1000000), new LoggingRecorder(LOG));

    // Main
    new Fiber<Void>("jbender", () -> {
      JBender.loadTestThroughput(intervalGenerator, 0, requestCh, requestExecutor, eventCh);
    }).start().join();

    histogram.outputPercentileDistribution(System.out, 1000.0);
  }
}

public RmendRequestAdapter(String _targetHost, String _endpointTemplate) throws IOReactorException {
    ioReactor = new DefaultConnectingIOReactor();
    cm = new PoolingNHttpClientConnectionManager(ioReactor);
    cm.setMaxTotal(20);

    httpclient = HttpAsyncClients.createPipelining();
    this.targetHost = _targetHost;
    this.endpointTemplate = _endpointTemplate;
}

/**
   * 
   * @throws Exception .
   */
  private void init() throws IOReactorException {
HttpParams params = new SyncBasicHttpParams();  
HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1);
HttpConnectionParams.setSocketBufferSize(params, 8192);

HttpConnectionParams.setConnectionTimeout(params, connectionTimeOut);       
HttpConnectionParams.setSoTimeout(params, 3000);

httpclient = new DefaultHttpAsyncClient(params);
httpclient.start();

  }

private HttpAsyncClientBuilder initialize() {
    try {
        final PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor( IOReactorConfig.custom()
                .setConnectTimeout( connectTimeout )
                .setSoTimeout( readTimeout )
                .build() ),
            RegistryBuilder.<SchemeIOSessionStrategy>create()
                .register( "http", NoopIOSessionStrategy.INSTANCE )
                .register( "https",
                    new SSLIOSessionStrategy( certificateLocation != null ?
                        createSSLContext( certificateLocation, certificatePassword )
                        : SSLContexts.createDefault(),
                        split( System.getProperty( "https.protocols" ) ),
                        split( System.getProperty( "https.cipherSuites" ) ),
                        new DefaultHostnameVerifier( PublicSuffixMatcherLoader.getDefault() ) ) )
                .build() );

        connManager.setMaxTotal( maxConnTotal );
        connManager.setDefaultMaxPerRoute( maxConnPerRoute );

        return ( certificateLocation != null ?
            HttpAsyncClients.custom()
                .setSSLContext( createSSLContext( certificateLocation, certificatePassword ) )
            : HttpAsyncClients.custom() )
            .setMaxConnPerRoute( maxConnPerRoute )
            .setConnectionManager( connManager )
            .setMaxConnTotal( maxConnTotal )
            .setKeepAliveStrategy( DefaultConnectionKeepAliveStrategy.INSTANCE )
            .setDefaultRequestConfig( RequestConfig
                .custom()
                .setRedirectsEnabled( redirectsEnabled )
                .setCookieSpec( cookieSpec )
                .build() )
            .setDefaultCookieStore( basicCookieStore );
    } catch( IOReactorException e ) {
        throw new UncheckedIOException( e );
    }
}

/**
 * Create connection manager for asynchronous http client.
 *
 * @return Connection manager for asynchronous http client.
 * @throws IOReactorException in case if a non-recoverable I/O error.
 */
protected NHttpClientConnectionManager createNHttpClientConnectionManager() throws IOReactorException {
    ConnectingIOReactor ioReactor =
            new DefaultConnectingIOReactor(IOReactorConfig.custom()
                    .setSoTimeout(this.config.getSocketTimeoutInMillis()).setTcpNoDelay(true).build());
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(ioReactor);
    connectionManager.setDefaultMaxPerRoute(this.config.getMaxConnections());
    connectionManager.setMaxTotal(this.config.getMaxConnections());
    return connectionManager;
}

@SuppressWarnings("unused")
protected TopicBasedSelectorBot(String _targetHost, String _userId, String _topic) throws IOReactorException {
    this(_targetHost, _userId, _topic, ResultsType.RANDOM_20);
}

public FiberApacheHttpClientRequestExecutor(final Validator<CloseableHttpResponse> resValidator, final int maxConnections, final int timeout) throws IOReactorException {
  this(resValidator, maxConnections, timeout, Runtime.getRuntime().availableProcessors());
}

public FiberApacheHttpClientRequestExecutor(final Validator<CloseableHttpResponse> resValidator, final int maxConnections) throws IOReactorException {
  this(resValidator, maxConnections, 0);
}

public FiberApacheHttpClientRequestExecutor(final int maxConnections) throws IOReactorException {
  this(null, maxConnections, 0);
}