Множество нишки чакат групова операция

Имам процес, който изпраща множество задачи в ExecutorService, да речем MyTask. Моята задача изисква стойност от външна услуга, да кажем ExternalService. Това, което се опитвам да направя, е ExternalService да обработва групово заявките, да кажем на всеки 100 входящи или на всяка 1 секунда и до този момент да накара нишките MyTask да бъдат задържани, докато не получат отговор:

public class MyTask implements Runnable {

    @Override
    public void run() {
        try {
            // .... code ....


            ExternalData data = externalService.getData(id);

            // ..... code after batch ...

        }catch (Exception e){

        }
    }
}

Имам нужда от услуга externalService, за да задържа MyTask, докато изпълни своята операция групово (чрез натрупване на 100 заявки или като изчака 1 секунда) и върне резултата на задачата продължавам.

Какъв е най-добрият начин за подход към този проблем?

Благодаря ти


java multithreading concurrency batch-processing
person tbo    schedule 24.01.2015    source източник

Отговори (1)


arrow_upward
1
arrow_downward

Това може да се постигне чрез използване на CyclicBarrier.

CyclicBarriers са полезни в програми, включващи група нишки с фиксиран размер, които понякога трябва да чакат една друга.

Създайте бариера, до която всички задачи имат достъп. Аргументите са количеството задачи, които искате да изчакате, когато извиквате await() и действието, което трябва да се извърши след изчакване:

CyclicBarrier barrier = new CyclicBarrier(100, action);

Всяка задача трябва да извиква await. Това ще изчака, докато общо 100 задачи достигнат тази точка или до 1 секунда:

barrier.await(1, TimeUnit.SECONDS);

В случай на изчакване, изпълнете action въпреки това. action е изпълняващ се файл, който извиква външната услуга. Тъй като action работи едновременно, вероятно също трябва да накарате задачите си да изчакат да приключи и тогава да получите резултатите.

За да получа действителните резултати от външната услуга, вероятно бих използвал нещо като следното. По принцип запазете ID в списък и го направете блокиран, докато резултатът е готов.

action.addId(id);
barrier.await(...); // run batch action using all IDs
result = action.getResult(id); // blocks until result is ready
person kapex    schedule 24.01.2015
comment
Това е хубава функция, за която не знаех, така че ако моята външна услуга е действието, което предавам на цикличната бариера, как всички нишки ще вземат резултата от изпълнението и ще продължат? всеки ще има различен идентификатор, благодаря ви отново - person tbo; 25.01.2015
comment
@tbo Действието ще бъде обвивка около вашата услуга, която също обработва идентификатори и отговор. Добавих пример. Мисля, че този подход по принцип трябва да работи, но съм сигурен, че все още липсват някои неща и също така се нуждае от внимателно обработване на грешки. - person kapex; 25.01.2015
comment
На този етап мисля, че този подход е наистина близък до това, което се опитвам да направя. Бих предпочел да не изпълнявам кода си при улавяне, тъй като ако бъде достигнат лимитът на времето за изчакване, се хвърля TimeoutException и действието не се изпълнява и други нишки получават изключение brokeBarrier, след като каза, че трябва да уловя изключенията и да действам съответно, за да постигна моя сценарий - person tbo; 25.01.2015