问同步不同的任务(I/O侦听守护进程、提示调度程序、输出等)

EN

private void runFuturesConsumer() {
    new Thread(() -> {
        try {
            while (true) {
                Future<Integer> consumedFuture = completionService.take();
                Integer consumedValue = consumedFuture.get();

                if (Objects.equals(consumedValue, INTERRUPTION_VALUE)) {
                    break;  // clients were interrupted, stop futures consuming
                }           // happens after cancelServerFuture procedure call

                valuesContainer.add(consumedValue);

                if (consumedValue == SHORT_CIRCUIT_CONDITION) {
                    cancelServerFuture();
                    transferResult(true, null);
                    break;
                }

                if (valuesContainer.size() == clientsNumber) {
                    transferResult(false, null);
                    break;
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }).start();
}

private void transferResult(boolean isShortCircuited, String failureReport) {

    // simple lock/sync will produce a deadlock (if computations were completed during the prompt)
    if (outputLock.tryLock()) {
        try {
            if (failureReport != null) {
                serverListener.onFailureReported(failureReport);
            } else if (isShortCircuited) {
                serverListener.onCompletedComputation(SHORT_CIRCUIT_CONDITION, true);
            } else {
                serverListener.onCompletedComputation(valuesContainer.stream()
                        .reduce(1, (accumulator, elem) -> accumulator * elem), false);
            }
        } finally {
            cancelServerFuture();
            outputLock.unlock();
        }
    }
}

void stopServer() {
    if (outputLock.tryLock()) {
        transferResult(false, "stopped before the completion");
    }
}

private void runKeyPressDaemon() {
    Thread thread = new Thread(() -> {
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));

        String message = null;
        while (true) {
            try {
                message = bufferedReader.readLine();
            } catch (IOException e) {
                e.printStackTrace();
            }

            if (message != null && message.equals("q")) {
                // it's definitely bad if scheduler decides to reschedule here
                // because output lock happens only inside stop server method
                // (trying to lock right after readLine procedure creates possibility of losing the result)
                server.stopServer();
                break;
            }
        }
    });

    thread.setDaemon(true);
    thread.start();
}

private void runPromptScheduler() {
    scheduledExecutor.scheduleWithFixedDelay(this::displayPrompt, INITIAL_DELAY, SCHEDULER_PERIOD, TimeUnit.SECONDS);
}

private void displayPrompt() {

    promptLock.lock();
    try {
        Scanner scanner = new Scanner(System.in);

        System.out.print("continue(1), continue without prompt(2), cancel(3): ");
        int readValue = scanner.nextInt();

        switch (readValue) {
            case 1:
                handlePrompt(PromptOptions.CONTINUE);
                break;
            case 2:
                handlePrompt(PromptOptions.CONTINUE_WITHOUT_PROMPT);
                break;
            case 3:
                handlePrompt(PromptOptions.CANCEL);
                break;
        }
    } finally {
        promptLock.unlock();
    }
}

private void handlePrompt(PromptOptions promptOptions) {
    switch (promptOptions) {
        case CONTINUE:
            break;
        case CONTINUE_WITHOUT_PROMPT:
            scheduledExecutor.shutdownNow();
            break;
        case CANCEL:
            server.stopServer();
            break;
    }
}

@Override
public void onCompletedComputation(int result, boolean isShortCircuited) {

    promptLock.lock();
    try {
        if (isShortCircuited) {
            System.out.println("result (short-circuit): " + result);
        } else {
            System.out.println("result: " + result);
        }
    } finally {
        scheduledExecutor.shutdownNow();
        promptLock.unlock();
    }
}

@Override
public void onFailureReported(String cause) {

    promptLock.lock();
    try {
        System.out.println("failure caused by: " + cause);
    } finally {
        scheduledExecutor.shutdownNow();
        promptLock.unlock();
    }
}