限制传入消息以避免重复的最佳方法答案

【问题标题】：Best way to limit incoming messages to avoid duplicates限制传入消息以避免重复的最佳方法
【发布时间】：2013-11-10 22:00:27
【问题描述】：

我有一个接受包含 url 的消息的系统，如果消息中包含某些关键字，则会使用 url 作为参数进行 api 调用。

为了节省处理并保持我的最终演示效率。

我不希望在某个时间范围内提交重复的网址。

所以如果这个 url ---> http://instagram.com/p/gHVMxltq_8/ 进来并提交给 api

          url = incoming.msg['urls']
          url = urlparse(url)
          if url.netloc  == "instagram.com":                
            r = requests.get("http://api.some.url/show?url=%s"% url)

然后 3 秒后相同的 url 进来，我不希望它提交给 api。

我可以部署什么编程方法来消除/限制基于时间提交给 api 的重复消息？

使用 TIM PETERS 方法更新：

         limit = DecayingSet(86400)
         l = limit.add(longUrl)
         if l == False:
           pass
         else:
           r = requests.get("http://api.some.url/show?url=%s"% url)

这个 sn-p 在一个长时间运行的进程中，即通过 tcp 接受流式消息。

每次我传入相同的 url 时，l 每次都返回 True。

但是当我在解释器中尝试时一切都很好，当设置的时间没有到期时它返回 False。

这是否与脚本正在运行而集合被添加到这一事实有关？

实例问题？

【问题讨论】：

我不明白f 在你的代码的第三行做了什么。
不得不重新启动我的键盘搞砸了..检查编辑

标签： python algorithm parsing message-queue messaging

【解决方案1】：

也许有点矫枉过正，但我喜欢为这种事情创建一个新类。你永远不知道什么时候需求会变得更高级 ;-) 例如，

from time import time

class DecayingSet:
    def __init__(self, timeout): # timeout in seconds
        from collections import deque
        self.timeout = timeout
        self.d = deque()
        self.present = set()

    def add(self, thing):
        # Return True if `thing` not already in set,
        # else return False.
        result = thing not in self.present
        if result:
            self.present.add(thing)
            self.d.append((time(), thing))
        self.clean()
        return result

    def clean(self):
        # forget stuff added >= `timeout` seconds ago
        now = time()
        d = self.d
        while d and now - d[0][0] >= self.timeout:
            _, thing = d.popleft()
            self.present.remove(thing)

正如所写，每当尝试添加新事物时，它都会检查是否过期。也许这不是您想要的，但它应该是一张便宜的支票，因为deque 按添加顺序保存物品，所以如果没有物品过期，请立即退出。很多可能性。

为什么是deque？因为当项目数量变得不平凡时，deque.popleft() 比 list.pop(0) 快很多。

【讨论】：

这就是哇！这正是我想要的。而且我知道如何实现它，但我不知道在哪里/如何使用 clean 方法
基本上干净只是从集合中取出过期的东西，以使一个已经广泛增长的集合更快？？
@sirvon，你没有有使用 clean 方法 - 它作为调用 add() 的副作用运行。如果您愿意，可以显式运行它。如果它根本没有运行，那么任何东西都不会从集合中移除。因此，例如，在第一次.add()'ing "instagram.com" 之后，.add() 会永远在您再次尝试.add() "instagram.com" 时返回False。
你到底为什么要使用float("inf")？大声笑 ;-) 尝试使用一些合理的东西——我猜不出你的平台和你的 Python 版本与无穷大有什么关系。
如果您仍然遇到问题，请在代码中添加一些 prints 以查看发生了什么。从你说的我猜不出什么。

【解决方案2】：

假设您希望的时间间隔是 1 小时，请保留 2 个每小时递增的计数器，但它们之间的间隔为 30 分钟。一世。 e.计数器 A 转到 1, 2, 3, 4 11:17, 12:17, 13:17, 14:17，计数器 B 转到 1, 2, 3, 4 11:47, 12:47, 13:47, 14:47。

现在，如果一个链接进入并且两个计数器中的一个与之前的链接相同，则认为它是重复的。

这种方案相对于显式时间戳的好处是可以对 url+counterA 和 url+counterB 进行哈希运算，从而快速检查 url 是否存在

更新：您需要两个数据存储：一个是常规数据库表（慢），其列：(url, counterA, counterB) 和两个，一大块n 内存（快）。给定一个 url so.com、counterA 17 和 counterB 18，首先将 "17,so.com" 散列到 0 到 n - 1 的范围内，然后查看该地址的位是否打开。同样，哈希“18，so.com”并查看该位是否打开。

如果在任何一种情况下都没有打开该位，您可以确定它在一小时内是一个新的 URL，所以我们完成了（很快）。

如果在任何一种情况下都打开了该位，则在数据库表中查找 url 以检查它是否确实是该 url 或其他一些哈希到同一位的 URL。

进一步更新：布隆过滤器是此方案的扩展。

【讨论】：

你能提供一个例子吗..这个方法看起来很有趣，如果不是有点混淆抽象

【解决方案3】：

我建议保留最近使用的 URL 的内存缓存。像字典一样的东西：

urls = {}

然后对于每个 URL：

if url in urls and (time.time() - urls[url]) < SOME_TIMEOUT:
    # Don't submit the data
else:
    urls[url] = time.time()
    # Submit the data

【讨论】：