Синхронизация объектов String в Java

Не вдаваясь в подробности, при быстром просмотре того, что вы говорите, похоже, что вам нужно intern () ваши строки:

final String firstkey = "Data-" + email;
final String key = firstkey.intern();

Две строки с одинаковым значением в противном случае не обязательно являются одним и тем же объектом.

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

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

Ответ на обновление вопроса:

Я думаю, это потому, что строковый литерал всегда дает один и тот же объект. Дэйв Коста отмечает в комментарии, что это даже лучше: литерал всегда дает каноническое представление. Таким образом, все строковые литералы с одинаковым значением в любом месте программы будут давать один и тот же объект.

Редактировать

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

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

Вот альтернатива - он по-прежнему использует одиночную блокировку, но мы знаем, что нам в любом случае понадобится одна из них для кеша, и вы говорили о 50 потоках, а не о 5000, так что это не может быть фатальным. Я также предполагаю, что узким местом производительности здесь является медленный блокирующий ввод-вывод в DoSlowThing (), который, следовательно, получит огромную выгоду, если не будет сериализован. Если проблема не в этом, то:

• Если ЦП занят, этого подхода может быть недостаточно, и вам нужен другой подход.
• Если ЦП не занят и доступ к серверу не является узким местом, тогда этот подход является излишним, и вы можете также забыть и об этом, и о блокировке по ключам, поместить большой синхронизированный (StaticCache) вокруг всей операции и это легкий способ.

Очевидно, что этот подход необходимо протестировать на масштабируемость перед использованием - я ничего не гарантирую.

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

IN_PROGRESS - фиктивное значение - не совсем чистое, но код простой и позволяет избежать двух хэш-таблиц. Он не обрабатывает InterruptedException, потому что я не знаю, что ваше приложение хочет делать в этом случае. Кроме того, если DoSlowThing () постоянно терпит неудачу для данного ключа, этот код в его нынешнем виде не совсем элегантен, поскольку каждый проходящий поток будет повторять его. Поскольку я не знаю, каковы критерии отказа и могут ли они быть временными или постоянными, я тоже не занимаюсь этим, я просто проверяю, не блокируются ли потоки навсегда. На практике вы можете захотеть поместить значение данных в кеш, которое указывает «недоступно», возможно, по причине, и тайм-аут, когда следует повторить попытку.

// do not attempt double-check locking here. I mean it.
synchronized(StaticObject) {
    data = StaticCache.get(key);
    while (data == IN_PROGRESS) {
        // another thread is getting the data
        StaticObject.wait();
        data = StaticCache.get(key);
    }
    if (data == null) {
        // we must get the data
        StaticCache.put(key, IN_PROGRESS, TIME_MAX_VALUE);
    }
}
if (data == null) {
    // we must get the data
    try {
        data = server.DoSlowThing(key);
    } finally {
        synchronized(StaticObject) {
            // WARNING: failure here is fatal, and must be allowed to terminate
            // the app or else waiters will be left forever. Choose a suitable
            // collection type in which replacing the value for a key is guaranteed.
            StaticCache.put(key, data, CURRENT_TIME);
            StaticObject.notifyAll();
        }
    }
}

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

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

Вызов:

   final String key = "Data-" + email;

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

Это дальнейшее объяснение вашего редактирования. Когда у вас есть статическая строка, она будет работать.

Использование intern () решает проблему, потому что оно возвращает строку из внутреннего пула, хранимого классом String, что гарантирует, что если две строки равны, будет использоваться строка из пула. Видеть

http://java.sun.com/j2se/1.4.2/docs/api/java/lang/String.html#intern ()

Ваша основная проблема заключается не только в том, что может быть несколько экземпляров String с одним и тем же значением. Основная проблема заключается в том, что вам нужен только один монитор, на котором будет выполняться синхронизация для доступа к объекту StaticCache. В противном случае несколько потоков могут закончить одновременное изменение StaticCache (хотя и с разными ключами), что, скорее всего, не поддерживает одновременное изменение.

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

В вашем случае вы можете использовать синхронизированную хеш-таблицу для хранения объектов блокировки для ваших ключей.

Например.:

Object data = StaticCache.get(key, ...);
if (data == null) {
  Object lock = lockTable.get(key);
  if (lock == null) {
    // we're the only one looking for this
    lock = new Object();
    synchronized(lock) {
      lockTable.put(key, lock);
      // get stuff
      lockTable.remove(key);
    }
  } else {
    synchronized(lock) {
      // just to wait for the updater
    }
    data = StaticCache.get(key);
  }
} else {
  // use from cache
}

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

