Starting mupltiple threads and collect the results

user_10923401 0
Viewed 64

I have an Interface for searching in some subsystems gathered by an ENUM. The Interface looks like this:

public interface ReferenceController {

    public Map<String, ReferenceElement> searchElements(String searchField, List<String> searchItems, SystemStage systemStage) throws Exception;

    public Boolean isAvailable(SystemStage systemStage) throws Exception;

    public Boolean isAvailable(SystemStage systemStage) throws Exception;
}

And the ENUM looks like this

public enum ReferenceSystem implements ReferenceController{
    UCMDB    (UcmdbFunctions.class),
    PROIPS   (ProIPSFunctions.class),
    KV       (KvFunctions.class),
    FISERVICE(FiServiceFunctions.class),
    COMMAND  (CommandFunctions.class),
    FII          (FiiFunctions.class);

    private Class<? extends ReferenceController> clazz;

    private ReferenceSystem(Class<? extends ReferenceController> controllerClass) {
        this.clazz = controllerClass;
    }

    public String displayName() {
        return displayName(Locale.GERMAN); 
    }

    public String displayName(Locale locale) {
        ResourceBundle bundle = ResourceBundle.getBundle("EnumI18n", locale);
        return bundle.getString(toString()); 
    }

    public Class<? extends ReferenceController> getClassname() { return clazz; }

    @Override
    public Map<String, ReferenceElement> searchElements(String searchField, List<String> searchItems, SystemStage systemStage) throws Exception {
        Map<String, ReferenceElement> result = clazz.newInstance().searchElements(searchField, searchItems, systemStage);
        return result;
    }

    @Override
    public String getStateMapping(String value) {
        try {
            return clazz.newInstance().getStateMapping(value);
        } catch (IllegalAccessException | InstantiationException e) {
            return null;
        }
    }

    @Override
    public Boolean isAvailable(SystemStage systemStage) throws Exception {
        return clazz.newInstance().isAvailable(systemStage);
    }
}

At the moment I start the searches one after another. So my server hast to wait for a search to finish before it starts the next one. So the user has to wait very lang for the results to be presented. This code starts the search

    public static void performSingleSearch(ReferenceSystem referenceSystem, String searchField, List<String> searchValues, SystemStage systemStage) throws Exception {

        if(!isAvailable(referenceSystem, systemStage)) return;
        Map<String, ReferenceElement> result = new HashMap<>();
        try {
            result = referenceSystem.searchElements(searchField, searchValues, systemStage);
        } catch (Exception e) { 
            return;
        }
        if(result != null) orderResults(result, referenceSystem);
    }

The resultmap is the same object for all subsystems, so I need a system that all searches are started at once and are able to put their result into the result-object.

I hope it is possible to fill those objects nearly synchron so the User doesn't have to wait for all systems to finish.

Best regards Daniel

multithreading java
1 Answers

If you are willing to use Guava you can use ListeningExecutorService.submit to submit a Callable wrapping each call to searchElements and collect the returned ListenableFuture instances in a list. Using Futures.allAsList the list can be converted into a future that returns a list of all results if all futures complete successfully and that completes unsuccessfully otherwise.

For example:

<T> void submitAll(
        Collection<Callable<T>> callables,
        Consumer<List<T>> onSuccess,
        Consumer<Throwable> onError,
        ExecutorService executor) {

    ListeningExecutorService decoratedExecutor = MoreExecutors.listeningDecorator(executor);

    List<ListenableFuture<T>> futures = Lists.newArrayList();
    for (Callable<T> callable : callables) {
        futures.add(decoratedExecutor.submit(callable));
    }

    FutureCallback<? super List<T>> cb = new FutureCallback<List<T>>() {
        public void onSuccess(List<T> result) {
            onSuccess.accept(result);
        }

        public void onFailure(Throwable t) {
            onError.accept(t);
        }
    };

    Futures.addCallback(Futures.allAsList(futures), cb, executor);
}
user_4024280
Related