netty - 在 ThreadPoolExecutors 中优先处理 Netty 任务

Question

我在 netty 中实现了一个服务器，它处理用户请求、与中间件对话并发送响应。与到中间件的往返相比，I/O 预计可以忽略不计，因此为了最大限度地减少阻塞，我在 OrderedMemoryAwareThreadPoolExecutor 之上的管道中有一个 ExecutionHandler。到目前为止没有问题。

我正在研究服务器在重负载下的行为。从过去使用我们协议(protocol)的经验来看，我们往往会被意外的 DOS 攻击所淹没——最常见的是，用户的脚本陷入无限循环或类似情况。理想情况下，一旦他们的 channel 超过一定的使用阈值，我们就可以取消他们的 channel 的优先级，这样其他用户的服务就不会受到影响。

我实现了一个简单的 ThreadPoolExecutor，它使用 PriorityBlockingQueue 并根据从我们自己的 Session 类(附加到 ChannelHandler 中的上下文)提取的数据设置优先级。同样，到目前为止没有问题。

当尝试利用 netty 的内置 ThreadPoolExecutors 的排序和内存感知时，困难就来了。理想情况下，MyThreadPoolExecutor 可以扩展 OrderedMemoryAwareThreadPoolExecutor 并连接到优先队列的东西中。唉，这是不可能的，原因有两个:私有(private)和最终。更详细地说:

a) ThreadPoolExecutor.workQueue 可以在其构造函数中设置，但 MemoryAwareThreadPoolExecutor 将其硬编码为 LinkedTransferQueue，并且不会将其公开给其子 OrderedMemoryAwareThreadPoolExecutor(即 MyThreadPoolExecutor 无权设置它)。如果有必要，这可以通过一些基于反射的私有(private)字段调整来克服。

b) 我希望能够覆盖 MyThreadPoolExecutor.doUnorderedExecute()，以便我可以插入优先级处理并构造必要的对象，但它被声明为 final。调用它的代码不需要更改。

结果是为了保留所有漂亮的网络功能，但使用优先级队列，我必须复制'n'粘贴 OrderedMemoryAwareThreadPoolExecutor 和 MemoryAwareThreadPoolExecutor，调整每行几行，然后从那里扩展。这对我来说并不是很好的编码习惯!即使考虑到它也会敲响警钟。

现在问几个问题:

1)我解决了错误的问题吗？我是否完全为我想要实现的目标而吠叫错误的树？

2)如果没有，有没有比上面讨论的更好的方法？

3)上述方法带来了总服务器负载始终处于容量状态的去优先级任务的饥饿风险。对于“顽皮”的用户，我准备容忍这种情况，但是一旦他们恢复正常状态，他们现有的任务仍然会挨饿，并且为了保持排序，必须在他们后面添加任何新的、更高优先级的任务。您对如何最好地处理这个问题有什么建议吗？ (企业不允许禁止用户。)

4)这是一半的问题，一半的反馈。 OrderedMemoryAwareThreadPoolExecutor 的 netty 文档有一个方便的线程 X 和 Y 图表 - 大概这些线程池在 ThreadPoolExecutor 而不是 I/O 工作线程？可能值得更清楚地说明这一点。此外，当不使用 ExecutionHandler 时，每个 channel 都绑定(bind)到单个 I/O 工作线程 - 在 ExecutionHandler 之后仍然是这种情况吗？即，任务添加到 ExecutionHandler 的顺序是否保证与它们​​到达 Channel 的顺序相同？如果是这种情况，那么我看不到 MemoryAwareThreadPoolExecutor 文档中的线程 X 如何在事件 1 之前处理事件 2 - 我接受这里不同的线程可以按任何顺序完成工作，但我看不出工作如何乱序分配给同一个线程(它从工作队列中弹出)。 ExecutionHandler 中的文档暗示了这一点，但会从更多细节中受益。

非常感谢您的阅读，非常感谢任何帮助。

Answer 1

1)不，你的想法很好。只是OrderedMemoryAwareThreadPoolExecutor缺少这样的功能。你愿意file an issue ?

2) 我只会 fork OrderedMemoryAwareThreadPoolExecutor ，简化它，并添加优先队列。这样，您可以更好地控制队列项目(事件)的处理方式。

3)您可以有两个队列，而不是使用优先级队列——一个用于高优先级项目，另一个用于低优先级项目。线程可以首先处理高优先级队列，但您可以控制循环，使其不会在那里停留太久。

4) 是的，它们是来自 ThreadPoolExecutor 的线程.如果不清楚，我们应该更新我们的文档。请随时提交问题或通过 fork 直接贡献。