优先处理ThreadPoolExecutors中的Netty任务

Question

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

我正在研究服务器在重负载下的行为。根据过去的协议经验，我们往往会被偶然的DOS攻击所淹没 - 通常是用户的脚本陷入无限循环或类似情况。理想情况下，我们可以在他们的频道超过某个使用率阈值后对频道进行优先排序，以便其他用户的服务不受影响。

我已经实现了一个简单的ThreadPoolExecutor，它使用PriorityBlockingQueue并基于从我们自己的Session类（连接到ChannelHandler中的上下文）中提取的数据设置优先级。再次，到目前为止没有问题。

尝试利用netty的内置ThreadPoolExecutors的排序和内存感知的困难。理想情况下，MyThreadPoolExecutor可以只扩展OrderedMemoryAwareThreadPoolExecutor并在优先级队列中连线。唉这是不可能的，原因有两个：私人和最终。更详细：

一）ThreadPoolExecutor.workQueue可以在构造函数中设置，但MemoryAwareThreadPoolExecutor硬编码这是一个LinkedTransferQueue，并且不公开这个到其子OrderedMemoryAwareThreadPoolExecutor（即MyThreadPoolExecutor不能访问设置它）。如果有必要，这可以通过基于反射的私人领域调整的丑陋克服。

b）我想能够重写MyThreadPoolExecutor.doUnorderedExecute（），这样我就可以插入优先级处理并构造必要的对象，但它被声明为final。调用它的代码不需要改变。

结果是，为了保持所有漂亮的netty功能，但使用优先级队列，我不得不复制'n'paste两个OrderedMemoryAwareThreadPoolExecutor和MemoryAwareThreadPoolExecutor，调整每一行的几行，然后从那里延伸。这不会让我成为好的编程习惯！即使考虑到它引发了警钟。

现在的几个问题：

1）我解决了错误的问题？我为了什么我想要完全吠叫完全错误的树？

2）如果不是，还有比上面讨论的更好的方法吗？

3）上述方法带来了总服务器负载一直处于容量状态的非优先级任务的饥饿风险。我已经准备好容忍'淘气'的用户，但是一旦它们恢复到正常状态，它们现有的任务仍然会挨饿，并且为了保持顺序，任何新的，更高优先级的任务都必须添加到它们后面。你有关于如何最好地处理这个问题的建议吗？ （禁止用户不被业务所允许。）

4）这是半问题，半反馈。 OrderedMemoryAwareThreadPoolExecutor的netty文档为线程X提供了一个方便的图表Y - 大概这些是ThreadPoolExecutor中汇集的线程而不是I/O工作线程？这可能值得更加清楚。另外，当不使用ExecutionHandler时，每个通道都绑定到单个I/O工作线程 - 当执行一个ExecutionHandler时，这仍然是这种情况吗？即是否将任务添加到ExecutionHandler的顺序保证与它们​​到达Channel的顺序相同？如果是这样的话，那么我看不出MemoryAwareThreadPoolExecutor的文档中的线程X如何在事件1之前处理事件2 - 我接受在这里不同的线程可以按任何顺序完成工作，但我看不到工作是如何进行的（它从workQueue中弹出）。 ExecutionHandler中的文档暗示了这一点，但会从更多的细节中受益。

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

Answer 1

您的'警钟'解决方案看起来很熟悉 - 只需从一个小池大小开始。

你似乎在说你需要一个慢速服务器的解决方案。 你可能想看看为什么你的服务器在负载下变慢。 - 线程争用（比赛）问题。 - 初始池大小 - GC配置

Answer 2

1）不，你的想法很好。这只是OrderedMemoryAwareThreadPoolExecutor缺乏这样的功能。你会file an issue？

2）我只是叉OrderedMemoryAwareThreadPoolExecutor，简化它，并添加优先级队列。这样，您就可以更好地控制队列项目（事件）的处理方式。

3）可以使用两个队列 - 一个用于高优先级项目，另一个用于低优先级项目，而不是使用优先级队列。该线程可以首先处理高优先级队列，但是您可以控制该循环，以防止循环时间过长。

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