Если вы делаете кеш недействительным через некоторое время, вы должны проверить, являются ли данные снова нулевыми после извлечения их из кеша в случае lock! = Null.

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

Почему бы просто не отобразить статическую html-страницу, которая обслуживается пользователем и восстанавливается каждые x минут?

Другие предлагали интернировать струны, и это сработает.

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

Я видел два решения этого:

Вы можете синхронизировать на другом объекте

Вместо электронного письма создайте объект, содержащий электронное письмо (скажем, объект «Пользователь»), который содержит значение электронной почты в качестве переменной. Если у вас уже есть другой объект, представляющий человека (скажем, вы уже вытащили что-то из БД на основе его электронной почты), вы можете его использовать. Реализуя метод equals и метод хэш-кода, вы можете убедиться, что Java считает объекты одинаковыми, когда вы выполняете статический cache.contains (), чтобы узнать, находятся ли данные уже в кеше (вам нужно будет синхронизировать в кеше ).

Фактически, вы могли бы сохранить вторую карту для объектов, на которых можно было бы закрепиться. Что-то вроде этого:

Map<String, Object> emailLocks = new HashMap<String, Object>();

Object lock = null;

synchronized (emailLocks) {
    lock = emailLocks.get(emailAddress);

    if (lock == null) {
        lock = new Object();
        emailLocks.put(emailAddress, lock);
    }
}

synchronized (lock) {
    // See if this email is in the cache
    // If so, serve that
    // If not, generate the data

    // Since each of this person's threads synchronizes on this, they won't run
    // over eachother. Since this lock is only for this person, it won't effect
    // other people. The other synchronized block (on emailLocks) is small enough
    // it shouldn't cause a performance problem.
}

Это предотвратит получение 15 запросов на один и тот же адрес электронной почты за один раз. Вам понадобится что-то, чтобы предотвратить попадание слишком большого количества записей на карту emailLocks. Это можно сделать с помощью LRUMaps из Apache Commons.

Это потребует некоторой настройки, но это может решить вашу проблему.

Используйте другой ключ

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

Резюме

Надеюсь, это поможет. Распределение потоков - это весело, не так ли? Вы также можете использовать сеанс, чтобы установить значение, означающее «Я уже работаю над поиском этого», и проверить это, чтобы увидеть, нужно ли второму (третьему, N-му) потоку попытаться создать или просто дождаться появления результата. в кеше. Думаю, у меня было три предложения.

Строки являются хорошими кандидатами на синхронизацию нет. Если вам необходимо выполнить синхронизацию по идентификатору строки, это можно сделать, используя строку для создания мьютекса (см. «синхронизация по идентификатору»). Стоит ли стоимость этого алгоритма, зависит от того, требует ли вызов вашей службы каких-либо значительных операций ввода-вывода.

Также:

• Я надеюсь, что методы StaticCache.get () и набор() безопасны для потоков.
• String.intern () имеет свою стоимость (которая зависит от реализации виртуальной машины) и должна использоваться с осторожностью.

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

final String key = "Data-" + email;

Есть ли в кеше другие объекты / типы объектов, которые используют адрес электронной почты, который вам нужен, эти дополнительные «Data-» в начале ключа?

если нет, я бы просто сделал это

final String key = email;

и вы также избегаете создания лишних строк.

Вы можете использовать утилиты параллелизма 1.5 для предоставления кеша, предназначенного для обеспечения множественного одновременного доступа, и единственной точки добавления (т.е. только один поток, когда-либо выполняющий дорогостоящее «создание» объекта):

 private ConcurrentMap<String, Future<SomeData[]> cache;
 private SomeData[] getSomeDataByEmail(final WebServiceInterface service, final String email) throws Exception {

  final String key = "Data-" + email;
  Callable<SomeData[]> call = new Callable<SomeData[]>() {
      public SomeData[] call() {
          return service.getSomeDataForEmail(email);
      }
  }
  FutureTask<SomeData[]> ft; ;
  Future<SomeData[]> f = cache.putIfAbsent(key, ft= new FutureTask<SomeData[]>(call)); //atomic
  if (f == null) { //this means that the cache had no mapping for the key
      f = ft;
      ft.run();
  }
  return f.get(); //wait on the result being available if it is being calculated in another thread
}

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

