下面列出了怎么用org.apache.http.nio.reactor.ConnectingIOReactor的API类实例代码及写法，或者点击链接到github查看源代码。
public HttpAsyncClientFactory() {
try {
int ioThreadCount = Runtime.getRuntime().availableProcessors() * 8;
IOReactorConfig ioReactorConfig =
IOReactorConfig.custom().setIoThreadCount(ioThreadCount)
.setConnectTimeout(httpConfig.getConnectTimeout()).setSoTimeout(httpConfig.getReadTimeout()).build();
ConnectingIOReactor ioReactor =
new DefaultConnectingIOReactor(ioReactorConfig, new NamedThreadFactory("flower-http"));
this.connectionManager = new PoolingNHttpClientConnectionManager(ioReactor);
this.connectionManager.setMaxTotal(1024);
this.connectionManager.setDefaultMaxPerRoute(256);
initHttpClient();
} catch (Exception e) {
logger.error("", e);
}
}
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;
}
private NHttpClientConnectionManager getNHttpConnManager(Config config) throws IOException {
NHttpClientConnectionManager httpConnManager;
String connMgrStr = config.getString(HTTP_CONN_MANAGER);
switch (ApacheHttpClient.ConnManager.valueOf(connMgrStr.toUpperCase())) {
case POOLING:
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
PoolingNHttpClientConnectionManager poolingConnMgr = new PoolingNHttpClientConnectionManager(ioReactor);
poolingConnMgr.setMaxTotal(config.getInt(POOLING_CONN_MANAGER_MAX_TOTAL_CONN));
poolingConnMgr.setDefaultMaxPerRoute(config.getInt(POOLING_CONN_MANAGER_MAX_PER_CONN));
httpConnManager = poolingConnMgr;
break;
default:
throw new IllegalArgumentException(connMgrStr + " is not supported");
}
LOG.info("Using " + httpConnManager.getClass().getSimpleName());
return httpConnManager;
}
public void initialize(final NettyRpcProperties properties) throws Exception {
final CommonProperties cp = properties.getCommonProperties();
final CyclicBarrier barrier = new CyclicBarrier(2);
Executors.newCachedThreadPool().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
try {
IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
.setIoThreadCount(CommonProperties.CPUS * 2)
.setConnectTimeout(Integer.parseInt(cp.getHttpConnectTimeout()))
.setSoTimeout(Integer.parseInt(cp.getHttpSocketTimeout()))
.setSndBufSize(Integer.parseInt(cp.getHttpSendBufSize()))
.setRcvBufSize(Integer.parseInt(cp.getHttpRcvBufSize()))
.setBacklogSize(Integer.parseInt(cp.getHttpBackLogSize()))
.setTcpNoDelay(true)
.setSoReuseAddress(true)
.setSoKeepAlive(true)
.build();
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
PoolingNHttpClientConnectionManager httpManager = new PoolingNHttpClientConnectionManager(ioReactor);
httpManager.setMaxTotal(Integer.parseInt(cp.getHttpMaxTotal()));
httpAsyncClient = HttpAsyncClients.custom().setConnectionManager(httpManager).build();
httpAsyncClient.start();
LOG.info("Create apache async client successfully");
barrier.await();
} catch (IOReactorException e) {
LOG.error("Create apache async client failed", e);
}
return null;
}
});
barrier.await();
}
/***
* 创建httpclient
*
* @param config
* 配置文件
* @return
* @throws IOReactorException
*/
public static HttpClient createHttpClient(OpenTSDBConfig config) throws IOReactorException {
Objects.requireNonNull(config);
ConnectingIOReactor ioReactor = initIOReactorConfig();
PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
RequestConfig requestConfig = initRequestConfig(config);
CloseableHttpAsyncClient httpAsyncClient = createPoolingHttpClient(requestConfig, connManager, config);
return new HttpClient(config, httpAsyncClient, initFixedCycleCloseConnection(connManager));
}
public void initialize() throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(2);
Executors.newCachedThreadPool().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
try {
IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
.setIoThreadCount(CPUS * 2)
.setConnectTimeout(Integer.parseInt(HTTPCLIENT_CONNCT_TIMEOUT_DEFAULT))
.setSoTimeout(Integer.parseInt(HTTPCLIENT_SOCKET_TIMEOUT_DEFAULT))
.setSndBufSize(Integer.parseInt(HTTPCLIENT_SEDBUFSIZE_DEFAULT))
.setRcvBufSize(Integer.parseInt(HTTPCLIENT_RCV_BUFSIZE_DEFAULT))
.setBacklogSize(Integer.parseInt(HTTPCLIENT_BACK_LOG_SIZE_DEFAULT))
.setTcpNoDelay(true)
.setSoReuseAddress(true)
.setSoKeepAlive(true)
.build();
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
PoolingNHttpClientConnectionManager httpManager = new PoolingNHttpClientConnectionManager(ioReactor);
httpManager.setMaxTotal(Integer.parseInt(HTTPCLIENT_MAX_TOTAL_DEFAULT));
httpAsyncClient = HttpAsyncClients.custom().setConnectionManager(httpManager).build();
httpAsyncClient.start();
LOG.info("Create apache async client successfully");
isStarted=true;
barrier.await();
} catch (IOReactorException e) {
LOG.error("Create apache async client failed", e);
}
return null;
}
});
barrier.await();
}
public AsyncInternalClient(Config config){
this.config = config;
try {
IOReactorConfig ioReactorConfig = IOReactorConfig.custom().build();
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
cm.setMaxTotal(config.getMaxConnectCount());
cm.setDefaultMaxPerRoute(config.getMaxPerRoute());
httpClient = createHttpAsyncClient(config, cm);
httpClient.start();
} catch (IOReactorException e) {
throw new ClientException(e);
}
}
protected static ConnectingIOReactor createConnectingIOReactor(int ioThreadCount) {
IOReactorConfig ioReactorConfig = IOReactorConfig.custom().setIoThreadCount(ioThreadCount).setTcpNoDelay(true).build();
try {
return new DefaultConnectingIOReactor(ioReactorConfig);
} catch (IOReactorException ex) {
throw new IOReactorRuntimeException(ex);
}
}
public static CloseableHttpAsyncClient getCloseableHttpAsyncClient() throws IOReactorException {
if (closeableHttpAsyncClient == null) {
synchronized (WxMpTemplateMsgSender.class) {
if (closeableHttpAsyncClient == null) {
RequestConfig requestConfig = RequestConfig.custom()
.setConnectTimeout(-1)
.setSocketTimeout(-1)
.setConnectionRequestTimeout(-1)
.build();
//配置io线程
IOReactorConfig ioReactorConfig = IOReactorConfig.custom().
setIoThreadCount(Runtime.getRuntime().availableProcessors())
.setSoKeepAlive(true).setConnectTimeout(-1).setSoTimeout(-1)
.build();
//设置连接池大小
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
//最大连接数
connManager.setMaxTotal(5000);
//per route最大连接数
connManager.setDefaultMaxPerRoute(5000);
closeableHttpAsyncClient = HttpAsyncClients.custom().
setConnectionManager(connManager)
.setDefaultRequestConfig(requestConfig)
.build();
closeableHttpAsyncClient.start();
}
}
}
return closeableHttpAsyncClient;
}
public AsyncServiceClient(ClientConfiguration 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 = 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);
}
}
public void initialize(final ThunderProperties properties) throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(2);
Executors.newCachedThreadPool().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
try {
IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
.setIoThreadCount(ThunderConstant.CPUS)
.setConnectTimeout(properties.getInteger(ThunderConstant.APACHE_CONNECT_TIMEOUT_ATTRIBUTE_NAME))
.setSoTimeout(properties.getInteger(ThunderConstant.APACHE_SO_TIMEOUT_ATTRIBUTE_NAME))
.setSndBufSize(properties.getInteger(ThunderConstant.APACHE_SNDBUF_SIZE_ATTRIBUTE_NAME))
.setRcvBufSize(properties.getInteger(ThunderConstant.APACHE_RCVBUF_SIZE_ATTRIBUTE_NAME))
.setBacklogSize(properties.getInteger(ThunderConstant.APACHE_BACKLOG_SIZE_ATTRIBUTE_NAME))
.setTcpNoDelay(true)
.setSoReuseAddress(true)
.setSoKeepAlive(true)
.build();
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
PoolingNHttpClientConnectionManager httpManager = new PoolingNHttpClientConnectionManager(ioReactor);
httpManager.setMaxTotal(ThunderConstant.CPUS * properties.getInteger(ThunderConstant.APACHE_MAX_TOTAL_ATTRIBUTE_NAME));
httpAsyncClient = HttpAsyncClients.custom().setConnectionManager(httpManager).build();
httpAsyncClient.start();
LOG.info("Create apache async client successfully");
barrier.await();
} catch (IOReactorException e) {
LOG.error("Create apache async client failed", e);
}
return null;
}
});
barrier.await(properties.getLong(ThunderConstant.APACHE_CONNECT_TIMEOUT_ATTRIBUTE_NAME) * 2, TimeUnit.MILLISECONDS);
}
/**
* 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;
}
private ConnectingIOReactor ioReactor() throws Exception {
// Create I/O reactor configuration
Map<String, Object> asyncRestTemplateMap = (Map)configMap.get(ASYNC_REST_TEMPLATE);
Map<String, Object> reactorMap = (Map)asyncRestTemplateMap.get(REACTOR);
Integer ioThreadCount = (Integer)reactorMap.get(REACTOR_IO_THREAD_COUNT);
IOReactorConfig.Builder builder = IOReactorConfig.custom();
builder.setIoThreadCount(ioThreadCount == null? Runtime.getRuntime().availableProcessors(): ioThreadCount);
Integer connectTimeout = (Integer)reactorMap.get(REACTOR_CONNECT_TIMEOUT);
builder.setConnectTimeout(connectTimeout == null? DEFAULT_REACTOR_CONNECT_TIMEOUT: connectTimeout);
Integer soTimeout = (Integer)reactorMap.get(REACTOR_SO_TIMEOUT);
builder.setSoTimeout(soTimeout == null? DEFAULT_REACTOR_SO_TIMEOUT: soTimeout);
IOReactorConfig ioReactorConfig = builder.build();
return new DefaultConnectingIOReactor(ioReactorConfig);
}
private void initialize() {
if (initialized.getAndSet(true)) {
return;
}
IOReactorConfig.Builder config = createConfig();
// params.setParameter(CoreProtocolPNames.USER_AGENT, "jsonrpc4j/1.0");
final ConnectingIOReactor ioReactor = createIoReactor(config);
createSslContext();
int socketBufferSize = Integer.getInteger("com.googlecode.jsonrpc4j.async.socket.buffer", 8 * 1024);
final ConnectionConfig connectionConfig = ConnectionConfig.custom().setBufferSize(socketBufferSize).build();
BasicNIOConnFactory nioConnFactory = new BasicNIOConnFactory(sslContext, null, connectionConfig);
pool = new BasicNIOConnPool(ioReactor, nioConnFactory, Integer.getInteger("com.googlecode.jsonrpc4j.async.connect.timeout", 30000));
pool.setDefaultMaxPerRoute(Integer.getInteger("com.googlecode.jsonrpc4j.async.max.inflight.route", 500));
pool.setMaxTotal(Integer.getInteger("com.googlecode.jsonrpc4j.async.max.inflight.total", 500));
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
HttpAsyncRequestExecutor protocolHandler = new HttpAsyncRequestExecutor();
IOEventDispatch ioEventDispatch = new DefaultHttpClientIODispatch(protocolHandler, sslContext, connectionConfig);
ioReactor.execute(ioEventDispatch);
} catch (InterruptedIOException ex) {
System.err.println("Interrupted");
} catch (IOException e) {
System.err.println("I/O error: " + e.getMessage());
}
}
}, "jsonrpc4j HTTP IOReactor");
t.setDaemon(true);
t.start();
HttpProcessor httpProcessor = new ImmutableHttpProcessor(new RequestContent(), new RequestTargetHost(), new RequestConnControl(), new RequestUserAgent(), new RequestExpectContinue(false));
requester = new HttpAsyncRequester(httpProcessor, new DefaultConnectionReuseStrategy());
}
ConnectingIOReactor createIOReactor() throws IOReactorException {
return new DefaultConnectingIOReactor(createIoReactorConfig());
}
ConnectingIOReactor createIOReactor() throws IOReactorException {
return new DefaultConnectingIOReactor(createIOReactorConfig());
}
public static void main(String[] args) throws Exception {
ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
cm.setMaxTotal(100);
CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(cm).build();
try {
httpclient.start();
// create an array of URIs to perform GETs on
String[] urisToGet = { "http://hc.apache.org/", "http://hc.apache.org/httpcomponents-core-ga/",
"http://hc.apache.org/httpcomponents-client-ga/", };
IdleConnectionEvictor connEvictor = new IdleConnectionEvictor(cm);
connEvictor.start();
final CountDownLatch latch = new CountDownLatch(urisToGet.length);
for (final String uri : urisToGet) {
final HttpGet httpget = new HttpGet(uri);
httpclient.execute(httpget, new FutureCallback<HttpResponse>() {
@Override
public void completed(final HttpResponse response) {
latch.countDown();
System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());
}
@Override
public void failed(final Exception ex) {
latch.countDown();
System.out.println(httpget.getRequestLine() + "->" + ex);
}
@Override
public void cancelled() {
latch.countDown();
System.out.println(httpget.getRequestLine() + " cancelled");
}
});
}
latch.await();
// Sleep 10 sec and let the connection evictor do its job
Thread.sleep(20000);
// Shut down the evictor thread
connEvictor.shutdown();
connEvictor.join();
} finally {
httpclient.close();
}
}