Параллельные блокировки Java на уровне ключа карты

Использование ConcurrentMap сильно зависит от версии Java. При использовании Java 8 или более поздней версии вы получаете почти все бесплатно:

public final class PriceHolder {

    private ConcurrentMap<String, Price> prices;

    public PriceHolder() {
        this.prices = new ConcurrentHashMap<>();

        //Receive starting prices..
        Price EUR = new Price();
        EUR.setHasChangedSinceLastRead(true);
        EUR.setPrice(BigDecimal.ZERO);

        Price USD = new Price();
        USD.setHasChangedSinceLastRead(true);
        USD.setPrice(BigDecimal.ZERO);
        this.prices.put("EUR", EUR);
        this.prices.put("USD", USD);

    }

    /** Called when a price ‘p’ is received for an entity ‘e’ */
    public void putPrice(String e, BigDecimal p) {
        prices.compute(e, (k, price)-> {
            if(price==null) price=new Price();
            price.setHasChangedSinceLastRead(true);
            price.setPrice(p);
            return price;
        });
    }

    /** Called to get the latest price for entity ‘e’ */
    public BigDecimal getPrice(String e) {
        Price price = prices.computeIfPresent(e, (key, value) -> {
            value.setHasChangedSinceLastRead(false);
            return value;
        });
        return price==null? null: price.getPrice();
    }

    /**
     * Called to determine if the price for entity ‘e’ has
     * changed since the last call to getPrice(e).
     */
    public boolean hasPriceChanged(String e) {
        final Price price = prices.get(e);
        return price!=null && price.isHasChangedSinceLastRead();
    }
}

Методы compute… на параллельной карте блокируют затронутую запись на время вычисления, позволяя продолжить обновление всех других записей. Для простого доступа get, как в hasPriceChanged, дополнительная синхронизация не требуется, если вы вызываете ее только один раз в методе, т.е. сохраняете результат в локальной переменной во время проверки.

До появления Java 8 все было сложнее. Там все предложения ConcurrentMap представляют собой определенные атомарные методы обновления, которые можно использовать для создания более высокоуровневых методов обновления методом проб и повторений.

Чтобы использовать его аккуратно, лучше всего сделать класс значения неизменяемым:

public final class Price {

    private final BigDecimal price;
    private final boolean hasChangedSinceLastRead;

    Price(BigDecimal value, boolean changed) {
      price=value;
      hasChangedSinceLastRead=changed;
    }
    public boolean isHasChangedSinceLastRead() {
        return hasChangedSinceLastRead;
    }
    public BigDecimal getPrice() {
        return price;
    }
}

Затем используйте его, чтобы всегда создавать новый объект, отражающий желаемое новое состояние, и выполнять атомарные обновления, используя либо putIfAbsent, либо replace:

public final class PriceHolder {

    private ConcurrentMap<String, Price> prices;

    public PriceHolder() {
        this.prices = new ConcurrentHashMap<>();

        //Receive starting prices..
        Price EUR = new Price(BigDecimal.ZERO, true);
        Price USD = EUR; // we can re-use immutable objects...
        this.prices.put("EUR", EUR);
        this.prices.put("USD", USD);
    }

    /** Called when a price ‘p’ is received for an entity ‘e’ */
    public void putPrice(String e, BigDecimal p) {
      Price old, _new=new Price(p, true);
      do old=prices.get(e);
      while(old==null? prices.putIfAbsent(e,_new)!=null: !prices.replace(e,old,_new));
    }

    /** Called to get the latest price for entity ‘e’ */
    public BigDecimal getPrice(String e) {
        for(;;) {
          Price price = prices.get(e);
          if(price==null) return null;
          if(!price.isHasChangedSinceLastRead()
          || prices.replace(e, price, new Price(price.getPrice(), false)))
            return price.getPrice();
        }
    }

    /**
     * Called to determine if the price for entity ‘e’ has
     * changed since the last call to getPrice(e).
     */
    public boolean hasPriceChanged(String e) {
        final Price price = prices.get(e);
        return price!=null && price.isHasChangedSinceLastRead();
    }
}

Как насчет чего-то вроде

class AtomicPriceHolder {

  private volatile BigDecimal value;
  private volatile boolean dirtyFlag;

  public AtomicPriceHolder( BigDecimal initialValue) {
    this.value = initialValue;
    this.dirtyFlag = true;
  }

  public synchronized void updatePrice( BigDecimal newPrice ) {
    if ( this.value.equals( newPrice ) == false) {
      this.value = newPrice;
      this.dirtyFlag = true;
    }
  }

  public boolean isDirty() {
    return this.dirtyFlag;
  }

  public BigDecimal peek() {
    return this.value;
  }