Используйте достойный фреймворк кеширования, такой как ehcache.

Реализовать хороший кеш не так просто, как думают некоторые.

Что касается комментария о том, что String.intern () является источником утечек памяти, это на самом деле неверно. Собран мусор Interned Strings являются, это может занять больше времени, потому что на некоторых JVM (SUN) они хранятся в пермском пространстве, которое затрагивается только полными сборщиками мусора.

Это довольно поздно, но здесь представлено довольно много некорректного кода.

В этом примере:

private SomeData[] getSomeDataByEmail(WebServiceInterface service, String email) {


  SomeData[] data = null;
  final String key = "Data-" + email;

  synchronized(key) {      
    data =(SomeData[]) StaticCache.get(key);

    if (data == null) {
        data = service.getSomeDataForEmail(email);
        StaticCache.set(key, data, CACHE_TIME);
    }
    else {
      logger.debug("getSomeDataForEmail: using cached object");
    }
  }

  return data;
}

Неправильная область синхронизации. Для статического кеша, поддерживающего API получения / ввода, должна быть как минимум синхронизация операций типа get и getIfAbsentPut для безопасного доступа к кешу. Объем синхронизации будет сам кеш.

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

SynchronizedMap можно использовать вместо явной синхронизации, но при этом необходимо соблюдать осторожность. Если используются неправильные API (get и put вместо putIfAbsent), то операции не будут иметь необходимой синхронизации, несмотря на использование синхронизированной карты. Обратите внимание на сложности, возникающие при использовании putIfAbsent: либо значение put должно вычисляться даже в тех случаях, когда оно не требуется (потому что функция put не может знать, требуется ли значение put, пока не будет исследовано содержимое кеша), либо требуется тщательная проверка. использование делегирования (скажем, с использованием Future, которое работает, но в некоторой степени несовместимо; см. ниже), при котором значение пут получается по запросу, если это необходимо.

Использование Futures возможно, но кажется довольно неудобным и, возможно, немного изощренным. Future API лежит в основе асинхронных операций, в частности, для операций, которые могут не завершиться немедленно. Вовлечение Future, скорее всего, добавляет уровень создания потоков - дополнительные, вероятно, ненужные сложности.

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

другой способ синхронизации на строковом объекте:

String cacheKey = ...;

    Object obj = cache.get(cacheKey)

    if (obj==null){
    synchronized (Integer.valueOf(Math.abs(cacheKey.hashCode()) % 127)){
          obj = cache.get(cacheKey)
         if (obj==null){
             //some cal obtain obj value,and put into cache
        }
    }
}

Вот безопасное короткое решение для Java 8, которое использует карту выделенных объектов блокировки для синхронизации:

private static final Map<String, Object> keyLocks = new ConcurrentHashMap<>();

private SomeData[] getSomeDataByEmail(WebServiceInterface service, String email) {
    final String key = "Data-" + email;
    synchronized (keyLocks.computeIfAbsent(key, k -> new Object())) {
        SomeData[] data = StaticCache.get(key);
        if (data == null) {
            data = service.getSomeDataForEmail(email);
            StaticCache.set(key, data);
        }
    }
    return data;
}

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

Это можно обойти следующим образом:

private SomeData[] getSomeDataByEmail(WebServiceInterface service, String email) {
    final String key = "Data-" + email;
    synchronized (keyLocks.computeIfAbsent(key, k -> new Object())) {
        try {
            SomeData[] data = StaticCache.get(key);
            if (data == null) {
                data = service.getSomeDataForEmail(email);
                StaticCache.set(key, data);
            }
        } finally {
            keyLocks.remove(key); // vulnerable to race-conditions
        }
    }
    return data;
}

Но тогда популярные ключи будут постоянно повторно вставляться в карту с перераспределением объектов блокировки.

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

Так что может быть более безопасно и эффективно использовать истекающий срок действия кэша Guava:

private static final LoadingCache<String, Object> keyLocks = CacheBuilder.newBuilder()
        .expireAfterAccess(10, TimeUnit.MINUTES) // max lock time ever expected
        .build(CacheLoader.from(Object::new));

private SomeData[] getSomeDataByEmail(WebServiceInterface service, String email) {
    final String key = "Data-" + email;
    synchronized (keyLocks.getUnchecked(key)) {
        SomeData[] data = StaticCache.get(key);
        if (data == null) {
            data = service.getSomeDataForEmail(email);
            StaticCache.set(key, data);
        }
    }
    return data;
}

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

