java - Netty 服务器停止响应

Question

我有ServerHandshakeHandler，它扩展了ChannelInboundHandlerAdapter。客户端模拟对服务器的多次访问。成功通信一段时间后，当客户端尝试连接时，服务器停止响应。它不显示任何传入连接。客户端重启没有帮助，只能重启服务器。

我尝试在服务器停止响应时设置telnet连接:连接已建立，但我无法从服务器获得任何响应(当服务器处于正常状态时，它会发送响应)。与 nmap -v --packet-trace -sT localhost -p {port} 的情况类似:nmap 发现端口已打开，但服务器上没有有关传入连接的日志信息。

服务器:

public class ServerHandshakeHandler extends ChannelInboundHandlerAdapter {

private final ChannelGroup group;
private static final byte HANDSHAKE_SUCCEDED = 1;
private static final byte HANDSHAKE_FAILED = 0;
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());


public ServerHandshakeHandler(ChannelGroup group) {
    this.group = group;
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    LOG.debug("in ServerHandshakeHandler.channelRead");
    ByteBuf buf = (ByteBuf) msg;
    String someField = getSomeField(buf);
    ReferenceCountUtil.release(msg);

    if (someField.isEmpty()) {
        this.fireHandshakeFailed(ctx);
        return;
    }

    LOG.debug("Removing handshake handler from pipeline.");

    ctx.pipeline().remove(this);
    this.fireHandshakeSucceeded(ctx);

}

@Override
public void channelActive(final ChannelHandlerContext ctx) {
    LOG.debug("in ServerHandshakeHandler.channelActive, group size = " + this.group.size());
    this.group.add(ctx.channel());
    LOG.debug("Incoming connection from: {}",
            ctx.channel().remoteAddress().toString());
}

@Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    LOG.error("exception caught ", cause);
    if (ctx.channel().isActive()) {
        ctx.channel().close();
    } else {
        this.fireHandshakeFailed(ctx);
    }
}

private void fireHandshakeFailed(ChannelHandlerContext ctx) {
    LOG.debug("fire handshake failed");
    ByteBuf buf = Unpooled.buffer(1);
    buf.writeByte(HANDSHAKE_FAILED);
    ctx.channel().writeAndFlush(buf);

    ctx.channel().close();
    ctx.fireUserEventTriggered(HandshakeEvent.handshakeFailed(ctx.channel()));
}

private void fireHandshakeSucceeded(ChannelHandlerContext ctx) {
    LOG.debug("fire handshake succeded");
    ByteBuf buf = Unpooled.buffer(1);
    buf.writeByte(HANDSHAKE_SUCCEDED);
    ctx.channel().writeAndFlush(buf);

    ctx.fireUserEventTriggered(HandshakeEvent
            .handshakeSucceeded(ctx.channel()));
}

}

客户:

public class MyClient {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private String host;
private int port;
private Socket socket;

public Client(String host, int port) {
    this.host = host;
    this.port = port;
}

public void send(String id, String message) {
    try {
        socket = new Socket(host, port);
        LOG.debug("connected to server");

        if (performHandshake(id)) {
            LOG.debug("handshake success");
            sendMessage(message);
        }
        socket.close();;
    } catch (IOException ex) {
        LOG.error("error while sending data", ex);
    }
}

private boolean performHandshake(String id) {
    try {
        byte[] request = handshakeRequest(id);
        writeBytes(request);
        byte[] response = readBytes(1);
        return (response != null && response.length == 1 && response[0] == 1);
    } catch (IOException ex) {
        LOG.error("perform handshake error", ex);
        return false;
    }
}

private byte[] handshakeRequest(String id) throws UnsupportedEncodingException {...}

private void writeBytes(byte[] data) throws IOException {
    OutputStream out = socket.getOutputStream();
    out.write(data);
}

private byte[] readBytes(int length) throws IOException {
    InputStream in = socket.getInputStream();
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    byte buffer[] = new byte[1024];
    int currentLength = 0;
    while (currentLength < length) {
        int size = in.read(buffer); //here client stops waiting server response
        if (size == -1) {
            throw new IOException("unexpected end of stream");
        }
        baos.write(buffer, 0, size);
        currentLength += size;
    }
    return baos.toByteArray();
}

}

Answer 1

解决了!有一段狭窄的代码，我在其中调用与数据库连接的同步函数。由于某些原因无法建立此连接并且该功能挂起。该函数中的线程进入 WAITING状态。一段时间后，其他进程尝试访问此功能并被阻止。这就是服务器停止处理传入连接的原因。

我推荐jvisualvm分析工具，它帮助我找到了这个bug。