параллелизм Java: много устройств записи, одно средство чтения

Ответ Криса Дейла выглядит хорошим подходом.

Другой альтернативой может быть использование параллельного Multiset . Один из них находится в библиотеке Google Collections . Вы можете использовать это следующим образом:

private Multiset<String> stats = ConcurrentHashMultiset.create();

public void notify ( String key )
{
    stats.add(key, 1);
}

Если посмотреть на источник , это реализовано с помощью ConcurrentHashMap и с использованием putIfAbsent и версии с тремя аргументами замените , чтобы обнаружить одновременные изменения и повторить попытку.

Почему бы вам не использовать java.util.concurrent.ConcurrentHashMap ? Он обрабатывает все внутренне, избегая бесполезных блокировок на карте и экономя вам много работы: вам не придется заботиться о синхронизации при получении и размещении ..

Из документации:

Хеш-таблица, поддерживающая полный параллелизм получение и регулируемый ожидаемый параллелизм для обновлений. Этот класс подчиняется той же функциональной спецификации, что и Hashtable, и включает версии методов, соответствующие каждому методу Hashtable. Однако, несмотря на то, что все операции являются потокобезопасными, операции извлечения не влекут за собой блокировку , и нет никакой поддержки для блокировки всей таблицы таким образом, чтобы предотвратить любой доступ.

Вы можете указать его уровень параллелизма :

Разрешенный параллелизм между операциями обновления определяется необязательным аргументом конструктора concurrencyLevel (по умолчанию 16), который используется в качестве подсказки для внутреннего изменения размера . Таблица внутренне разделена на разделы, чтобы попытаться разрешить указанное количество одновременных обновлений без конкуренции. Поскольку размещение в хеш-таблицах по существу случайное, фактический параллелизм будет отличаться. В идеале, вы должны выбрать значение, позволяющее разместить столько потоков, сколько когда-либо будет одновременно изменять таблицу . Использование значительно большего значения, чем нужно, может привести к потере места и времени, а значительно меньшее значение может привести к конфликту потоков. Но завышенные и недооцененные в пределах порядка величины обычно не оказывают заметного влияния.Значение один подходит, когда известно, что только один поток будет изменять, а все остальные будут только читать. Кроме того, изменение размера этой или любой другой хэш-таблицы является относительно медленной операцией, поэтому, когда это возможно, рекомендуется предоставлять оценки ожидаемых размеров таблиц в конструкторах.

Как указано в комментариях, внимательно прочтите документацию ConcurrentHashMap , особенно когда в ней говорится об атомарных или неатомарных операциях.

Чтобы получить гарантию атомарности, вы должны учитывать, какие операции являются атомарными. Из интерфейса ConcurrentMap вы узнаете, что:

V putIfAbsent(K key, V value)
V replace(K key, V value)
boolean replace(K key,V oldValue, V newValue)
boolean remove(Object key, Object value)

можно использовать безопасно.

Я бы предложил взглянуть на библиотеку util.concurrent в Java. Я думаю, вы сможете реализовать это решение намного чище. Я не думаю, что здесь вообще нужна карта. Я бы рекомендовал реализовать это с помощью ConcurrentLinkedQueue. Каждый "производитель" может свободно писать в эту очередь, не беспокоясь о других. Он может поместить в очередь объект с данными для своей статистики.

Сборщик может потреблять очередь, постоянно извлекая данные и обрабатывая их. Затем он может хранить их так, как ему нужно.

Другой подход к проблеме заключается в использовании (тривиальной) безопасности потоков с помощью ограничения потоков. По сути, создайте один фоновый поток, который заботится как о чтении, так и о записи. Имеет довольно неплохие характеристики с точки зрения масштабируемости и простоты.

Идея состоит в том, что вместо того, чтобы все потоки пытались обновить данные напрямую, они создают задачу «обновления» для обработки фоновым потоком. Этот же поток также может выполнять задачу чтения, если допустимо некоторое отставание в обработке обновлений.

Этот дизайн довольно хорош, потому что потокам больше не придется конкурировать за блокировку для обновления данных, а поскольку карта ограничена одним потоком, вы можете просто использовать простой HashMap для получения / размещения и т. Д. С точки зрения реализации, это будет означать создание однопоточного исполнителя и отправку задач записи, которые также могут выполнять необязательную операцию «collectAndSave».

Набросок кода может выглядеть следующим образом:

public class StatsService {
    private ExecutorService executor = Executors.newSingleThreadExecutor();
    private final Map<String,Long> stats = new HashMap<String,Long>();

    public void notify(final String key) {
        Runnable r = new Runnable() {
            public void run() {
                Long value = stats.get(key);
                if (value == null) {
                    value = 1L;
                } else {
                    value++;
                }
                stats.put(key, value);
                // do the optional collectAndSave periodically
                if (timeToDoCollectAndSave()) {
                    collectAndSave();
                }
            }
        };
        executor.execute(r);
    }
}

Существует BlockingQueue, связанный с исполнителем, и каждый поток, который создает задачу для StatsService, использует BlockingQueue. Ключевой момент заключается в следующем: продолжительность блокировки для этой операции должна быть намного короче , чем продолжительность блокировки в исходном коде, поэтому конкуренция должна быть намного меньше. В целом это должно привести к гораздо большей пропускной способности и задержке.

Еще одно преимущество состоит в том, что, поскольку только один поток читает и записывает в карту, можно использовать простой HashMap и примитивный длинный тип (без использования ConcurrentHashMap или атомарных типов). Это также значительно упрощает код, который на самом деле его обрабатывает.

Надеюсь, это поможет.

Если мы проигнорируем часть сбора урожая и сосредоточимся на записи, основным узким местом программы является то, что статистика фиксируется на очень грубом уровне детализации. Если два потока хотят обновить разные ключи, они должны ждать.

Если вы заранее знаете набор ключей и можете предварительно инициализировать карту так, чтобы к моменту прихода потока обновления ключ гарантированно существовал, вы могли бы сделать блокировку переменной accumulator вместо всей карты, или использовать безопасный для потока объект accumulator.

Вместо того, чтобы реализовывать это самостоятельно, существуют реализации map, разработанные специально для параллелизма и выполняющие эту более тонкую блокировку для вас.

Однако есть одно предостережение - это статистика, поскольку вам придется получать блокировки на все аккумуляторы примерно в одно и то же время. Если вы используете существующую карту с поддержкой параллелизма, возможно, существует конструкция для получения моментального снимка.

Вы изучали ScheduledThreadPoolExecutor ? Вы можете использовать это для планирования ваших писателей, которые все могут писать в параллельную коллекцию, такую ​​как ConcurrentLinkedQueue , упомянутую @Chris Dail. У вас может быть отдельно запланированное задание для чтения из очереди по мере необходимости, и Java SDK должен обрабатывать практически все ваши проблемы параллелизма, не требуя ручной блокировки.