Насколько я могу судить, если у других есть аналогичная проблема, следующий код работает:

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

public class KeySynchronizer<T> {

    private Map<T, CounterLock> locks = new ConcurrentHashMap<>();

    public <U> U synchronize(T key, Supplier<U> supplier) {
        CounterLock lock = locks.compute(key, (k, v) -> 
                v == null ? new CounterLock() : v.increment());
        synchronized (lock) {
            try {
                return supplier.get();
            } finally {
                if (lock.decrement() == 0) {
                    // Only removes if key still points to the same value,
                    // to avoid issue described below.
                    locks.remove(key, lock);
                }
            }
        }
    }

    private static final class CounterLock {

        private AtomicInteger remaining = new AtomicInteger(1);

        private CounterLock increment() {
            // Returning a new CounterLock object if remaining = 0 to ensure that
            // the lock is not removed in step 5 of the following execution sequence:
            // 1) Thread 1 obtains a new CounterLock object from locks.compute (after evaluating "v == null" to true)
            // 2) Thread 2 evaluates "v == null" to false in locks.compute
            // 3) Thread 1 calls lock.decrement() which sets remaining = 0
            // 4) Thread 2 calls v.increment() in locks.compute
            // 5) Thread 1 calls locks.remove(key, lock)
            return remaining.getAndIncrement() == 0 ? new CounterLock() : this;
        }

        private int decrement() {
            return remaining.decrementAndGet();
        }
    }
}

В случае OP это будет использоваться так:

private KeySynchronizer<String> keySynchronizer = new KeySynchronizer<>();

private SomeData[] getSomeDataByEmail(WebServiceInterface service, String email) {
    String key = "Data-" + email;
    return keySynchronizer.synchronize(key, () -> {
        SomeData[] existing = (SomeData[]) StaticCache.get(key);
        if (existing == null) {
            SomeData[] data = service.getSomeDataForEmail(email);
            StaticCache.set(key, data, CACHE_TIME);
            return data;
        }
        logger.debug("getSomeDataForEmail: using cached object");
        return existing;
    });
}

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

public void synchronize(T key, Runnable runnable) {
    CounterLock lock = locks.compute(key, (k, v) -> 
            v == null ? new CounterLock() : v.increment());
    synchronized (lock) {
        try {
            runnable.run();
        } finally {
            if (lock.decrement() == 0) {
                // Only removes if key still points to the same value,
                // to avoid issue described below.
                locks.remove(key, lock);
            }
        }
    }
}

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

public class ValueLock<T> {

    private Lock lock = new ReentrantLock();
    private Map<T, Condition> conditions  = new HashMap<T, Condition>();

    public void lock(T t){
        lock.lock();
        try {
            while (conditions.containsKey(t)){
                conditions.get(t).awaitUninterruptibly();
            }
            conditions.put(t, lock.newCondition());
        } finally {
            lock.unlock();
        }
    }

    public void unlock(T t){
        lock.lock();
        try {
            Condition condition = conditions.get(t);
            if (condition == null)
                throw new IllegalStateException();// possibly an attempt to release what wasn't acquired
            conditions.remove(t);
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

После (внешней) операции lock (внутренняя) блокировка приобретается для получения монопольного доступа к карте на короткое время, и если соответствующий объект уже находится на карте, текущий поток будет ждать, в противном случае он поместит новый Condition на карту, снимет (внутреннюю) блокировку и продолжит, и (внешняя) блокировка считается полученной. Операция (внешняя) unlock, сначала установив (внутреннюю) блокировку, отправит сигнал на Condition, а затем удалит объект с карты.

Класс не использует параллельную версию Map, потому что каждый доступ к нему защищен одиночной (внутренней) блокировкой.

Обратите внимание, семантика метода lock() этого класса отличается от семантики ReentrantLock.lock(), повторные вызовы lock() без парного unlock() будут зависать в текущем потоке на неопределенное время.

Пример использования, который может быть применим к ситуации, описанной OP

    ValueLock<String> lock = new ValueLock<String>();
    // ... share the lock   
    String email = "...";
    try {
        lock.lock(email);
        //... 
    } finally {
        lock.unlock(email);
    }

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

См. Реализацию для Java 8, Java 6 и небольшой тест.

Java 8:

public class DynamicKeyLock<T> implements Lock
{
    private final static ConcurrentHashMap<Object, LockAndCounter> locksMap = new ConcurrentHashMap<>();

    private final T key;

    public DynamicKeyLock(T lockKey)
    {
        this.key = lockKey;
    }

    private static class LockAndCounter
    {
        private final Lock lock = new ReentrantLock();
        private final AtomicInteger counter = new AtomicInteger(0);
    }

    private LockAndCounter getLock()
    {
        return locksMap.compute(key, (key, lockAndCounterInner) ->
        {
            if (lockAndCounterInner == null) {
                lockAndCounterInner = new LockAndCounter();
            }
            lockAndCounterInner.counter.incrementAndGet();
            return lockAndCounterInner;
        });
    }

    private void cleanupLock(LockAndCounter lockAndCounterOuter)
    {
        if (lockAndCounterOuter.counter.decrementAndGet() == 0)
        {
            locksMap.compute(key, (key, lockAndCounterInner) ->
            {
                if (lockAndCounterInner == null || lockAndCounterInner.counter.get() == 0) {
                    return null;
                }
                return lockAndCounterInner;
            });
        }
    }

    @Override
    public void lock()
    {
        LockAndCounter lockAndCounter = getLock();

        lockAndCounter.lock.lock();
    }

    @Override
    public void unlock()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);
        lockAndCounter.lock.unlock();

        cleanupLock(lockAndCounter);
    }


    @Override
    public void lockInterruptibly() throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        try
        {
            lockAndCounter.lock.lockInterruptibly();
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }
    }

