Я использую простой потребитель для kafka, такой как это:
int timeout = 80000;
int bufferSize = 64*1024;
consumer = new SimpleConsumer(host, port,timeout, bufferSize, clientName);
Это работает отлично в течение нескольких часов, но я получаю исключение позже на kafka.consumer.SimpleConsumer: Повторно подключиться из-за ошибки сокета:
java.nio.channels.ClosedChannelException
и потребители останавливаются... кто-нибудь сталкивался с этой проблемой раньше?
Немного другой вопрос, но, возможно, с той же основной причиной и решением был дан ответ, соответствующая часть:
Вы закрыли канал и все еще пытаетесь его использовать.
Есть несколько проблем с вашим кодом.
Во-первых, ваш тест на EOS неисправен. Убрать ограничение() == 0 test. Это не указывает на EOS, это просто указывает на чтение нулевой длины, которое может происходить в неблокирующем режиме в любое время. Это не значит, что пир закрыл свой конец соединения, и это не значит, что вы должны закрыть свой конец.
Во-вторых, закрытие канала также закрывает сокет. Вы должны закрыть только канал, а не сокет.
В-третьих, закрытие канала отменяет ключ. Вам не нужно следовать каждому закрытию с отменой.
Возможно, вам также не удалось проверить, является ли готовый ключ действительным в цикле выбора, прежде чем использовать его, например, для чтения.