  public synchronized BigDecimal read() {
    this.dirtyFlag = false;
    return this.value;
  }

}

...

public void updatePrice( String id, BigDecimal value ) {

  AtomicPriceHolder holder;
  synchronized( someGlobalSyncObject ) {
    holder = prices.get(id);
    if ( holder == null ) {
      prices.put( id, new AtomicPriceHolder( value ) );
      return;
    }
  }

  holder.updatePrice( value );

}

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

Условные операции «проверить, есть ли это на карте, создать новую и вставить, если нет» должны быть атомарными и должны выполняться путем блокировки всей карты на этот короткий период. Все остальное потребует отдельного объекта синхронизации для каждого ключа. Их нужно было бы где-то хранить и управлять ими, а доступ к этому хранилищу нужно было бы снова синхронизировать и т. д.

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

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

Простого использования ConcurrentHashMap достаточно, чтобы обеспечить свободный доступ к ключам; get не создают конфликтов, а put блокируют только подмножество ключей, а не всю карту.

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

Чтобы обеспечить согласованность, вам необходимо синхронизировать какой-либо общий объект (или использовать другой механизм блокировки, например ReentrantLock); Я бы предложил создать ConcurrentHashMap<String, Object> объектов блокировки, чтобы вы могли:

synchronized (locks.get(e)) { ... }

Просто заполните карту new Object(). Риск используемого вами паттерна (блокировка объектов Price) заключается в том, что теперь эти объекты должны сохраняться и никогда не заменяться. Легче обеспечить это, имея выделенную частную коллекцию блокировок, а не перегружая свой тип значения в качестве механизма блокировки.

Кроме того, если вы пытаетесь выполнять денежные операции на Java, вам обязательно следует использовать Joda- Money вместо того, чтобы изобретать велосипед.

hasChangedMethod возвращает логическое значение, определяющее, изменилась ли цена с момента последнего вызова getPrice.

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

Тем не менее, вот как я могу реализовать поведение, которое вы ищете. Вполне могут быть альтернативы, это только первый проход.

Держите ConcurrentHashMap<String, ThreadLocal<Boolean>>; ThreadLocal будет хранить статус вызовов get для каждого потока. Я также использую отдельную приватную карту замков.

ConcurrentHashMap<String, Price> pricesMap;
ConcurrentHashMap<String, ThreadLocal<Boolean>> seenMap;
ConcurrentHashMap<String, Object> lockMap;

private Object getLock(String key) {
  return lockMap.computeIfAbsent(key, k -> new Object());
}

private ThreadLocal<Boolean> getSeen(String key) {
  return seenMap.computeIfAbsent(e,
      ThreadLocal.<Boolean>withInitial(() -> false));
}

public void putPrice(String e, BigDecimal p) {
  synchronized (getLock(e)) {
    // price has changed, clear the old state to mark all threads unseen
    seenMap.remove(e);
    pricesMap.get(e).setPrice(p);
  }
}

public BigDecimal getPrice(String e) {
  synchronized (getLock(e)) {
    // marks the price seen for this thread
    getSeen(e).set(true);
    BigDecimal price = pricesMap.get(e);
    return price != null ? price.getPrice() : null;
  }
}

public boolean hasPriceChanged(String e) {
  synchronized (getLock(e)) {
    return !getSeen(e).get();
  }
}

Обратите внимание, что несмотря на то, что структура данных является потокобезопасной, здесь все еще существует риск состояния гонки — вы можете вызвать hasPriceChanged() и получить обратно false, сразу после чего цена будет изменена другим потоком. Отказ от этого hasPriceChanged() поведения, скорее всего, упростит ваш код.

Я бы рекомендовал использовать наблюдаемый шаблон наблюдателя. Не нужно заново изобретать велосипед. См. Наблюдатель и Наблюдаемый

Я бы также рекомендовал изучить условие Так как нет необходимости запирать полный объект для всех читателей. Чтение может быть одновременным, а письмо — нет.

Если коллекция параллельна, это не означает, что она волшебным образом синхронизирует все. Это просто гарантирует, что их методы потокобезопасны. Как только вы покидаете область действия функции, блокировка снимается. Поскольку вам нужен более продвинутый способ управления синхронизацией, лучше всего взять это в свои руки и использовать обычный HashMap.

Некоторые примечания:

Вы злоупотребляете HashMap.get. Подумайте о том, чтобы получить его один раз и сохранить в переменной.

synchronized (prices.get(e))

Это может вернуть null, и вы должны проверить это. синхронизация на нулевых объектах не допускается.

prices.put(e, currentPrice);

Я не уверен, что это задумано, но это действие не нужно. См. это

