Netty 不使用 for 循环向服务器发送消息

Question

我使用以下代码将消息从客户端发送到服务器：

客户端类：

public class Client {

  String host = "localhost";
  int port = 14930;
  private final ClientHandler clientHandler = new ClientHandler();

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

  public void run() throws Exception {

    try {
      workerGroup = new NioEventLoopGroup();
      Bootstrap b = new Bootstrap();
      b.group(workerGroup);
      b.channel(NioSocketChannel.class);
      b.option(ChannelOption.SO_KEEPALIVE, true);
      b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
          public void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline pipeline = ch.pipeline();
          pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
          pipeline.addLast(new StringDecoder());
          pipeline.addLast(new StringEncoder());
          pipeline.addLast(clientHandler);
        }
      }
      );

      ChannelFuture f = b.connect(host, port).sync();

      f.channel().closeFuture().sync();
    }
    finally {
      workerGroup.shutdownGracefully();
    }
  }
  public void writeMessage(String msg) {
    clientHandler.sendMessage(msg);
  }
}

处理程序类：

public class ClientHandler extends SimpleChannelInboundHandler<String> {
  ChannelHandlerContext ctx;

  public void sendMessage(String msgToSend) {
    ctx.writeAndFlush(Unpooled.copiedBuffer(msgToSend, CharsetUtil.UTF_8));
  }

  @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
    this.ctx = ctx;
  }

  @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    ctx.close();
  }

  @Override
    protected void channelRead0(ChannelHandlerContext arg0, String msg) throws Exception {
  }

  void channelInactive(ChannelHandlerContext ctx) throws Exception {
  }
}

我正在这样创建客户端：

Client client;
client = new Client(ipAddress, serverPort);
client.run();
client.writeMessage("random_text");

服务器：

public final class ChatServer {

  public ChatServer(int PORT) throws Exception {

    bossGroup = new NioEventLoopGroup();
    workerGroup = new NioEventLoopGroup();

    try {
      ServerBootstrap b = new ServerBootstrap();
      b.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
          .handler(new LoggingHandler(LogLevel.INFO))
            .childHandler(new ChatServerInitializer());

      b.bind(PORT).sync().channel().closeFuture().sync();
    } 
    finally {
      bossGroup.shutdownGracefully();
      workerGroup.shutdownGracefully();
    }
  }
}

服务器处理程序：

public class ChatServerHandler extends SimpleChannelInboundHandler<String> {

  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    Channel incoming = ctx.channel();
    channels.add(ctx.channel());
  }

  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    Channel incoming = ctx.channel();
    channels.remove(ctx.channel());
  } 

  @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    ctx.close();
  }

  @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
    println(msg);
  }
}

从服务器向客户端发送消息的方法：

public void sendMessage(String message) {
  if (message != "" && message != null) {
    for (Channel c : channels) {
      c.writeAndFlush(message + "\r\n");
    }
  }
}

问题是服务器没有收到任何消息。我尝试了一些调试步骤，服务器将客户端添加到其通道中，并且 channelActive 在客户端成功执行。

Answer 1

您应该尝试将通信传输到处理程序（您的ChatClientInitializer 或您最后添加的额外处理程序），而不是一直轮询变量。给这个处理程序一个你在输入新消息时调用的方法。然后处理程序写入通道。

也许这个例子有帮助：

public class Client implements Runnable {

    String host = "localhost";
    int port = 9128;
    private final ClientHandler clientHandler = new ClientHandler();
    private boolean isRunning = false;
    private ExecutorService executor = null;


    public static void main(String[] args) {
        Client client = new Client();
        client.startClient();
        client.writeMessage("random_text");
        //client.stopClient();  //call this at some point to shutdown the client
    }

    public synchronized void startClient() {
        if (!isRunning) {
            executor = Executors.newFixedThreadPool(1);
            executor.execute(this);
            isRunning = true;
        }
    }

    public synchronized boolean stopClient() {
        boolean bReturn = true;
        if (isRunning) {
            if (executor != null) {
                executor.shutdown();
                try {
                    executor.shutdownNow();
                    if (executor.awaitTermination(calcTime(10, 0.66667), TimeUnit.SECONDS)) {
                        if (!executor.awaitTermination(calcTime(10, 0.33334), TimeUnit.SECONDS)) {
                            bReturn = false;
                        }
                    }
                } catch (InterruptedException ie) {
                    executor.shutdownNow();
                    Thread.currentThread().interrupt();
                } finally {
                    executor = null;
                }
            }
            isRunning = false;
        }
        return bReturn;
    }

    private long calcTime(int nTime, double dValue) {
        return (long) ((double) nTime * dValue);
    }

    @Override
    public void run() {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.SO_KEEPALIVE, true);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline(); 
                    pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                    pipeline.addLast(new StringDecoder());
                    pipeline.addLast(new StringEncoder());
                    pipeline.addLast(clientHandler);
                }
            });

            ChannelFuture f = b.connect(host, port).sync();

            f.channel().closeFuture().sync();
        } catch (InterruptedException ex) {
            // do nothing
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    public void writeMessage(String msg) {
        clientHandler.sendMessage(msg);
    }
}

一个非常基本的ClientHandler：

public class ClientHandler extends SimpleChannelInboundHandler<String> {
    ChannelHandlerContext ctx;

    public void sendMessage(String msgToSend) {
        if (ctx != null) {
            ChannelFuture cf = ctx.write(Unpooled.copiedBuffer(msgToSend, CharsetUtil.UTF_8));
            ctx.flush();
            if (!cf.isSuccess()) {
                System.out.println("Send failed: " + cf.cause());
            }
        } else {
            //ctx not initialized yet. you were too fast. do something here
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext arg0, String msg) throws Exception {
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    }
}