Use a thread pool for HTTP dispatch, especially faster with compression
@@ -28,6 +28,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
@@ -51,7 +52,8 @@ class ApacheCluster implements Cluster {
private final boolean gzip;
private int someNumber = 0;
- private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(t -> new Thread(t, "request-timeout-thread"));
+ private final ExecutorService dispatchExecutor = Executors.newFixedThreadPool(8, t -> new Thread(t, "request-dispatch-thread"));
+ private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor(t -> new Thread(t, "request-timeout-thread"));
ApacheCluster(FeedClientBuilderImpl builder) throws IOException {
for (int i = 0; i < builder.connectionsPerEndpoint; i++)
@@ -77,6 +79,7 @@ class ApacheCluster implements Cluster {
Endpoint endpoint = leastBusy;
endpoint.inflight.incrementAndGet();
+ dispatchExecutor.execute(() -> {
try {
SimpleHttpRequest request = new SimpleHttpRequest(wrapped.method(), wrapped.path());
request.setScheme(endpoint.url.getScheme());
@@ -100,13 +103,14 @@ class ApacheCluster implements Cluster {
@Override public void cancelled() { vessel.cancel(false); }
});
long timeoutMillis = wrapped.timeout() == null ? 200_000 : wrapped.timeout().toMillis() * 11 / 10 + 1_000;
- Future<?> cancellation = executor.schedule(() -> { future.cancel(true); vessel.cancel(true); }, timeoutMillis, TimeUnit.MILLISECONDS);
+ Future<?> cancellation = timeoutExecutor.schedule(() -> { future.cancel(true); vessel.cancel(true); }, timeoutMillis, TimeUnit.MILLISECONDS);
vessel.whenComplete((__, ___) -> cancellation.cancel(true));
}
catch (Throwable thrown) {
vessel.completeExceptionally(thrown);
}
vessel.whenComplete((__, ___) -> endpoint.inflight.decrementAndGet());
+ });
}
private byte[] gzipped(byte[] content) throws IOException{
@@ -120,6 +124,7 @@ class ApacheCluster implements Cluster {
@Override
public void close() {
Throwable thrown = null;
+ dispatchExecutor.shutdownNow().forEach(Runnable::run);
for (Endpoint endpoint : endpoints) {
try {
endpoint.client.close();
@@ -129,7 +134,7 @@ class ApacheCluster implements Cluster {
else thrown.addSuppressed(t);
}
}
- executor.shutdownNow().forEach(Runnable::run);
+ timeoutExecutor.shutdownNow().forEach(Runnable::run);
if (thrown != null) throw new RuntimeException(thrown);
}