    @Override
    public boolean tryLock()
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired = lockAndCounter.lock.tryLock();

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired;
        try
        {
            acquired = lockAndCounter.lock.tryLock(time, unit);
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public Condition newCondition()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);

        return lockAndCounter.lock.newCondition();
    }
}

Java 6:

открытый класс DynamicKeyLock реализует Lock { частный финальный статический ConcurrentHashMap locksMap = new ConcurrentHashMap (); закрытый конечный ключ T;

    public DynamicKeyLock(T lockKey) {
        this.key = lockKey;
    }

    private static class LockAndCounter {
        private final Lock lock = new ReentrantLock();
        private final AtomicInteger counter = new AtomicInteger(0);
    }

    private LockAndCounter getLock()
    {
        while (true) // Try to init lock
        {
            LockAndCounter lockAndCounter = locksMap.get(key);

            if (lockAndCounter == null)
            {
                LockAndCounter newLock = new LockAndCounter();
                lockAndCounter = locksMap.putIfAbsent(key, newLock);

                if (lockAndCounter == null)
                {
                    lockAndCounter = newLock;
                }
            }

            lockAndCounter.counter.incrementAndGet();

            synchronized (lockAndCounter)
            {
                LockAndCounter lastLockAndCounter = locksMap.get(key);
                if (lockAndCounter == lastLockAndCounter)
                {
                    return lockAndCounter;
                }
                // else some other thread beat us to it, thus try again.
            }
        }
    }

    private void cleanupLock(LockAndCounter lockAndCounter)
    {
        if (lockAndCounter.counter.decrementAndGet() == 0)
        {
            synchronized (lockAndCounter)
            {
                if (lockAndCounter.counter.get() == 0)
                {
                    locksMap.remove(key);
                }
            }
        }
    }

    @Override
    public void lock()
    {
        LockAndCounter lockAndCounter = getLock();

        lockAndCounter.lock.lock();
    }

    @Override
    public void unlock()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);
        lockAndCounter.lock.unlock();

        cleanupLock(lockAndCounter);
    }


    @Override
    public void lockInterruptibly() throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        try
        {
            lockAndCounter.lock.lockInterruptibly();
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }
    }

    @Override
    public boolean tryLock()
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired = lockAndCounter.lock.tryLock();

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired;
        try
        {
            acquired = lockAndCounter.lock.tryLock(time, unit);
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public Condition newCondition()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);

        return lockAndCounter.lock.newCondition();
    }
}

Тестовое задание:

public class DynamicKeyLockTest
{
    @Test
    public void testDifferentKeysDontLock() throws InterruptedException
    {
        DynamicKeyLock<Object> lock = new DynamicKeyLock<>(new Object());
        lock.lock();
        AtomicBoolean anotherThreadWasExecuted = new AtomicBoolean(false);
        try
        {
            new Thread(() ->
            {
                DynamicKeyLock<Object> anotherLock = new DynamicKeyLock<>(new Object());
                anotherLock.lock();
                try
                {
                    anotherThreadWasExecuted.set(true);
                }
                finally
                {
                    anotherLock.unlock();
                }
            }).start();
            Thread.sleep(100);
        }
        finally
        {
            Assert.assertTrue(anotherThreadWasExecuted.get());
            lock.unlock();
        }
    }

