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

Question

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

После реализации этого базового кэширования в ходе дальнейшего тестирования выяснилось, что я не рассматривал возможность одновременного доступа к кэшу одновременно. Я обнаружил, что в течение ~ 100 мс около 50 потоков пытались извлечь объект из кэша, обнаружив, что он истек, ударив веб-службу, чтобы извлечь данные, а затем вернул объект в кеш.

Исходный код выглядел примерно так:

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

    final String key = "Data-" + email; 
    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; 
} 

Таким образом, чтобы убедиться, что только один поток вызова веб-службы, когда объект в key истек, я думал, что мне нужно, чтобы синхронизировать кэш получить/и, казалось бы, использование ключа кеша было бы хорошим кандидатом для синхронизации объекта (таким образом, вызовы этого метода для электронной почты b@b.com не будут блокироваться вызовами методов на a@a.com).

Я обновил способ выглядеть следующим образом:

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

Я также добавил регистрации линий для вещей, как «перед блоком синхронизации», «внутри блока синхронизации», «уходить блок синхронизации», и " после блока синхронизации ", поэтому я мог бы определить, была ли я эффективно синхронизировать операцию get/set.

Однако похоже, что это не сработало. Мои тестовые журналы имеют выход как:

(журнал выход 'threadname' 'имя регистратора' 'сообщение')
HTTP-80-Processor253 jsp.view-страница - getSomeDataForEmail: собирается ввести блок синхронизации
http-80-Processor253 jsp.view-page - getSomeDataForEmail: внутренний блок синхронизации
http-80-Processor253 cache.StaticCache - get: объект у ключа [SomeData-test@test.com] истек
http-80-Processor253 cache.StaticCache - get: key [SomeData-test@test.com] возвращающее значение [null]
http-80-Processor263 jsp.view-page - getSomeDataForEmai l: собирается ввести блок синхронизации
http-80-Processor263 jsp.view-page - getSomeDataForEmail: внутренний блок синхронизации
http-80-Processor263 cache.StaticCache - get: object at key [SomeData-test@test.com] истек
http-80-Processor263 cache.StaticCache - get: key [SomeData-test@test.com] возвращающее значение [null]
http-80-Processor131 jsp.view-page - getSomeDataForEmail: собирается войти в блок синхронизации
http-80-Processor131 jsp.view-page - getSomeDataForEmail: внутренний блок синхронизации
http-80-Processor131 cache.StaticCache - get: объект у ключа [SomeData-test@test.com] истек
HTTP-80-Processor131 cache.StaticCache - получить: ключ [SomeData-test@test.com] возвращая значение [NULL]
HTTP-80-Processor104 jsp.view-страница - getSomeDataForEmail: внутри блока синхронизации
HTTP-80 -Processor104.StaticCache - get: объект в ключе [SomeData-test@test.com] истек
http-80-Processor104 cache.StaticCache - get: key [SomeData-test@test.com] возвращающее значение [null]
http- 80-Processor252 jsp.view-страница - getSomeDataForEmail: около ввести блок синхронизации
HTTP-80-Processor283 jsp.view-страница - getSomeDataForEmail: около ввести блок синхронизации
HTTP-80-Processor2 jsp.view-страница - getSomeDataForEmail : около ввести блок синхронизации
HTTP-80-Processor2 jsp.view-страница - getSomeDataForEmail: внутри блока синхронизации

Я хотел видеть только один поток за один раз, входящий/выходящий из блока синхронизации вокруг операций get/set.

Есть ли проблема в синхронизации объектов String? Я думал, что ключ кеша будет хорошим выбором, поскольку он уникален для операции, и хотя в этом методе объявлен final String key, я думал, что каждый поток будет получать ссылку на на тот же объект и, следовательно, будет синхронизация на этом одном объекте.

Что я здесь делаю неправильно?

Update: после того, смотря в журналах, похоже, методы с той же логикой синхронизации где ключ всегда одинаков, таких как

final String key = "blah"; 
... 
synchronized(key) { ... 

не проявляют такую ​​же проблему параллелизма - только один поток за один раз входит в блок.

Обновление 2: Спасибо всем за помощь! Я принял первый ответ около intern() ing Strings, который решил мою первоначальную проблему - когда несколько потоков входили в синхронизированные блоки, где я думал, что они не должны, потому что key имели такое же значение.

Как указывали другие, использование intern() для такой цели и синхронизация на этих строках действительно оказывается плохой идеей - при запуске тестов JMeter против webapp для имитации ожидаемой нагрузки я видел использованный размер кучи вырастают почти до 1 ГБ чуть менее 20 минут.

В настоящее время я использую простое решение только синхронизации весь метод - но я действительно как образцы кода, предоставляемых martinprobst и MBCook, но так как у меня есть около 7 подобных getData() методов в этом классе в настоящее время (с ней требуется около 7 различных частей данных из веб-службы), я не хотел добавлять почти дублированную логику о получении и освобождении блокировок для каждого метода. Но это определенно очень, очень ценная информация для будущего использования. Я думаю, что это, в конечном счете, правильные ответы на то, как лучше всего сделать такую ​​операцию, как эта потокобезопасная, и я бы дал больше голосов на эти ответы, если бы мог!

Answer 1

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

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

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

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

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

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

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

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

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

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

Вот альтернатива - она ​​по-прежнему использует особый замок, но мы знаем, что нам понадобится один из них для кеша, и вы говорили о 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 и медленную операцию для выполнения. Непростая идея, связанная с этим методом в кеше.

Answer 2

Вызов:

final String key = "Data-" + email; 

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

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

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

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

Answer 3

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

Answer 4

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

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

т.д .:

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 
} 

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

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

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

Answer 5

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

Answer 6

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

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

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

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

Вместо электронной почты, сделать объект, который содержит электронную почту (скажем, объект User), который имеет значение электронной почты как переменной. Если у вас уже есть другой объект, который представляет человека (скажем, вы уже вытащили что-то из БД на основе их электронной почты), вы можете использовать это. Внедряя метод equals и метод hashcode, вы можете убедиться, что Java считает объекты одинаковыми, когда вы делаете static 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 выборок на одном и том же адресе электронной почты. Вам нужно что-то, чтобы не допустить попадания слишком большого количества записей на карту электронной почты. Используя это LRUMap s из Apache Commons, это сделает это.

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

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

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

Резюме

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

Answer 7

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

также:

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

Answer 8

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

final String key = "Data-" + email; 

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

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

final String key = email; 

и избежать все, что дополнительное создание строки тоже.

Answer 9

Вы можете использовать 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.

Answer 10

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

Реализация хорошего кеша не так проста, как считают некоторые люди.

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

Answer 11

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

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

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/put, должна быть, по крайней мере, синхронизация вокруг операций типа get и getIfAbsentPut для безопасного доступа к кешу. Объем синхронизации будет представлять собой сам кеш.

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

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

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

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

Answer 12

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

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 
     } 
    } 
} 

Answer 13

Вот безопасный короткий 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); 
     } 
    } 
    return data; 
} 

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

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

Так что может быть более безопасным и эффективным для использования expiring Guava Cache:

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 потокобезопасно и не будет страдать от одновременного чтения и записи для разных ключей.

Answer 14

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

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

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

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