Лучший способ получить то, что вы хотите, — это, вероятно, пойти дальше и использовать ConcurrentMap для карты, а всю остальную синхронизацию поместить в ваш класс Price. Это приведет к более простому коду, который всегда очень ценен в многопоточной среде, чтобы избежать незначительных ошибок, а также достичь ваших целей одновременного доступа к карте и отслеживания того, была ли запись с момента последнего чтения для каждой валюты.

В классе Price всякий раз, когда вы устанавливаете цену, вы также хотите установить hasChangedSinceLastRead; эти две вещи идут вместе как одна операция, которая должна быть атомарной. Всякий раз, когда вы читаете цену, вы также хотите очистить hasChangedSinceLastRead; это вы тоже хотите быть атомарным. Таким образом, класс должен разрешать модифицировать hasChangedSinceLastRead только этим двум операциям, а не оставлять логику другим классам, а методы должны быть синхронизированы, чтобы гарантировать, что price и hasChangedSinceLastRead не могут выйти из синхронизации из-за доступа нескольких потоков. Теперь класс должен выглядеть так:

public class Price {

    public boolean isHasChangedSinceLastRead() {
        return hasChangedSinceLastRead;
    }

    // setHasChangedSinceLastRead() removed

    public synchronized BigDecimal getPrice() {
        hasChangedSinceLastRead = false;
        return price;
    }

    public synchronized void setPrice(BigDecimal newPrice) {
        if (null != price && price.equals(newPrice) {
            return;
        }
        price = newPrice;
        hasChangedSinceLastRead = true;
    }

    private BigDecimal price;
    private volatile boolean hasChangedSinceLastRead = false;
}

Обратите внимание, что вы можете либо сделать isHasChangedSinceLastRead() синхронизированным, либо hasChangedSinceLastRead volatile; Я выбрал последнее, оставив isHasChangedSinceLastRead() несинхронизированным, потому что для синхронизации метода требуется полный барьер памяти, а для создания переменной volatile требуется только считываемая половина барьера памяти при чтении переменной.

Причина, по которой чтение hasChangedSinceLastRead требует некоторого барьера памяти, заключается в том, что синхронизированные методы, блоки и доступ к энергозависимой памяти гарантируют только эффективный порядок выполнения — отношение «происходит до» — с другими синхронизированными методами, блоками и доступом к энергозависимой памяти. Если isHasChangedSinceLastRead не синхронизирована и переменная не является изменчивой, отношения «произошло до» не существует; в этом случае isHasChangedSinceLastRead может вернуть "true", но вызывающий его поток может не увидеть изменение цены. Это связано с тем, что установка price и hasChangedSinceLastRead в setPrice() может быть воспринята другими потоками в обратном порядке, если отношение "происходит раньше" не установлено.

Теперь, когда вся необходимая синхронизация находится в классах ConcurrentMap и Price, вам больше не нужно выполнять какую-либо синхронизацию в PriceHolder, а PriceHolder больше не нужно беспокоиться об обновлении hasChangedSinceLastRead. Код упрощен до:

public final class PriceHolder {

    private ConcurrentMap<String, Price> prices;

    public PriceHolder() {
        prices = new ConcurrentHashMap<>();

        //Receive starting prices..
        Price EUR = new Price();
        EUR.setPrice(new BigDecimal(0));
        this.prices.put("EUR", EUR);

        Price USD = new Price();
        USD.setPrice(new BigDecimal(0));
        this.prices.put("USD", USD);
    }

    /** Called when a price ‘p’ is received for an entity ‘e’ */
    public void putPrice(
        String e,
        BigDecimal p
    ) throws InterruptedException {
        Price currentPrice = prices.get(e);
        if (null == currentPrice) {
            currentPrice = new Price();
            currentPrice = prices.putIfAbsent(e);
        }
        currentPrice.setPrice(p);
    }

    /** Called to get the latest price for entity ‘e’ */
    public BigDecimal getPrice(String e) {
        Price currentPrice = prices.get(e);
        if (currentPrice != null){
            return currentPrice.getPrice();
        }
        return null;
    }

    /**
     * Called to determine if the price for entity ‘e’ has
     * changed since the last call to getPrice(e).
     */
    public boolean hasPriceChanged(String e) {
        Price currentPrice = prices.get(e);
        return null != currentPrice ? currentPrice.isHasChangedSinceLastRead() : false;
    }
}

Я думаю, что вы можете заблокировать только свой конкретный ключ.

import java.util.HashMap;

public class CustomHashmap<K, V> extends HashMap<K, V>{

    private static final long serialVersionUID = 1L;

    @Override
    public V put(K key, V value) {
        V val = null;
        synchronized (key) {
            val = super.put(key, value);
        }

        return val;
    }

    @Override
    public V get(Object key) {
        return super.get(key);
    }


}