Параллелизм Kotlin для ConcurrentHashMap

Вы можете избавиться от блокировки.

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

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

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

Когда вы получаете от x до retrieveA и изменяете объект, вам нужно убедиться, что он все еще находится в кеше. Просто вызовите функцию извлечения еще раз. Если вы получаете точно такой же объект, все в порядке. Если отличается, то это означает, что объект был удален и отправлен по сети, но вы не знаете, была ли также отправлена ​​модификация или было отправлено состояние объекта до модификации. Тем не менее, я думаю, в вашем случае вы можете просто повторить весь процесс (применить изменение и проверить, совпадают ли объекты). Но это зависит от специфики вашего приложения.

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

Вы также можете заменить incrementAndGet на compareAndSwap, но это может немного ухудшить производительность. В этом подходе вместо увеличения вы пытаетесь поменять местами значение, которое больше на единицу. И перед отправкой по сети вы пытаетесь поменять значение на -1, чтобы обозначить значение как недопустимое. Если второй обмен не удался, это означает, что кто-то изменил значение одновременно, вам нужно проверить его еще раз, чтобы отправить самое свежее значение по сети, и вы повторяете процесс в цикле (разрывая цикл только после того, как обмен на -1 успешно). В случае замены на единицу больше вы также повторяете процесс в цикле до тех пор, пока замена не завершится успешно. Если это не удается, это либо означает, что кто-то другой переключился на какое-то большее значение, либо Flusher поменялся на -1. В последнем случае вы знаете, что вам нужно вызвать retrieveA еще раз, чтобы получить новый объект.

Самое простое решение (но с худшей производительностью) — полностью полагаться на блокировки.

Вы можете изменить ConcurrentHashMap на обычный HashMap.

Затем вы должны применить все свои изменения непосредственно в функции получения:

fun retrieveA(key: String, mod: (A) -> Unit): A {
    synchronized(lock) {
        val obj: A = cache.getOrPut(key) { A(key, 1) }
        mod(obj)
        cache.put(obj)
        return obj
    }
}

Надеюсь, он скомпилируется (я не эксперт по Kotlin).

Затем вы используете его как:

class CacheChanger {
    fun incrementData() {
        flusher.retrieveA("x") { it.changeA() }
    }
}

Хорошо, я признаю, что этот код на самом деле не Kotlin ;) вы должны использовать лямбду Kotlin вместо интерфейса Consumer. Прошло некоторое время с тех пор, как я немного поиграл с Kotlin. Если кто-то может исправить это, я был бы очень благодарен.