流式数据的理想Java数据结构

Question

我脑海中有一个特定的用例，但不能找出正确的数据结构来使用。

我有一个线程，它保持将对象流到HashMap中。一些类似于市场数据的东西，在那里你有一个很高的未知的滴答频率。

另一个线程，它不断读取此映射以获取更新的Price对象，并按键按不特定顺序进行查询。在给定的循环中，对于同一关键字，查询可以是多次的。读取和写入非常频繁，但是读取线程只对完全更新的最新可用数据感兴趣，并且在写入完成之前不一定会阻塞。

我希望您对这类用例的理想数据结构有什么想法。有比ConcurrentHashMap更好的实现吗？

谢谢

Answer 1

一种方法是写入时复制方案，如下所示：

public class Prices {
    private volatile Map<String, Integer> prices = Collections.emptyMap();

    public void putPrice(String ticker, int price) {
        HashMap<String, Integer> newPrices = new HashMap<String, Integer>(prices);
        newPrices.put(ticker, price);
        prices = newPrices;
    }

    public Integer getPrice(String ticker) {
        return prices.get(ticker);
    }
}

这对于gets有最小的开销-一次从易失性读取，然后是普通的散列查找。然而，对于puts来说，它有相当大的开销--创建一个全新的映射，外加对易失性的写入。如果您的读写比很高，这可能仍然是一个很好的权衡。

您可以通过仅在实际需要添加新条目时更改映射来改进这一点，而不是更新现有条目；您可以通过使用可变值来实现：

public class Prices {
    private volatile Map<String, AtomicInteger> prices = Collections.emptyMap();

    public void putPrice(String ticker, int price) {
        AtomicInteger priceHolder = prices.get(ticker);
        if (priceHolder != null) {
            priceHolder.set(price);
        }
        else {
            HashMap<String, AtomicInteger> newPrices = new HashMap<String, AtomicInteger>(prices);
            newPrices.put(ticker, new AtomicInteger(price));
            prices = newPrices;
        }
    }

    public Integer getPrice(String ticker) {
        AtomicInteger priceHolder = prices.get(ticker);
        if (priceHolder != null) return priceHolder.get();
        else return null;
    }
}

我不确定AtomicInteger的性能特征是什么；这可能比看起来慢。假设AtomicInteger不是不合理的慢，这应该是相当快的-它包括两次从易失性读取加上对每个get的普通散列查找，以及一次从易失性读取、散列查找和一次写入易失性以更新现有价格。它仍然涉及到复制地图以添加新的价格。然而，在典型的市场中，这种情况并不经常发生。

Answer 2

ConcurrentHashMap。来自Javadoc

一个哈希表，支持完全并发的检索和可调整的预期更新并发性。该类遵循与Hashtable相同的功能规范，并且包含与Hashtable的每个方法相对应的方法版本。但是，即使所有操作都是线程安全的，检索操作也不需要锁定，并且不支持以阻止所有访问的方式锁定整个表。在依赖于Hashtable线程安全性而不依赖于其同步细节的程序中，这个类完全可以与Hashtable互操作。

检索操作(包括get)通常不会阻塞，因此可能会与更新操作(包括put和remove)重叠。检索反映最近完成的更新操作的结果，这些操作在开始时保持不变。对于putAll和clear等聚合操作，并发检索可能只反映某些条目的插入或删除。类似地，迭代器和枚举返回元素，反映在迭代器/枚举创建时或之后的某个点上的哈希表的状态。

Answer 3

如果在更新数据时地图没有被修改(即没有puts或removes)，那么您甚至不需要像ConcurrentHashMap这样的同步地图。如果在程序执行期间不断有puts和removes，则需要同步这些调用。然而，当更新频率变得很高时(即在多线程程序中)，即使是ConcurrentHashMap也会开始抛出ConcurrentModificationExceptions。什么频率太高？您可能必须自己衡量，这取决于您的平台中的许多因素。

在这些情况下，我会尝试创建这样一种情况，即在程序执行期间，我不必在map中插入或删除，只有在数据流已停止时启动和关闭时才需要。如果这是不可能的，我使用普通的HashMap和优秀的数据结构CopyOnWriteArrayList的组合，并外部同步。我还没有测试过ConcurrentHashMap的局限性，但我不相信它适用于我自己的生产系统。

编辑: ConcurrentHashMap不会引起任何ConcurrentModificationExceptions，只有当你使用Collections.synchronizedMap时你才会有麻烦。