[case39]聊聊jdk httpclient的executor
序
本文主要研究一下jdk httpclient的executor
HttpClientImpl
java.net.http/jdk/internal/net/http/HttpClientImpl.java
private HttpClientImpl(HttpClientBuilderImpl builder, SingleFacadeFactory facadeFactory) { id = CLIENT_IDS.incrementAndGet(); dbgTag = "HttpClientImpl(" + id +")"; if (builder.sslContext == null) { try { sslContext = SSLContext.getDefault(); } catch (NoSuchAlgorithmException ex) { throw new InternalError(ex); } } else { sslContext = builder.sslContext; } Executor ex = builder.executor; if (ex == null) { ex = Executors.newCachedThreadPool(new DefaultThreadFactory(id)); isDefaultExecutor = true; } else { isDefaultExecutor = false; } delegatingExecutor = new DelegatingExecutor(this::isSelectorThread, ex); facadeRef = new WeakReference<>(facadeFactory.createFacade(this)); client2 = new Http2ClientImpl(this); cookieHandler = builder.cookieHandler; connectTimeout = builder.connectTimeout; followRedirects = builder.followRedirects == null ? Redirect.NEVER : builder.followRedirects; this.userProxySelector = Optional.ofNullable(builder.proxy); this.proxySelector = userProxySelector .orElseGet(HttpClientImpl::getDefaultProxySelector); if (debug.on()) debug.log("proxySelector is %s (user-supplied=%s)", this.proxySelector, userProxySelector.isPresent()); authenticator = builder.authenticator; if (builder.version == null) { version = HttpClient.Version.HTTP_2; } else { version = builder.version; } if (builder.sslParams == null) { sslParams = getDefaultParams(sslContext); } else { sslParams = builder.sslParams; } connections = new ConnectionPool(id); connections.start(); timeouts = new TreeSet<>(); try { selmgr = new SelectorManager(this); } catch (IOException e) { // unlikely throw new InternalError(e); } selmgr.setDaemon(true); filters = new FilterFactory(); initFilters(); assert facadeRef.get() != null; }
- 这里如果HttpClientBuilderImpl的executor为null,则会创建Executors.newCachedThreadPool(new DefaultThreadFactory(id))
HttpClientImpl.sendAsync
java.net.http/jdk/internal/net/http/HttpClientImpl.java
@Override public <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest userRequest, BodyHandler<T> responseHandler) { return sendAsync(userRequest, responseHandler, null); } @Override public <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest userRequest, BodyHandler<T> responseHandler, PushPromiseHandler<T> pushPromiseHandler) { return sendAsync(userRequest, responseHandler, pushPromiseHandler, delegatingExecutor.delegate); } private <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest userRequest, BodyHandler<T> responseHandler, PushPromiseHandler<T> pushPromiseHandler, Executor exchangeExecutor) { Objects.requireNonNull(userRequest); Objects.requireNonNull(responseHandler); AccessControlContext acc = null; if (System.getSecurityManager() != null) acc = AccessController.getContext(); // Clone the, possibly untrusted, HttpRequest HttpRequestImpl requestImpl = new HttpRequestImpl(userRequest, proxySelector); if (requestImpl.method().equals("CONNECT")) throw new IllegalArgumentException("Unsupported method CONNECT"); long start = DEBUGELAPSED ? System.nanoTime() : 0; reference(); try { if (debugelapsed.on()) debugelapsed.log("ClientImpl (async) send %s", userRequest); // When using sendAsync(...) we explicitly pass the // executor's delegate as exchange executor to force // asynchronous scheduling of the exchange. // When using send(...) we don't specify any executor // and default to using the client's delegating executor // which only spawns asynchronous tasks if it detects // that the current thread is the selector manager // thread. This will cause everything to execute inline // until we need to schedule some event with the selector. Executor executor = exchangeExecutor == null ? this.delegatingExecutor : exchangeExecutor; MultiExchange<T> mex = new MultiExchange<>(userRequest, requestImpl, this, responseHandler, pushPromiseHandler, acc); CompletableFuture<HttpResponse<T>> res = mex.responseAsync(executor).whenComplete((b,t) -> unreference()); if (DEBUGELAPSED) { res = res.whenComplete( (b,t) -> debugCompleted("ClientImpl (async)", start, userRequest)); } // makes sure that any dependent actions happen in the CF default // executor. This is only needed for sendAsync(...), when // exchangeExecutor is non-null. if (exchangeExecutor != null) { res = res.whenCompleteAsync((r, t) -> { /* do nothing */}, ASYNC_POOL); } return res; } catch(Throwable t) { unreference(); debugCompleted("ClientImpl (async)", start, userRequest); throw t; } }
- 这里如果是sendAsync的话,executor参数传递的是delegatingExecutor.delegate;如果是同步的send方法,则executor传的值是null
- 这里创建了一个MultiExchange,然后调用mex.responseAsync(executor).whenComplete((b,t) -> unreference()),这里使用了executor
MultiExchange.responseAsync
java.net.http/jdk/internal/net/http/MultiExchange.java
public CompletableFuture<HttpResponse<T>> responseAsync(Executor executor) { CompletableFuture<Void> start = new MinimalFuture<>(); CompletableFuture<HttpResponse<T>> cf = responseAsync0(start); start.completeAsync( () -> null, executor); // trigger execution return cf; } private CompletableFuture<HttpResponse<T>> responseAsync0(CompletableFuture<Void> start) { return start.thenCompose( v -> responseAsyncImpl()) .thenCompose((Response r) -> { Exchange<T> exch = getExchange(); return exch.readBodyAsync(responseHandler) .thenApply((T body) -> { this.response = new HttpResponseImpl<>(r.request(), r, this.response, body, exch); return this.response; }); }); }
- 可以看到这里使用的是CompletableFuture的completeAsync方法(
注意这个方法是java9才有的
),executor也是在这里使用的 - 由于默认是使用Executors.newCachedThreadPool创建的executor,要注意控制并发数及任务执行时间,防止线程数无限制增长过度消耗系统资源
/** * Creates a thread pool that creates new threads as needed, but * will reuse previously constructed threads when they are * available, and uses the provided * ThreadFactory to create new threads when needed. * * @param threadFactory the factory to use when creating new threads * @return the newly created thread pool * @throws NullPointerException if threadFactory is null */ public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), threadFactory); }
RejectedExecutionException
- 实例代码
@Test public void testAsyncPool(){ ThreadPoolExecutor executor = ThreadPoolBuilder.fixedPool() .setPoolSize(2) .setQueueSize(5) .setThreadNamePrefix("test-") .build(); List<CompletableFuture<String>> futureList = IntStream.rangeClosed(1,100) .mapToObj(i -> new CompletableFuture<String>()) .collect(Collectors.toList()); futureList.stream() .forEach(future -> { future.completeAsync(() -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e1) { e1.printStackTrace(); } return "message"; },executor); }); CompletableFuture.allOf(futureList .toArray(new CompletableFuture<?>[futureList.size()])) .join(); }这里创建的是fixedPool,指定queueSize为5
- 日志输出
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.CompletableFuture$AsyncSupply@76b10754 rejected from java.util.concurrent.ThreadPoolExecutor@2bea5ab4[Running, pool size = 2, active threads = 2, queued tasks = 5, completed tasks = 0] at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055) at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825) at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355) at java.base/java.util.concurrent.CompletableFuture.completeAsync(CompletableFuture.java:2591)可以看到线程池队列大小起到了限制作用
小结
jdk httpclient的executor在进行异步操作的时候使用,默认创建的是使用Executors.newCachedThreadPool创建的executor,其线程池大小是Integer.MAX_VALUE,因此在使用的时候要注意,最好是改为有界队列,然后再加上线程池的监控。
doc
相关推荐
Dullonjiang 2020-07-05
adayan0 2020-05-19
yanqianglifei 2020-03-03
Oeljeklaus 2020-02-18
Dullonjiang 2020-01-23
Notzuonotdied 2020-01-21
YarnSup 2020-01-10
hooopo 2014-07-12
zhixingheyitian 2019-12-06
yanqianglifei 2019-10-23
weeniebear 2015-10-16
chenchuanwen 2015-07-28
finalcola 2015-07-25
农村外出务工男 2015-05-28
lzzyok 2019-06-29
purpen 2013-01-31