Поток Java блокируется при регистрации канала с помощью селектора при вызове select(). Что делать?

У меня есть основной вопрос. Почему и как метод регистров SelectableChannel может быть при блокировке вызова. Позвольте мне представить сценарий.

Я создал объект Selector в классе Register следующим образом.

private static Selector selector = Selector.open();

У меня также есть метод в том же классе (Регистрация), чтобы зарегистрировать канал с помощью селектора.

public static SelectionKey registerChannel(SelectableChannel channel, int ops)
                             throws IOException {
   channel.configureBlocking(false);
   return channel.register(selector, ops);
}

И есть еще один класс с именем Request, у которого есть метод, который читает данные из каналов, обрабатывает и вызывает следующий метод для регистрации канала.

selectonKey = Register.register(socketChannel, SelectionKey.OP_READ);

Здесь на этом этапе поток заблокирован, не давая понять, что он ждет. Я проверил, что селектор открыт. Пожалуйста, предоставьте мне некоторую помощь, чтобы понять, как я могу решить эту проблему. Есть ли замок, который я могу снять.

Любой вклад будет оценен.

Добавление к тому, что я описал. Дальнейшие тесты показали, что если метод Register.register вызывается из того же потока, он может зарегистрироваться, но после этого, если какой-то другой поток пытается вызвать метод, поток не продвигается.

Ответ 1

Это основная особенность большинства реализаций NIO, которые не очевидны из документации.

Вам нужно сделать все вызовы регистраций из того же потока, который будет делать ваш выбор или взаимоблокировки. Обычно это делается путем предоставления очереди регистрации/дерегистраций/изменений процентов, которые записываются и вызывается selector.wakeup(). Когда пробущий поток пробуждается, он проверяет очередь и выполняет любые запрошенные операции.

Ответ 2

Вам нужно использовать блокировку и синхронизировать вручную.

В том же потоке, в котором вы запускаете цикл выбора, есть ReentrantLock:

final ReentrantLock selectorLock = new ReentrantLock();

Затем, когда вам нужно зарегистрироваться в селекторе, сделайте что-то вроде этого:

selectorLock.lock();
try {
    selector.wakeup();
    socketChannel.register(selector, ops);
} finally {
    selectorLock.unlock();
}

Наконец, во время вашего цикла, который вы вызываете accept(), что-то вроде этого:

selectorLock.lock();
selectorLock.unlock();

selector.select(500);

А затем продолжайте с остальной частью вашей логики.

Эта конструкция гарантирует, что вызов register() не будет блокироваться, гарантируя, что никогда не будет другого select() между соответствующими вызовами wakeup() и register().

Ответ 3

Вы пробовали печатать трассировку стека всех потоков в вашей программе (используя kill -QUIT в Unix или Ctrl + Break в Windows или с помощью утилиты jstack)?

AbstractSelectableChannel содержит блокировку, на которой необходимо синхронизировать configureBlocking и register. Эта блокировка также доступна через метод blockingLock(), и поэтому другой поток потенциально может удерживать блокировку, заставляя ваш вызов реестра блокироваться бесконечно (но без трассировки стека это трудно сказать).

Ответ 4

Если вы используете JConsole, вы можете обнаружить тупики со вкладки Threads.

Ответ 5

Зарегистрируйте свой канал из любого потока:

synchronized (selectorLock2) {
   selector.wakeup();
   synchronized (selectorLock1) {
       channel.register(selector, ops);
   }
}

Ваш цикл выбора должен выглядеть так:

while (true) {
   synchronized (selectorLock1) {
       selector.select();
   }
   synchronized (selectorLock2) {}

   ....
}

Ответ 6

Я согласен с ответом @Darron о том, что вы должны передавать register вызовы потоку селектора, но вы не должны использовать selector.wakeup поскольку он вводит условия гонки (представьте, что поток селектора занят обработкой других регистраций, и ваше wakeup не может разбудить кого-либо). К счастью, Java NIO предоставил Pipe так что вы можете позволить селектору прослушивать как register вызовы, так и другие события.

В основном вот что нужно сделать:

val registrationPipe = Pipe.open()
registrationPipe.source().configureBlocking(false)
registrationPipe.source().register(selector, SelectionKey.OP_READ)
// now start your selector thread

// now to register a call from other threads using message pleaseRegisterMe
registrationPipe.sink().write(pleaseRegisterMe)

// inside your selector thread
val selectionKey = iterator.next()
if (selectionKey.channel() === registrationPipe.source()) {
    registrationPipe.source().read(pleaseRegisterMe)
    // do something with the message pleaseRegisterMe and do the actual register
}

Вот полный рабочий пример.