java从高速数据流更新映射

Question

我有一个多线程Java应用程序，其中一个方法[update(key, value)]更新一个ConcurrentHashMap。对于每个密钥，接收到的值将超过可以放入映射中的值，因此，一旦更新了密钥，则应仅使用等待线程的最新值，然后再次更新映射。或者，在只有一个线程等待的情况下，可以使用某种类型的锁—最后一个到达锁的线程（有效地处理已经等待的线程）？重要的是，整个映射没有被锁定，这就是为什么我没有在正常的HashMap周围使用同步块的原因，因为即使有线程等待键a，只要没有线程已经在更新为B存储的值，就应该允许更新键B

更简洁地说，如何更新一个映射，其中键值对的接收速度比更新速度快，使用上一次接收的值作为下一次更新？因此，当A更新为1时，将接收到5、3、6、8的值，这意味着A的下一次更新将为8

Answer 1

这是一个困难的问题，困难的根源在于捕获更新到达的顺序

如果更新已经有相关的（细粒度）时间戳，那么解决方案相当简单：

1. 定义一个Value类，该类保存实际值和时间戳。它需要一个同步的setIfNewer(ActualValue v, Timestamp t)，如果提供的时间戳是最近的，它将更新实际值
2. 将映射定义为ConcurrentHashMap<Key, Value>
3. 使用putIfAbsent将值放入映射。如果putIfAbsent()返回非空值，请使用setIfNewer(...)更新它

请注意，这仅在地图更新能够长期保持时有效；i、 e.平均数据率不太高，无法应对

如果更新没有关联的时间戳，那么您就有问题了。如果您很难跟上更新，那么您将很难为更新添加准确反映到达时间的时间戳。这意味着更新有可能（实际上）被重新排序。（如果是这种情况，那么我认为问题是无法解决的……如果不改变问题，请参见下文。）

一些可能有效的方法：

• 进行一些分析/性能分析，找出瓶颈所在。它可能根本不在地图更新中。（毕竟ConcurrentHashMap的设计是高度可伸缩的。）

• 如果线程和键值之间有很强的相关性，则可以尝试1）使用每线程LRU映射消除每个线程中的更新，或2）使用每线程计数器而不是时间戳

• 您可以尝试根据键空间对映射进行分区

• 您可以尝试添加更多处理器和/或内存。。。取决于您的分析和监视报告的内容

• 您可以尝试根据键空间对整个应用程序进行分区。如果真正的问题是应用程序无法跟上，那么这可能是唯一可能的方法

Answer 2

怎么做

有一个相当简单的解决方案来实现一个sequencer，您添加的每个对象都需要一个长字段，该字段在构建时分配，与AtomicLong类似。getAndIncrement（）

更新看起来像，不需要同步

Class Value{
private static final AtomicLong sequencer = new AtomicLong()
final long seq = sequencer.getAndIncrement():
public boolean equals(Object o){
  //include seq as well :)
}
....
}
ConcurrentMap map;
for (Value cur;;){
    cur = map.get(key);
    if (cur==null){
        if (null==(cur=map.putIfAbsent(key, value))){
            break;
        }
    }           
    if (cur.seq>=value.seq){
        break;
    }
    if (map.replace(key, cur, value))
        break;
}