java - Java 中 UDP DatagramSocket 线程的 CPU 使用率过高

Question

我正在运行一个多线程 java 服务器应用程序，除其他外，它在 3 个不同的线程上从 3 个不同的多播源(端口)接收 UDP 数据包。

它在最新的双插槽 Redhat 机器上运行(总共 8 个核心(4 x 2 CPU)，无超线程)。

“top”命令显示 CPU 使用率为 250~300%。shift-H 显示 2 个线程的使用率约为 99%，1 个线程的使用率为 70%。快速线程 jstack 分析显示这些线程对应于我的 UDP 处理线程。

考虑到 CPU 速度与 UDP 消息速率(大约 300 条消息/秒，大约 250 字节的有效负载)，我对 CPU 使用率水平感到有点惊讶，我正在对此进行调查。有趣的是，第三个线程(对应于较低的 CPU 使用率)具有较低的数据速率(50~100 msg/s)

我添加了一些调试代码来测量花费最多时间的位置，它似乎位于 DatagramSocket 的“receive()”方法中:

_running    = true;
_buf        = new byte[300];
_packet     = new DatagramPacket(_buf, _buf.length);

while(_running) {
    try {
        long t0 = System.nanoTime();
        _inSocket.receive(_packet);
        long t1 = System.nanoTime();
        this.handle(_packet);
        long t2 = System.nanoTime();
        long waitingAndReceiveTime = t1-t0;
        long handleTime = t2-t1;
        _logger.info("{} : {} : update : {} : {}", t1, _port, waitingAndReceiveTime, handleTime);
    }
    catch(Exception e) {
        _logger.error("Exception while receiving multicast packet", e);
    }
}

handleTime 平均为 4000ns，速度极快，并且不会对 CPU 使用率负责。waitingAndReceiveTime 要高得多，从大约 30,000ns 到几毫秒。我知道该方法是阻塞的，所以时间包括阻塞时间和接收时间。

我有几个问题:

1. 我怀疑有什么奇怪的事情是对的吗？
2. 我认为“receive()”是阻塞的，它不应该“浪费”CPU周期，所以等待部分不应该对高CPU使用率负责，对吧？
3. 是否有一种方法可以将阻塞时间的测量和接收方法中接收数据报的时间分开？
4. 什么可能导致 CPU 使用率如此高？

编辑:我使用了中断合并参数，将 rx-usecs 设置为 0，将 rx-frames 设置为 10。我现在可以看到以下内容:

• UPD 消息确实以 10 条为一组出现:对于每组，第一条消息有一个 LONG waitingAndReceiveTime (>= 1ms)，而接下来的 9 个 waitingAndReceiveTime 则要短得多(~2000ns)。 (handleTime是一样的)
• CPU 使用率降低了!前 2 个线程的占用率下降至 55% 左右。

还是不知道如何解决这个问题

Answer 1

这并不是一个真正的答案，但是:

有一件事我可以向你保证，这不是 Java 代码。我用Python做了一个多线程UDP服务器，它做了同样的事情，CPU使用率在3到4秒内跳到100%。我猜这确实与 UDP 本身有关，因为我也做了一个多线程 TCP 服务器，但它的 CPU 使用率勉强达到 10%。

代码如下:

import socket
from _thread import*
import threading
import time
def threaded(s,serverIP,serverPort):
    while True:
        try:
            d = s.recvfrom(128)
            data = d[0]
            addr = d[1]
            message= str(data)
            if (message== "b'1'"):
                time.sleep(5)
            s.sendto(str.encode(message) , addr)
            print(message)
        except:
            break
    s.close()

def Main():
    serverPort = 11000
    serverIP= "127.0.0.1"
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.bind((serverIP, serverPort))

    while True:
        start_new_thread(threaded, (s,serverIP,serverPort))
    s.close)

if __name__ == '__main__':
    Main()

注意:

如果您找到答案，请告诉我。祝你好运。