如何用netty客户端获取服务器响应

Question

public class Client {
public static void main(String[] args) throws InterruptedException {
    Client client = new Client();

}

private Channel channel;

public Client() throws InterruptedException {
    EventLoopGroup loopGroup = new NioEventLoopGroup();

    Bootstrap b = new Bootstrap();
    b.group(loopGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new StringDecoder()).
                    addLast(new StringEncoder()).
                    addLast(new ClientHandler());
        }
    });
    channel = b.connect("localhost", 9091).sync().channel();
}

public String sendMessage(String msg) {
    channel.writeAndFlush(msg);
    return ??????????;
}

在调用writeAndFlush（）之后，我不知道如何从服务器检索响应；我该怎么办？

我也使用Netty 4.0.18.final

Answer 1

为该方法返回future 很简单，我们将实现以下方法签名：

public Futute<String> sendMessage(String msg) {

当您了解异步编程结构时，这样做相对容易。为了解决设计问题，我们将执行以下步骤：

>

写入消息时，将promise 添加到ArrayBlockingQueue

这将作为最近发送的消息的列表，并允许我们更改未来 对象返回结果。

当消息返回到处理程序时，根据队列的头部解析它

这让我们得到正确的未来去改变。

我们调用promise.setsuccess()来最终设置对象的状态，这将传播回未来的对象。

public class ClientHandler extends SimpleChannelInboundHandler<String> {
    private ChannelHandlerContext ctx;
    private BlockingQueue<Promise<String>> messageList = new ArrayBlockingQueue<>(16);

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

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        super.channelInactive(ctx);
        synchronized(this){
            Promise<String> prom;
            while((prom = messageList.poll()) != null) 
                prom.setFailure(new IOException("Connection lost"));
            messageList = null;
        }
    }

    public Future<String> sendMessage(String message) {
        if(ctx == null) 
            throw new IllegalStateException();
        return sendMessage(message, ctx.executor().newPromise());
    }

    public Future<String> sendMessage(String message, Promise<String> prom) {
        synchronized(this){
            if(messageList == null) {
                // Connection closed
                prom.setFailure(new IllegalStateException());
            } else if(messageList.offer(prom)) { 
                // Connection open and message accepted
                ctx.writeAndFlush(message).addListener();
            } else { 
                // Connection open and message rejected
                prom.setFailure(new BufferOverflowException());
            }
            return prom;
        }
    }
    @Override
    protected void messageReceived(ChannelHandlerContext ctx, String msg) {
        synchronized(this){
            if(messageList != null) {
                 messageList.poll().setSuccess(msg);
            }
        }
    }
}

>

专用ChannelHandlerContext CTX；

用于存储对ChannelHandlerContext的引用，我们使用它来创建promise

public void channelActive（ChannelHandlerContext ctx）

当连接处于活动状态时由netty调用。在这里初始化变量。

public void channelInactive（ChannelHandlerContext ctx）

netty在由于错误或正常连接关闭而导致连接处于非活动状态时调用。

protected void messageReceed（ChannelHandlerContext ctx,String msg）

当新消息到达时由netty调用，在这里选择队列的头部，然后对其调用setsuccess。