您可以通过为每个上游服务器保留一个Queue<Promise> 来解决此问题，这将承诺与响应一起标记为完整。
问题的解决方案类似于this answer，但我们需要在流中编辑一些对流水线的支持，以便正确处理多个路由。
假设你的用例是这样的：
+------------+ +------------+
| | | |
| client |+--------+ +----->| upstream |
| 1 | | | | 1 |
+------------+ | | +------------+
| |
+------------+ | +----------+ | +------------+
| | | | this | | | |
| client +---------+------>| server |-----+----->| upstream |
| 2 | | | | | | 2 |
+------------+ | +----------+ | +------------+
| |
+------------+ | | +------------+
| | | | | |
| client +---------+ +----->| upstream |
| 3 | | 3 |
+------------+ +------------+
我们在实施时需要确保满足以下几点：
- 如果我们允许对请求进行流水线处理（建议提高性能），我们需要一种使用例如序列号来识别请求的方法，或者我们需要对数据包进行分解以按照正确的顺序进入。
- 如果上游超时，我们需要一种向客户端发送超时响应的方法
我们将假设协议没有任何数据包序列号，因为这是最难解决的情况。
假设“请求处理程序服务”是在考虑支持大量请求的情况下编写的，它具有可以通过一些回调调用的方法。
如果您的“请求处理程序服务”具有如下所示的 api 实现：
public interface RequestHandlerService {
public default Future<Response> callMethod(Command cmd) {
return callMethod(cmd, promise);
}
public Future<Response> callMethod(Command cmd, Promise promise);
}
public interface ServerSelector {
public RequestHandlerService selectNextServer(Command cmd);
}
public interface Command {
// ....
}
public interface Response {
// ....
}
上面的系统足够通用，可以用于几乎所有类型的api，一种快速实现的方法，即使上面的系统不支持传递回调类，如this所示答案是为 Netty 连接做的。
基本处理程序骨架
我们可以制作以下处理程序来处理来自客户端的请求：
public class UpstreamDispachHandler extends SimpleInboundHandler<String> {
private final static int MAX_PIPELINED_REQUESTS = 32;
private final ServerSelector servers;
private final ArrayDeque<Future<Response>> messageList =
new ArrayDeque<>(MAX_PIPELINED_REQUESTS);
protected ChannelHandlerContext ctx;
public UpstreamDispachHandler (ServerSelector servers) {
this.servers = servers;
}
public void channelRegistered(ChannelHandlerContext ctx) {
this.ctx = ctx;
}
// The following is called messageReceived(ChannelHandlerContext, Command) in Netty 5.0
@Override
public void channelRead0(ChannelHandlerContext ctx, Command msg) {
if(messageList.size() >= MAX_PIPELINED_REQUESTS) {
// Fast fail if the max requests is exceeded
ctx.writeAndFlush(new FailedResponse(msg));
return;
}
RequestHandlerService nextServer = servers.selectNextServer(msg);
if (nextServer == null) {
// Fast fail if the max requests is exceeded
ctx.writeAndFlush(new FailedResponse(msg));
return;
}
sendCommandUpstream(msg, nextServer);
}
}
上面的代码声明了构造函数和我们使用的变量，我们使用ArrayDeque作为响应的临时存储，以确保保持顺序。
我们现在定义sendCommandUpstream(Command, RequestHandlerService) 将请求添加到队列中，并将其传递到上游。
private void sendCommandUpstream(Command cmd, RequestHandlerService nextServer) {
synchronized(messageList) {
messageList.add(nextServer.callMethod(cmd, ctx.executor().newPromise()
.addListener(f->recalculatePendingReplies())));
}
}
我们之所以通过 messageList 进行同步，而不是使用支持内部同步的 Queue，是为了确保我们未来的一些操作能够正常工作。
我们现在处于软件客户端-服务器部分的最后一部分，确保回复以正确的顺序返回。为此，我们将peeking 在Queue 的前面，以检查Future isdone 是否存在。如果头部完成，我们可以remove()它，从中提取Response，并将其发送回客户端，确保我们在发送所有响应后flush()管道。出于性能原因，我们只在最后而不是在每个数据包之后刷新。
private void recalculatePendingReplies() {
boolean hasSendMessage = false;
boolean interruped = false;
synchronized(messageList) {
Future<Response> elm = messageList.peek();
while (elm != null && elm.isDone()) {
elm.remove();
Response result;
try {
while(true) {
try {
result = elm.get();
break;
} catch (InterruptedException e) {
interrupted = true;
}
}
} catch (ExecutionException e) {
// Do something better with e
e.printStackTrace();
result = new FailedResponse();
} catch (ExecutionException e) {
// Task is cancelled
result = new FailedResponse();
}
ctx.write(result);
hasSendMessage = true;
}
if(hasSendMessage)
ctx.flush();
}
if(interrupted)
Thread.currentThread().interrupt();
}
上述代码中使用了一些设计模式，例如布尔标志用于存储线程的中断状态，而不是吞没异常。即使在抛出 InterruptedException 的情况下（不应该发生），它也会得到妥善处理。
添加对请求超时的支持
有时，我们希望在协议中实现对超时的支持，例如，如果上游在 x 时间内没有响应。 HashedWheelTimer 非常适合这种操作，因为它的特性可以很容易地修改以适应应用程序的使用模式。
刻度持续时间
如“近似”所述，此计时器不会按时执行计划的 TimerTask。 HashedWheelTimer，在每个滴答声中，都会检查是否有任何 TimerTasks 落后于计划并执行它们。
您可以通过在构造函数中指定更小或更大的滴答持续时间来增加或减少执行时间的准确性。在大多数网络应用程序中，I/O 超时不需要准确。因此，默认的刻度持续时间为 100 毫秒，在大多数情况下您无需尝试不同的配置。
每个轮子的刻度（轮子尺寸）
HashedWheelTimer 维护一个名为“wheel”的数据结构。简单地说，轮子是一个 TimerTasks 的哈希表，其哈希函数是“任务的最后期限”。每个轮子的默认刻度数（即轮子的大小）为 512。如果要安排大量超时，可以指定一个更大的值。
我们在程序的顶部定义了以下一组常量：
private final static long TICK_DURATION = 100;
private final static int WHEEL_SIZE = 128;
private final static long DEFAULT_TASK_TIMEOUT = TICK_DURATION * DEFAULT_TASK_TIMEOUT;
private final static HashedWheelTimer timer = new HashedWheelTimer(
Executors.defaultThreadFactory(),
TICK_DURATION,
TimeUnit.MILLISECONDS,
WHEEL_SIZE);
然后我们修改我们的sendCommandUpstream 方法，在计时器中设置一个超时时间，所以任务cancel() 被调用。我们可以通过以下方式做到这一点：
private void sendCommandUpstream(Command cmd, RequestHandlerService nextServer) {
synchronized(messageList) {
Future<Response> r = nextServer.callMethod(cmd, ctx.executor().newPromise()
.addListener(f->recalculatePendingReplies()));
messageList.add(r);
timer.newTimeout((Timeout timeout) -> {
if (!r.isDone())
r.cancel(true);
}, DEFAULT_TASK_TIMEOUT, TimeUnit.MILLISECONDS);
}
}