Могу ли я синхронизировать метод по параметру

Question

Можно ли синхронизировать метод по параметру?

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

private void dosomething(Long id, Person person) {
    dosomethingelse(id, person);
}

Как вызывать dosomethingelse (id, person) только для одного и того же идентификатора один за другим?но я хочу, чтобы этот код для разных идентификаторов можно было назвать многопоточным

Я написал этот код, но, возможно, что-то здесь не так или что-то может быть лучше.

public static class LatchByValue <T> {
    public void latch(T value, ConsumerWithException<T> consummer) throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        try {
            CountDownLatch previousLatch = null;
            // we are checking if another thread is already calling this method with the same id
            // if sync has CountDownLatch so another thread is already calling this method 
            // or we put our latch and go on
            while ((previousLatch = sync.putIfAbsent(value, latch)) != null) {
                try {
                    // we are waiting for another thread, we are waiting for all threads that put their latch before our thread
                    previousLatch.await();
                } catch (InterruptedException e) {
                    return;
                }
            }
            consummer.accept(value);
        } finally {
            latch.countDown();
            sync.remove(value, latch);
        } 
    }
    private ConcurrentHashMap<T, CountDownLatch> sync = new ConcurrentHashMap<>();
}

Пример:

LatchByValue<Long> latch = new LatchByValue<>();

private void dosomething(Long id, Person person) {
     latch.latch(
        id,
        currentId -> { dosomethingelse(currentId, person); }
     );
}

Answer 1

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

Таким образом, реализация будет выглядеть следующим образом:

private void doSomething(Long id, Person person) {
  synchronized (person) {
    // do something
  }
}

Помните, что любые другие доступы (не в doSomething вызов) также должен иметь блок синхронизации, например:

// another method, unrelated, but does something with 'person'
private void doSomethingElse(Person person, ... /* other arguments */) {
  synchronized (person) {
    // do something
  }
}

Это был бы хороший документ (в javadoc Person), который требуется пользователю для получения блокировки для этого объекта.

---

Если вы хотите предоставить критическую секцию для кортежа <id, person>, вам нужно немного изменить свой API, а затем передать этот объект в ваше приложение.

private void doSomething(IdAndPerson idAndPerson) {
  synchronized (idAndPerson) {
    // do something
  }
}

class IdAndPerson {
    private final Long id;
    private final Person person;
    // constructor etc.
}

Answer 2

    private static final Set<Long> lockedIds = new HashSet<>();

    private void lock(Long id) throws InterruptedException {
        synchronized (lockedIds) {
            while (!lockedIds.add(id)) {
                lockedIds.wait();
            }
        }
    }

    private void unlock(Long id) {
        synchronized (lockedIds) {
            lockedIds.remove(id);
            lockedIds.notifyAll();
        }
    }

    public void doSomething(Long id) throws InterruptedException {
        try {
            lock(id);

            //Put your code here.
            //For different ids it is executed in parallel.
            //For equal ids it is executed synchronously.

        } finally {
            unlock(id);
        }
    }

• id может быть не только классом 'Long', но и любым классом с правильно переопределенными методами 'equals' и 'hashCode'.
• try-наконец - это очень важно - вы должны гарантировать разблокировку ожидающих потоков после вашей операции, даже если ваша операция вызвала исключение.
• Это не будет работать, если ваш бэкэнд распределен по нескольким серверам /JVM .

Answer 3

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

Вместо этого вы можете использовать a Semaphore с одним разрешением, которое позволит вам сделать то же самое, но более простым способом.

Semaphore s = sync.computeIfAbsent(value, x -> new Semaphore(1, true));
s.acquire(); //this blocks and throws InterruptedException, which you need to handle
try {
  consummer.accept(value);
} finally {
  s.release();
}