Java. Очереди потоков

27 декабря 2020, 10:40

Помогите составить архитектуру приложения.

Есть асинхронный сокет, т.е. передача в него команд и получение ответов происходит двумя разными потоками bufferedWriter и bufferedReader. Так вот вначале должна происходить инициализация подключения. т.е. вначале поднимается сокет:

Socket socket = new Socket();
socket.connect(socketAddress, timeoutInMs);

затем я отправляю на авторизацию параметры:

String request = "login and password";
BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
bufferedWriter.write(request);
bufferedWriter.flush();

и жду ответ в

BuffereReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));

Так вот, следом за авторизацией у меня отправляются туда несколько команд на выполнение разных задач. Для передачи в сокет этих команд, я использую

ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
threadPoolTaskExecutor.setCorePoolSize(1);
threadPoolTaskExecutor.setQueueCapacity(Integer.MAX_VALUE);
threadPoolTaskExecutor.setKeepAliveSeconds(1);
threadPoolTaskExecutor.initialize();
threadPoolTaskExecutor.submit(cmd1);
threadPoolTaskExecutor.submit(cmd2);
threadPoolTaskExecutor.submit(cmd3);

Объект-держатель сокета у меня ConnectorImpl.class в нем есть переменная

private boolean state;

которая обозначает залогинился я в сокете или нет.

ВОПРОС.

Как мне реализовать, чтоб выполнение команд из threadPoolTaskExecutor происходило только при состоянии state == true?

Причем, мне нужно, чтоб команда попадала в очередь вне зависимости от состояния подключения (происходило накопление команд), а исполнялась только при state == true.

Чем лучше проверять или может какую другую очередь задач использовать?

Answer 1

Вот тут есть пример кастомного ScheduledThreadPoolExecutor, который можно поставить на паузу. Как вариант использовать статус непосредственно самого пула вместо переменной.

import com.google.common.util.concurrent.Monitor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
public class PausableExecutor extends ScheduledThreadPoolExecutor {
    private boolean isPaused;
    private final Monitor monitor = new Monitor();
    private final Monitor.Guard paused = new Monitor.Guard(monitor) {
        @Override
        public boolean isSatisfied() {
            return isPaused;
        }
    };
    private final Monitor.Guard notPaused = new Monitor.Guard(monitor) {
        @Override
        public boolean isSatisfied() {
            return !isPaused;
        }
    };
    public PausableExecutor(int corePoolSize, ThreadFactory threadFactory) {
        super(corePoolSize, threadFactory);
    }
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        monitor.enterWhenUninterruptibly(notPaused);
        try {
            monitor.waitForUninterruptibly(notPaused);
        } finally {
            monitor.leave();
        }
    }
    public void pause() {
        monitor.enterIf(notPaused);
        try {
            isPaused = true;
        } finally {
            monitor.leave();
        }
    }
    public void resume() {
        monitor.enterIf(paused);
        try {
            isPaused = false;
        } finally {
            monitor.leave();
        }
    }
}