    @Test
    public void testSameKeysLock() throws InterruptedException
    {
        Object key = new Object();
        DynamicKeyLock<Object> lock = new DynamicKeyLock<>(key);
        lock.lock();
        AtomicBoolean anotherThreadWasExecuted = new AtomicBoolean(false);
        try
        {
            new Thread(() ->
            {
                DynamicKeyLock<Object> anotherLock = new DynamicKeyLock<>(key);
                anotherLock.lock();
                try
                {
                    anotherThreadWasExecuted.set(true);
                }
                finally
                {
                    anotherLock.unlock();
                }
            }).start();
            Thread.sleep(100);
        }
        finally
        {
            Assert.assertFalse(anotherThreadWasExecuted.get());
            lock.unlock();
        }
    }
}

Вы можете безопасно использовать String.intern для синхронизации, если можете разумно гарантировать, что строковое значение уникально в вашей системе. UUIDS - хороший способ подойти к этому. Вы можете связать UUID с вашим фактическим строковым ключом либо через кеш, либо через карту, либо, возможно, даже сохранить uuid как поле в вашем объекте сущности.

    @Service   
    public class MySyncService{

      public Map<String, String> lockMap=new HashMap<String, String>();

      public void syncMethod(String email) {

        String lock = lockMap.get(email);
        if (lock==null) {
            lock = UUID.randomUUID().toString();
            lockMap.put(email, lock);
        }   

        synchronized(lock.intern()) {
                //do your sync code here
        }
    }

В вашем случае вы можете использовать что-то вроде этого (это не приводит к утечке памяти):

private Synchronizer<String> synchronizer = new Synchronizer();

private SomeData[] getSomeDataByEmail(WebServiceInterface service, String email) {
    String key = "Data-" + email;

    return synchronizer.synchronizeOn(key, () -> {

        SomeData[] data = (SomeData[]) StaticCache.get(key);
        if (data == null) {
            data = service.getSomeDataForEmail(email);
            StaticCache.set(key, data, CACHE_TIME);
        } else {
          logger.debug("getSomeDataForEmail: using cached object");
        }
        return data;

    });
}

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

compile 'com.github.matejtymes:javafixes:1.3.0'

Последнее обновление 2019,

Если вы ищете новые способы реализации синхронизации в JAVA, этот ответ для вас.

Я нашел этот замечательный блог Анатолия Коровина, он поможет вам глубже понять синхронизацию.

Как синхронизировать блоки по значению объекта в Java.

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

Начиная с Java 8, вы можете использовать computeIfAbsent или ConcurrentHashMap для создания блокировок по имени. Для этого требуется всего несколько строк кода:

final ConcurrentHashMap<String, ReentrantLock> locks = new ConcurrentHashMap<>();
final ReentrantLock lock = locks.computeIfAbsent(lockName, (name) -> new ReentrantLock());

lock.lock();
try {
    // Do something.
} finally {
    lock.unlock();
}

Если именованная блокировка не существует, она будет создана computeIfAbsent атомарно.

Вы должны быть очень осторожны при использовании короткоживущих объектов с синхронизацией. К каждому объекту Java прикреплен монитор, и по умолчанию этот монитор спущен; однако, если 2 потока конкурируют за получение монитора, монитор раздувается. Если объект будет долгоживущим, это не проблема. Однако, если объект недолговечен, очистка этого раздутого монитора может серьезно сказаться на времени сборки мусора (что приводит к увеличению задержек и снижению пропускной способности). И может быть даже сложно определить время сборки мусора, поскольку оно не всегда указывается.

Если вы действительно хотите синхронизировать, вы можете использовать java.util.concurrent.Lock. Или воспользуйтесь созданным вручную полосатым замком и используйте хэш строки в качестве индекса для этого полосатого замка. Этот полосатый замок вы держите, чтобы не получить проблем с сборщиком мусора.

Так что-то вроде этого:

static final Object[] locks = newLockArray();

Object lock = locks[hashToIndex(key.hashcode(),locks.length];
synchronized(lock){
       ....
}

int hashToIndex(int hash, int length) {
    if (hash == Integer.MIN_VALUE return 0;
    return abs(hash) % length;
}