如何在 Python 中接收 UDP 数据包而不丢弃答案

【问题标题】：How to receive UDP packets in Python without dropping如何在 Python 中接收 UDP 数据包而不丢弃
【发布时间】：2014-10-25 09:42:28
【问题描述】：

我正在编写简单的软件来解析 MPEG-TS 流以检查 CC（一致性计数器）以查看是否有任何数据包被丢弃。当我针对文件运行脚本时，它可以完美运行。但是在 UDP 流上使用它时会显示丢失（其他软件未确认）：

很简单：

while True:
    received = sock.recv(7 * 188)
    parsepacket(received)

为了清楚起见，我省略了 parsepacket 函数。就是用bitstring逐位解析数据包。

据我了解，当我运行 parsepacket 函数时，其他 UDP 数据包会被忽略，因为我做 sock.recv 的速度不够快（在 5 Mbps 流上，它应该每秒解析大约 500 个数据包）。

我尝试使用sock.makefile，但没有运气。我得到相同的结果。

我的想法是让接收线程在后台运行，另一个线程不断解析它接收到的内容。但老实说，我不知道怎么做（除了把它全部放在内存中会很快用完）。

【问题讨论】：

为什么使用 UDP 协议进行流式传输。这似乎很反直觉。现在您需要手动实现缺少的功能，例如缓冲和排序。
我正在分析来自我们的 IPTV 提供商的 MPEG-TS 流，它们以 UDP 多播的形式提供。我的另一个想法是在 python 中使用线程，但不幸的是它没有帮助。我仍然收到丢包。

标签： python sockets udp mpeg2-ts

【解决方案1】：

如果看不到您的脚本，或者您实际上是如何执行这些操作的，我们将无法为您提供帮助。但是，如果您使用 Python 的 twisted 库，这将是微不足道的。这是一个简单的例子，取自他们的examples page：

from twisted.internet.protocol import DatagramProtocol
from twisted.internet import reactor


class MulticastPingPong(DatagramProtocol):

    def startProtocol(self):
        """
        Called after protocol has started listening.
        """
        # Set the TTL>1 so multicast will cross router hops:
        self.transport.setTTL(5)
        # Join a specific multicast group:
        self.transport.joinGroup("228.0.0.5")

    def datagramReceived(self, datagram, address):
        print "Datagram %s received from %s" % (repr(datagram), repr(address))
        if datagram == "Client: Ping":
            # Rather than replying to the group multicast address, we send the
            # reply directly (unicast) to the originating port:
            self.transport.write("Server: Pong", address)


# We use listenMultiple=True so that we can run MulticastServer.py and
# MulticastClient.py on same machine:
reactor.listenMulticast(8005, MulticastPingPong(),
                        listenMultiple=True)
reactor.run()

您可以在def datagramReceived 函数中解析每个数据包。

【讨论】：

【解决方案2】：

我的想法是让接收线程在后台运行，另一个线程不断解析它收到的内容。但老实说我没有想法如何（除了把它全部放在内存中，这会用完快）。

这可能是最好的方法，因为这样一来，当您的 python 代码忙于解析较早的数据包时，您的套接字的传入 UDP 数据包缓冲区（位于 TCP 堆栈中）就不会填满。

您可以通过生成第二个线程并使用synchronized FIFO Queue 将数据从网络处理线程发送到数据包解析线程来轻松实现该模式。一旦您的网络线程读取一个数据包，它就会将数据包放入队列中，然后返回读取下一个数据包。同时，您的处理线程在同一个队列上调用 get() 的循环中运行，并且只有当 get() 返回一个要处理的数据包时才会唤醒。

需要注意的一点是，如果您的网络线程连续接收数据包的速度超过了处理线程可以解析它们的速度，会发生什么情况。在这种情况下，要么你的队列会无限制地变大（最终耗尽你机器的所有内存），要么（如果你为队列设置了最大大小），队列将达到其最大大小，而 put() 将引发完全异常。

【讨论】：