Python:无间隙跟踪最新的消息序列ID

Question

我正在用 Python 编写一个网络应用程序，它从服务器接收带编号的消息。消息的序列号在 1..N 范围内，可能会出现乱序。我想跟踪收到的最新消息，条件是到目前为止消息中没有间隙。

例如，

• 如果消息是 1,3,2，我会将 3 标记为收到的最新消息。
• 如果消息是 1,2,5,4，我会将 2 标记为收到的最新消息，因为我还没有收到 3。
• 一旦 3 进来，我会将 5 标记为最新收到的消息。

最有效的方法是什么？是否有一些数据结构或编程惯用语实现了解决此问题的算法？

Answer 1

我环顾四周，没有立即找到很好的答案。

这是我通过一个小的阻塞类来实现这一点的尝试。对于单个 handle_new_index，这在技术上可能是 O(N)，但 handle_new_index 操作的平均时间仍应保证恒定。

我认为时间复杂度不会变得更好，因为无论您做什么，都必须对某种数据结构进行插入。

对于数十亿个请求和非常广泛的分布，non_contiguous 集可能具有适度的内存占用。

import random

class gapHandler:
    def __init__(self):
        self.greatest = 0
        self.non_contiguous = set()

    def handle_new_index(self, message_index):
        """
        Called when a new numbered request is sent. Updates the current
        index representing the greatest contiguous request.

        """
        self.non_contiguous.add(message_index)
        if message_index == self.greatest + 1:
            self._update_greatest()

    def _update_greatest(self):
        done_updating = False
        while done_updating is False:
            next_index = self.greatest + 1
            if next_index in self.non_contiguous:
                self.greatest = next_index
                self.non_contiguous.remove(next_index)
            else:
                done_updating = True


def demo_gap_handler():
    """ Runs the gapHandler class through a mock trial. """
    gh = gapHandler()

    for block_id in range(20000):
        start = block_id*500 + 1
        end = (block_id + 1)*500 + 1

        indices = [x for x in range(start, end)]
        random.shuffle(indices)
        while indices:
            new_index = indices.pop()
            gh.handle_new_index(new_index)
            if new_index % 50 == 0:
                print(gh.greatest)


if __name__ == "__main__":
    demo_gap_handler()

下面是一些基本测试:

import unittest
import gaps


class testGaps(unittest.TestCase):

    def test_update_greatest(self):
        gh = gaps.gapHandler()
        gh.non_contiguous = set((2, 3, 4, 6))

        gh._update_greatest()
        self.assertEqual(gh.greatest, 0)

        gh.greatest = 1
        gh._update_greatest()
        self.assertEqual(gh.greatest, 4)

    def test_handle_new_index(self):
        gh = gaps.gapHandler()
        gh.non_contiguous = set((2, 3, 4, 6, 2000))

        gh.handle_new_index(7)
        self.assertEqual(gh.greatest, 0)

        gh.handle_new_index(1)
        self.assertEqual(gh.greatest, 4)

        gh.handle_new_index(5)
        self.assertEqual(gh.greatest, 7)

if __name__ == "__main__":
    unittest.main()