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

Ответ 1

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

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 );

}

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

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

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

Ответ 2

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

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

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

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

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

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

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

Ответ 3

Использование 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 предлагает определенные методы атомарного обновления, которые можно использовать для создания более высокоуровневых методов обновления методом try-and-repeat.

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

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();
    }
}

Ответ 4

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

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

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

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

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(), скорее всего, упростит ваш код.

Ответ 5

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

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

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

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

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

synchronized (prices.get(e))

Это может вернуть значение null, и вы должны проверить его. синхронизированный по нулевым объектам, не разрешен.

prices.put(e, currentPrice);

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

Ответ 6

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

В классе 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;
    }
}