Netty Nio java中的通信

Question

我想在Netty nio中创建一个包含两个客户端和一个服务器的通信系统。更具体地说，首先，我希望当两个客户端与服务器连接时，从服务器发送消息，然后能够在两个客户端之间交换数据。我正在使用这个例子中提供的代码。我对代码的修改可以在这里找到:链接

似乎通道读取在服务器处理程序中工作，因此它始终返回 1，但当连接第二个客户端时，它不会更改为 2。当两个客户端都连接到服务器时，如何从服务器正确检查？如何从客户端的主要函数中动态读取此值？那么哪一种是让两个客户沟通的最佳方式呢？

EDIT1：显然，客户端服务似乎正在运行并直接关闭，所以每次我运行一个新的NettyClient时，都连接上了，但之后连接就关闭了。所以计数器总是从零到一的chnages。正如我在下面的评论中得到的建议，我在同一个端口上使用telnet进行了测试，计数器似乎正常增加，但是，使用NettyClient服务号。

EDIT2：我收到的问题似乎来自future.addListener（ChannelFutureListener.CLOSE）；，它位于ProcessingHandler类中的channelRead中。当我注释掉它时，似乎代码可以工作。但是，我不确定注释掉的后果是什么。此外，我想从客户端的主要功能中检查返回消息何时是特定的两个。如何创建一个方法来等待来自服务器的特定消息，同时它会阻止主要功能。

 static EventLoopGroup workerGroup = new NioEventLoopGroup();
 static Promise<Object> promise = workerGroup.next().newPromise(); 
 public static void callClient() throws Exception {
    String host = "localhost";
    int port = 8080;
    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 {
                ch.pipeline().addLast(new RequestDataEncoder(), new ResponseDataDecoder(), new ClientHandler(promise));
            }
        });
        ChannelFuture f = b.connect(host, port).sync();
    } finally {
        //workerGroup.shutdownGracefully();
    }
}

我希望在main函数中调用方法并返回结果，当它为2时，继续使用main函数。但是，我不能在while内调用callClient，因为它将在同一客户端上运行多次。

   callBack();
    while (true) {
        Object msg = promise.get();
        System.out.println("Case1: the connected clients is not two");
        int ret = Integer.parseInt(msg.toString());
        if (ret == 2){
            break;
        }
    }
   System.out.println("Case2: the connected clients is two");
   // proceed with the main functionality

如何更新第一个客户端的promise变量。当我运行两个客户端时，对于第一个客户端，我总是收到消息：

情况1：连接的客户端不是两个

似乎promise没有正常更新，而对于第二个客户端，我总是收到：

案例2：连接的客户端是两个

Answer 1

为什么ChannelGroup通道的大小总是一个。即使我连接了更多的客户端？

因为为每个新通道（客户端）调用子通道初始化程序。在那里，您正在创建处理处理程序的新实例，因此每个通道都会看到自己的通道组实例。

解决方案 1 - 通道属性

使用属性并将其与通道关联。

在某处创建属性（比如在＜code＞Constants＜/code＞class中）：

public static final AttributeKey<ChannelGroup> CH_GRP_ATTR = 
       AttributeKey.valueOf(SomeClass.class.getName());

现在，创建将被ProcessingHandler的所有实例使用的ChannelGroup：

final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

在NettyServer中更新您的孩子＜code＞ChannelInitializer＜/code＞：

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast(
        new RequestDecoder(), 
        new ResponseDataEncoder(), 
        new ProcessingHandler());

    ch.attr(Constants.CH_GRP_ATTR).set(channels);
}

现在您可以在处理程序中访问ChannelGroup的实例，如下所示：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    final ChannelGroup channels = ctx.channel().attr(Constants.CH_GRP_ATTR).get();
    channels.add(ctx.channel());

这将起作用，因为每次新客户端连接时，ChannelInitializer都将使用对ChannelGroup的相同引用进行调用。

解决方案2 -静态场

如果将＜code＞ChannelGroup＜/code＞声明为静态，则所有类实例都将看到相同的＜code＞ChannelGroup＞/code>实例：

private static final ChannelGroup channels =
     new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

解决方案3 -传播共享实例

将参数引入< code>ProcessingHandler的构造函数:

private final ChannelGroup channels;
public ProcessingHandler(ChannelGroup chg) {
    this.channels = chg;
}

现在，在NettyServer类中创建＜code＞ChannelGroup＜/code＞的实例并将其传播到ProcessingHandler构造函数：

final ChannelGroup channels = new 
      DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast(
        new RequestDecoder(), 
        new ResponseDataEncoder(), 
        new ProcessingHandler(channels)); // <- here
}

就我个人而言，我会选择第一种解决方案，因为

< li >它清楚地将频道组与频道上下文相关联 < li >您可以在其他处理程序中访问相同的channel group < li >您可以有多个服务器实例(在同一JVM中的不同端口上运行)

Answer 2

如果我的记忆正确，ChannelHandlerContext是每个通道一个，它的管道中可以有多个Channelhandler。通道变量是处理程序类的实例变量。然后为每个连接创建一个新的ProcessingHandler实例。因此，一旦初始化，每个通道在channels变量中都将有一个且只有一个连接，这就是它创建的目的。

请参阅服务器代码中 init 通道函数中的新处理处理程序（）（.java）。

您可以将通道变量设为静态，以便在ProcessingHandler实例之间共享。或者您可以在其他地方创建一个ProcessingHandler实例（例如，作为run（）函数中的局部变量），然后将该实例传递给addLast调用，而不是new ProcessingHandler（）。