Ищете чистое решение для синхронизации событий с Java Concurrency API

Question

Я использую внешнюю платформу с системой Event, для простоты, скажем, она выглядит так:

public interface EventListener {
    public void eventA(int a);
    public void eventB(int a, long b);
}

Таким образом, свойства системы событий таковы, что после события A может существовать или не быть событие B для того же самого a. Но если для этого есть событие B, оно должно быть в пределах 3000 мс.

Я не уверен, какое чистое решение есть для этой проблемы, я посмотрел в пакете java.util.concurrency и не нашел ничего, что могло бы предложить хорошее чистое решение для этой проблемы. Я думал, что мог бы использовать Future<?>, но я не мог найти хорошее решение.

Я напишу в псевдокоде, что я хочу сделать:

public void eventA(int a) {
    // fired Event A

    if (within the next 3 seconds there was an eventB for _exactly_ this a)
        doSomething(a, b)
    else
        doSomething(a)
}

public void eventB(int a, long b) {

}

Пока что я использую ConcurrentHashmap<Int, Long>, где я сохраняю все комбинации a-b из события B, и проверяю, получал ли я в течение 101000 такое сочетание a-b, но я не очень доволен этим решением. Я чувствую, что должен быть более чистый путь с API параллелизма высшего уровня Java 1.5.

Я был бы рад любому совету о чистом решении этой проблемы.

Answer 1

Я не совсем уверен, нужно ли вам работать так, как это, но логика заключается в том, что при запуске eventB он добавляет данные в ConcurrentHashMap и планирует удалить себя через 3 секунды. Когда событие A запускается, оно удаляет данные с карты (таким образом очищая событие для этого ключа), а если событие было, оно вызывает doSomething (int, long), в противном случае он вызывает doSomething (int).

import java.util.concurrent.*;

public class EventListenerImpl implements EventListener {

    // You can change how many threads there are, or use a scheduler that's
    // already in the application.
    private final static ScheduledExecutorService scheduler = Executors
            .newScheduledThreadPool(1);

    private ConcurrentHashMap<Integer, Data> data = new ConcurrentHashMap<Integer, Data>();

    @Override
    public void eventA(int a) {

        Data b = data.remove(a);
        if (b != null) {
            doSomething(a, b.getB());
        } else {
            doSomething(a);
        }

    }

    @Override
    public void eventB(final int a, final long b) {

        final Data thisData = new Data(a);

        // Needs to be synchronized because it may be removed by a scheduled event
        synchronized (data) {
            Data d = data.put(a, thisData);
            if (d != null) {
                d.getFuture().cancel(true);
            }
        }

        thisData.setFuture(scheduler.schedule(new Runnable() {

            @Override
            public void run() {
                data.remove(a, thisData);
            }

        }, 3, TimeUnit.SECONDS));

    }

    private class Data {
        private long b;
        private ScheduledFuture<?> future;

        public Data(long b) {
            this.b = b;

        }

        public long getB() {
            return b;
        }

        public void setFuture(ScheduledFuture<?> future) {
            this.future = future;
        }

        public ScheduledFuture<?> getFuture() {
            return future;
        }
    }
}

Дай мне знать, если это не совсем то, что тебе нужно, потому что я немного растерялся и в итоге догадался, что ты пытался сделать. Например, если a не является каким-либо ключом, то это можно значительно упростить ... Это также можно упростить, если eventA ждет три секунды перед вызовом doSomething, а не отслеживает последние три секунды все время.

Answer 2

Как насчет использования карты, содержащей для каждого полученного события A объект (назовем его Foo), содержащий время, когда было получено A, и потенциал B (null изначально).

Когда вы получаете событие A, вы создаете экземпляр Foo с текущим временем и нулевым B, сохраняете его на карте и запускаете поток, который спит в течение 3 секунд.затем проверяет, содержит ли Foo, связанный с A, B, удаляет запись с карты и действует соответствующим образом (то есть вызывает doSomething с соответствующими аргументами).

Когда вы получаете B, вы проверяете, есть ли на карте Foo и был ли он получен менее 3 секунд (хотя это не должно происходить), и сохраняете B в Foo.

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