实现支持老化的集合的最佳方法

Question

我正在寻找有关使用Java处理以下问题集的最佳方法的想法。在这种情况下，最好强调性能。

假设我们有一些银行分行。每个分支都有一个带有传感器的保险库，当门的状态发生变化时，传感器会发送一条消息（在本例中使用UDP）回服务器。

在服务器上存在某种类型的集合，用于存储从传感器发送的每条消息。代码在获取消息时插入一个事件（“开门分支1”）。当传感器发送后续消息（“关门分支1”）时，消息将从集合中删除。每条消息都与时间戳一起存储在集合中。

我们想要的是当消息在收集中已超过指定的经过时间（比如2分钟）时调用该方法。在这个用例中“金库门已经打开2分钟以上，请给警察打电话”。

最明显的解决方案是一个睡眠2分钟的线程，唤醒并运行收集检查时间戳。看起来很简单但不确定它是否是解决问题的更有效的方法。它还需要并发收集，这不是问题。

在现实世界中，集合需要处理大约50K条消息或更少。

有关如何处理此问题的其他想法？在这种情况下，有没有什么课程可以提供帮助？

谢谢

Answer 1

你可以使用一个或多个DelayQueue s到实现这一目标。

public static class BankCheck implements Delayed { 
    private static final long delay = TimeUnit.NANOSECONDS.convert(2, TimeUnit.MINUTES); 
    private final long created = System.nanoTime(); 
    private final Bank bank; 

    public BankCheck(Bank bank) { 
     this.bank = bank; 
    } 

    public Bank getBank() { 
     return bank; 
    } 

    @Override 
    public int compareTo(Delayed o) { 
     if (o instanceof BankCheck) { 
      BankCheck bc = (BankCheck) o; 
      if (created == bc.created) { 
       return 0; 
      } else { 
       return created < bc.created ? -1 : 1; 
      } 
     } 

     long d = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS); 
     if (d == 0) { return 0; } 
     return d < 0 ? -1 : 1; 
    } 

    @Override 
    public long getDelay(TimeUnit unit) { 
     long elapsed = System.nanoTime() - created; 
     long remaining = delay - elapsed; 
     return remaining > 0 ? unit.convert(remaining, TimeUnit.NANOSECONDS) : 0; 
    } 
} 

BlockingQueue<BankCheck> queue = new DelayQueue<>(); 

Thread messageReceiver = new Thread(new Runnable() { 
    @Override 
    public void run() { 
     for(;;) { 
      if(messageReceived) { 
       queue.add(new BankCheck(getBankFromLastMessage())); 
      } 
     } 
    } 
}).start(); 
Thread bankChecker = new Thread(new Runnable() { 
    @Override 
    public void run() { 
     try { 
      for(;;) { 
       Bank b = queue.take().getBank(); 
       if(!hasBeenClosed(b) { 
        alertAuthorities(b); 
       } 
      } 
     } catch(InterruptedException e) { 
      // handle exception 
     } 
    } 
}).start(); 

Answer 2

对于每个“门开了”的消息，创建使用ScheduledExecutorService2分钟延迟的任务。如果在超时前收到“门关闭”消息，请取消该任务。

Answer 3

下面是一个使用LinkedHashSet的解决方案应该是快，至少在theoritical界的术语：

LinkedHashSet<Bank> openSet; 

void onOpen(Bank b) { 
    openSet.add(b); // O(1) 
} 

void onClose(Bank b) { 
    openSet.remove(b); // O(1) 
} 

void onceInAWhile() { 
    Time alertTime = getAlertTime(); // If a door has been opened longer than this time, alarm. 

    Iterator<Bank> it = openSet.iterator(); 
    while(it.hasNext()) { 
     Bank b = it.next(); 

     if(b.getTime() < alertTime) { 
      NickyMinaj.poundTheAlarm(b); 
     } else { 
      // The set is FIFO, so you can break early 
      // You'll never examine more than N+1 banks, where N is the number that have to sound the alarm (so that's pretty optimal). 
      break; 
     } 
    } 
} 

Answer 4

我使用这样的数据结构，在我的代码，我叫EvictionList。每隔X秒，它会从列表中删除一段时间内未触及的元素。我用它作为心跳监视器。我有大约100个元素，表现从来都不是问题。下面是它的一些伪代码，作为真正的代码是无法访问的第二个：

public class EvictionList<T> implements Runnable 
{ 
    private evictionTime = 30; // seconds before eviction 
    private ConcurrentHashMap<T, Long> list = new ConcurrentHashMap(); 

    public EvictionList() 
    { 
     new Thread(this).start(); 
    } 

    public void run() 
    { 
     while (keepRunning) 
     { 
      Thread.sleep(5000); // sleep 5 seconds 
      List<T> l = evict(); 
      // do your actions here 
     } 
    } 

    public void add(T t) 
    { 
     list.put(t, System.currentMillis()); 
    } 

    public void touch(T t) 
    { 
     list.put(t, System.currentMillis()); 
    } 

    public List<T> evict() 
    { 
     List<T> evicted = new ArrayList(); 
     for (Iterator<T> i=list.keySet().iterator(); i.hasNext();) 
     { 
      T t = i.next(); 
      if (list.get(t) < System.currentMillis() - evictionTime) 
       evicted.add(t); 
     } 
     return evict(); 
    } 

}