Инкрементно увеличивающиеся счетчики, хранящиеся в ConcurrentHashMap

Question

Я хотел бы собрать некоторые метрики из разных мест в веб-приложении. Для простоты все они будут счетчиками, и поэтому единственной операцией модификатора является увеличение их на 1.

Приращения будут параллельными и частыми. Чтение (сброс статистики) - редкая операция.

Я думал использовать ConcurrentHashMap . Вопрос в том, как увеличить счетчики правильно. Поскольку на карте нет операции «увеличения», мне нужно сначала прочитать текущее значение, увеличить его, а затем поместить новое значение на карту. Без дополнительного кода это не атомарная операция.

Возможно ли достичь этого без синхронизации (что противоречило бы цели ConcurrentHashMap )? Нужно ли смотреть на Гуава ?

Спасибо за любые указатели.

---

приписка
Есть связанный вопрос по SO ( Самый эффективный способ увеличить значение Map в Java ), но он сосредоточен на производительности, а не на многопоточности

UPDATE
Для тех, кто прибывает сюда через поиск по той же теме: помимо ответов ниже, есть полезная презентация , которая случайно затрагивает ту же тему. Смотрите слайды 24-33.

Answer 1

В Java 8:

ConcurrentHashMap<String, LongAdder> map = new ConcurrentHashMap<>();

map.computeIfAbsent("key", k -> new LongAdder()).increment();

Answer 2

Новый Guava AtomicLongMap (в выпуске 11) может удовлетворить эту потребность.

Answer 3

Ты довольно близко. Почему бы вам не попробовать что-то вроде ConcurrentHashMap<Key, AtomicLong>? Если ваши Key s (метрики) неизменны, вы могли бы даже просто использовать стандартный HashMap (они являются поточно-ориентированными, если они доступны только для чтения, но было бы неплохо сделать это явно с ImmutableMap от Google Collections или Collections.unmodifiableMap и т. Д.).

Таким образом, вы можете использовать map.get(myKey).incrementAndGet() для увеличения статистики.

Answer 4

Кроме перехода с AtomicLong, вы можете сделать обычную вещь с циклом:

private final ConcurrentMap<Key,Long> counts =
    new ConcurrentHashMap<Key,Long>();

public void increment(Key key) {
    if (counts.putIfAbsent(key, 1)) == null) {
        return;
    }

    Long old;
    do {
       old = counts.get(key);
    } while (!counts.replace(key, old, old+1)); // Assumes no removal.
}

(я не писал цикл do - while целую вечность.)

Для небольших значений Long, вероятно, будет "кэшировано". Для более длинных значений может потребоваться распределение. Но распределение на самом деле очень быстрое (и вы можете кешировать дальше) - зависит от того, что вы ожидаете, в худшем случае.

Answer 5

Я сделал тест для сравнения производительности LongAdder и AtomicLong.

LongAdder показала лучшую производительность в моем тесте: для 500 итераций с использованием карты размером 100 (10 одновременных потоков) среднее время для LongAdder составляло 1270 мс, а для AtomicLong - 1315 мс.

Answer 6

Следующий код работал для меня без синхронизации для подсчета частот слов

protected void updateFreqCountForText(Map<String, Integer> wordFreq, String line) {
            ConcurrentMap<String, Integer> wordFreqConc = (ConcurrentMap<String, Integer>) wordFreq;
            Pattern pattern = Pattern.compile("[a-zA-Z]+");
            Matcher matcher  = pattern.matcher(line);
            while (matcher.find()) {
                String word = matcher.group().toLowerCase();
                Integer oldCount;

                oldCount = wordFreqConc.get(word);
                if (oldCount == null) {
                    oldCount = wordFreqConc.putIfAbsent(word, 1);
                    if (oldCount == null)
                        continue;
                    else wordFreqConc.put(word, oldCount + 1);
                }
                else
                    do {
                        oldCount = wordFreqConc.get(word);
                    } while (!wordFreqConc.replace(word, oldCount, oldCount + 1));

            }
        }

Answer 7

Есть необходимость сделать то же самое. Я использую ConcurrentHashMap + AtomicInteger. Кроме того, ReentrantRW Lock был введен для атомарного сброса (очень похожее поведение).

Проверено с 10 ключами и 10 нитями на каждый ключ. Ничего не было потеряно. Я просто еще не пробовал несколько потоков очистки, но надеюсь, что это сработает.

Массивный однопользовательский прилив мучает меня ... Я хочу удалить RWLock и разбить смыва на мелкие кусочки. Завтра.

private ConcurrentHashMap<String,AtomicInteger> counters = new ConcurrentHashMap<String, AtomicInteger>();
private ReadWriteLock rwLock = new ReentrantReadWriteLock();

public void count(String invoker) {

    rwLock.readLock().lock();

    try{
        AtomicInteger currentValue = counters.get(invoker);
        // if entry is absent - initialize it. If other thread has added value before - we will yield and not replace existing value
        if(currentValue == null){
            // value we want to init with
            AtomicInteger newValue = new AtomicInteger(0);
            // try to put and get old
            AtomicInteger oldValue = counters.putIfAbsent(invoker, newValue);
            // if old value not null - our insertion failed, lets use old value as it's in the map
            // if old value is null - our value was inserted - lets use it
            currentValue = oldValue != null ? oldValue : newValue;
        }

        // counter +1
        currentValue.incrementAndGet();
    }finally {
        rwLock.readLock().unlock();
    }

}

/**
 * @return Map with counting results
 */
public Map<String, Integer> getCount() {
    // stop all updates (readlocks)
    rwLock.writeLock().lock();
    try{
        HashMap<String, Integer> resultMap = new HashMap<String, Integer>();
        // read all Integers to a new map
        for(Map.Entry<String,AtomicInteger> entry: counters.entrySet()){
            resultMap.put(entry.getKey(), entry.getValue().intValue());
        }
        // reset ConcurrentMap
        counters.clear();
        return resultMap;

    }finally {
        rwLock.writeLock().unlock();
    }

}