Skip to content

Commit bf74bfc

Browse files
author
yzhfd
committed
重构
1 parent 482db7c commit bf74bfc

1 file changed

Lines changed: 93 additions & 60 deletions

File tree

‎trpc-proto/trpc-proto-http/src/main/java/com/tencent/trpc/proto/http/server/AbstractHttpExecutor.java‎

Lines changed: 93 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -38,11 +38,14 @@
3838
import com.tencent.trpc.proto.http.common.RpcServerContextWithHttp;
3939
import com.tencent.trpc.proto.http.common.TrpcServletRequestWrapper;
4040
import com.tencent.trpc.proto.http.common.TrpcServletResponseWrapper;
41+
import java.io.IOException;
4142
import java.lang.reflect.Type;
4243
import java.net.InetSocketAddress;
4344
import java.nio.charset.StandardCharsets;
4445
import java.util.Enumeration;
4546
import java.util.Map;
47+
import java.util.concurrent.CompletableFuture;
48+
import java.util.concurrent.TimeoutException;
4649
import java.util.concurrent.atomic.AtomicBoolean;
4750
import java.util.concurrent.CompletionStage;
4851
import java.util.concurrent.CountDownLatch;
@@ -66,36 +69,33 @@ public abstract class AbstractHttpExecutor {
6669

6770
protected void execute(HttpServletRequest request, HttpServletResponse response,
6871
RpcMethodInfoAndInvoker methodInfoAndInvoker) {
69-
70-
AtomicBoolean isTimeout = new AtomicBoolean(false);
72+
AtomicBoolean responded = new AtomicBoolean(false);
7173
try {
72-
7374
DefRequest rpcRequest = buildDefRequest(request, response, methodInfoAndInvoker);
74-
75-
CountDownLatch countDownLatch = new CountDownLatch(1);
76-
75+
CompletableFuture<Void> completionFuture = new CompletableFuture<>();
7776
// use a thread pool for asynchronous processing
78-
invokeRpcRequest(methodInfoAndInvoker.getInvoker(), rpcRequest, countDownLatch, isTimeout);
79-
80-
// If the request carries a timeout, use this timeout to wait for the request to be processed.
81-
// If not carried, use the default timeout.
77+
invokeRpcRequest(methodInfoAndInvoker.getInvoker(), rpcRequest, completionFuture, responded);
8278
long requestTimeout = rpcRequest.getMeta().getTimeout();
8379
if (requestTimeout <= 0) {
8480
requestTimeout = methodInfoAndInvoker.getInvoker().getConfig().getRequestTimeout();
8581
}
86-
if (requestTimeout > 0 && !countDownLatch.await(requestTimeout, TimeUnit.MILLISECONDS)) {
87-
isTimeout.set(true);
88-
throw TRpcException.newFrameException(ErrorCode.TRPC_SERVER_TIMEOUT_ERR,
89-
"wait http request execute timeout");
82+
if (requestTimeout > 0) {
83+
try {
84+
completionFuture.get(requestTimeout, TimeUnit.MILLISECONDS);
85+
} catch (TimeoutException ex) {
86+
if (responded.compareAndSet(false, true)) {
87+
sendTimeoutResponse(request, response);
88+
}
89+
}
9090
} else {
91-
countDownLatch.await();
91+
completionFuture.get();
9292
}
93-
9493
} catch (Exception ex) {
9594
logger.error("dispatch request [{}] error", request, ex);
96-
doErrorReply(request, response, ex);
95+
if (responded.compareAndSet(false, true)) {
96+
doErrorReply(request, response, ex);
97+
}
9798
}
98-
9999
}
100100

101101
/**
@@ -110,58 +110,91 @@ protected void execute(HttpServletRequest request, HttpServletResponse response,
110110
/**
111111
* Request processing
112112
*
113-
* @param countDownLatch latch used to wait for the request processing
113+
* @param invoker the invoker
114+
* @param rpcRequest the rpc request
115+
* @param completionFuture the completion future
116+
* @param responded the responded flag
114117
*/
115-
private void invokeRpcRequest(ProviderInvoker<?> invoker, DefRequest rpcRequest, CountDownLatch countDownLatch,
116-
AtomicBoolean isTimeout) {
118+
private void invokeRpcRequest(ProviderInvoker<?> invoker, DefRequest rpcRequest,
119+
CompletableFuture<Void> completionFuture,
120+
AtomicBoolean responded) {
117121

118122
WorkerPool workerPool = invoker.getConfig().getWorkerPoolObj();
119-
120-
if (null == workerPool) {
121-
logger.error("dispatch rpcRequest [{}] error, workerPool is empty", rpcRequest);
122-
throw TRpcException.newFrameException(ErrorCode.TRPC_SERVER_NOSERVICE_ERR,
123-
"not found service, workerPool is empty");
123+
if (workerPool == null) {
124+
logger.error("Worker pool is not available");
125+
completionFuture.completeExceptionally(
126+
TRpcException.newFrameException(ErrorCode.TRPC_SERVER_NOSERVICE_ERR, "Worker pool not available")
127+
);
128+
return;
124129
}
125130

126131
workerPool.execute(() -> {
127-
128-
// Get the original http response
129-
HttpServletResponse response = getOriginalResponse(rpcRequest);
130-
131-
// Invoke the routing implementation method to handle the request.
132-
CompletionStage<Response> future = invoker.invoke(rpcRequest);
133-
future.whenComplete((result, t) -> {
134-
try {
135-
if (isTimeout.get()) {
136-
return;
132+
try {
133+
HttpServletResponse response = getOriginalResponse(rpcRequest);
134+
135+
CompletionStage<Response> rpcFuture = invoker.invoke(rpcRequest);
136+
137+
rpcFuture.whenComplete((result, throwable) -> {
138+
try {
139+
if (responded.get()) {
140+
return;
141+
}
142+
143+
if (throwable != null) {
144+
throw throwable;
145+
}
146+
147+
if (result.getException() != null) {
148+
throw result.getException();
149+
}
150+
151+
if (responded.compareAndSet(false, true)) {
152+
response.setStatus(HttpStatus.SC_OK);
153+
httpCodec.writeHttpResponse(response, result);
154+
response.flushBuffer();
155+
}
156+
157+
completionFuture.complete(null);
158+
} catch (Throwable t) {
159+
handleError(t, rpcRequest, response, responded, completionFuture);
137160
}
161+
});
138162

139-
// Throw the call exception, which will be handled uniformly by the exception handling program.
140-
if (t != null) {
141-
throw t;
142-
}
163+
} catch (Exception e) {
164+
handleError(e, rpcRequest, getOriginalResponse(rpcRequest), responded, completionFuture);
165+
}
166+
});
167+
}
143168

144-
// Throw a business logic exception, which will be handled uniformly
145-
// by the exception handling program.
146-
Throwable ex = result.getException();
147-
if (ex != null) {
148-
throw ex;
149-
}
169+
/**
170+
* Handle error
171+
*/
172+
private void handleError(Throwable t, DefRequest rpcRequest, HttpServletResponse response,
173+
AtomicBoolean responded, CompletableFuture<Void> completionFuture) {
174+
try {
175+
if (responded.compareAndSet(false, true)) {
176+
HttpServletRequest request = getOriginalRequest(rpcRequest);
177+
logger.warn("Request processing failed: {}", request.getRequestURI(), t);
178+
httpErrorReply(request, response,
179+
ErrorResponse.create(request, HttpStatus.SC_INTERNAL_SERVER_ERROR, t));
180+
}
181+
} finally {
182+
completionFuture.completeExceptionally(t);
183+
}
184+
}
150185

151-
// normal response
152-
response.setStatus(HttpStatus.SC_OK);
153-
httpCodec.writeHttpResponse(response, result);
154-
response.flushBuffer();
155-
} catch (Throwable e) {
156-
HttpServletRequest request = getOriginalRequest(rpcRequest);
157-
logger.warn("reply message error, channel: [{}], msg:[{}]", request.getRemoteAddr(), request, e);
158-
httpErrorReply(request, response,
159-
ErrorResponse.create(request, HttpStatus.SC_SERVICE_UNAVAILABLE, e));
160-
} finally {
161-
countDownLatch.countDown();
162-
}
163-
});
164-
});
186+
/**
187+
* Send timeout response
188+
*/
189+
private void sendTimeoutResponse(HttpServletRequest request, HttpServletResponse response) {
190+
try {
191+
response.setStatus(HttpStatus.SC_GATEWAY_TIMEOUT);
192+
response.getWriter().write("Request Timeout");
193+
response.flushBuffer();
194+
logger.warn("Request timeout: {} {}", request.getMethod(), request.getRequestURI());
195+
} catch (IOException e) {
196+
logger.error("Failed to send timeout response", e);
197+
}
165198
}
166199

167200
/**

0 commit comments

Comments
 (0)