带 POJO 的 Netty 乒乓球答案

【问题标题】：Netty ping pong with a POJO带 POJO 的 Netty 乒乓球
【发布时间】：2014-09-26 10:42:32
【问题描述】：

为什么客户端收不到服务器发送的POJO？

这个blog sample 对我来说有点难以理解。

我意识到这是很多代码，但我不知道如何在仍然使用 POJO（在本例中为 Quote）的同时在客户端和服务器之间进行精简。服务器在建立连接时发送报价：

run:
     [java] Aug 03, 2014 5:32:20 PM net.bounceme.dur.netty.QuoteServerInitializer <init>
     [java] INFO: ..initializing..
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteServerInitializer initChannel
     [java] INFO: ..adding to pipeline..
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteServerHandler <init>
     [java] INFO: ..started..
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteServerHandler channelActive
     [java] INFO: ..sending new server Quote..
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteEncoder encode
     [java] INFO: 
     [java] 
     [java] id      0
     [java] quote   Where there is love there is life.
^Cthufir@dur:~/NetBeansProjects/QuoteServer$ 
thufir@dur:~/NetBeansProjects/QuoteServer$ ^C
thufir@dur:~/NetBeansProjects/QuoteServer$

但它似乎永远不会到达客户端：

run:
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteClientInitializer <init>
     [java] INFO: ..initializing..
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteClientHandler channelActive
     [java] INFO: ..sending new client Quote..
     [java] Aug 03, 2014 5:32:23 PM net.bounceme.dur.netty.QuoteEncoder encode
     [java] INFO: 
     [java] 
     [java] id      0
     [java] quote   client
^Cthufir@dur:~/NetBeansProjects/QuoteClient$ 
thufir@dur:~/NetBeansProjects/QuoteClient$

同样，客户端发送的报价似乎永远不会到达服务器。为什么？

服务器：

package net.bounceme.dur.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.util.logging.Logger;

public final class QuoteServer {

    private static final Logger log = Logger.getLogger(QuoteServer.class.getName());

    public static void main(String... args) throws InterruptedException {
        MyProps p = new MyProps();
        int port = p.getServerPort();
        new QuoteServer().pingPong(port);
    }

    private void pingPong(int port) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(new QuoteServerInitializer());
            b.bind(port).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务器初始化程序：包 net.bounceme.dur.netty;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import java.util.logging.Logger;

public class QuoteServerInitializer extends ChannelInitializer<SocketChannel> {

    private static final Logger log = Logger.getLogger(QuoteServerInitializer.class.getName());

    public QuoteServerInitializer() {
        log.info("..initializing..");
    }

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        log.info("..adding to pipeline..");
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new QuoteDecoder());
        pipeline.addLast(new QuoteEncoder());
        pipeline.addLast(new QuoteServerHandler());
    }
}

服务器处理程序：
包 net.bounceme.dur.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.util.Random;
import java.util.logging.Logger;
import net.bounceme.dur.jdbc.Quote;

public class QuoteServerHandler extends SimpleChannelInboundHandler<Quote> {

    private static final Logger log = Logger.getLogger(QuoteServerHandler.class.getName());
    private static final Random random = new Random();

    public QuoteServerHandler() {
        log.info("..started..");
    }

    // Quotes from Mohandas K. Gandhi:
    private static final String[] quotes = {
        "Where there is love there is life.",
        "First they ignore you, then they laugh at you, then they fight you, then you win.",
        "Be the change you want to see in the world.",
        "The weak can never forgive. Forgiveness is the attribute of the strong.",};

    private static Quote nextQuote() {
        int quoteId;
        synchronized (random) {
            quoteId = random.nextInt(quotes.length);
        }
        return new Quote(quotes[quoteId]);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("..sending new server Quote..");
        ctx.writeAndFlush(nextQuote());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext chc, Quote quote) throws Exception {
        log.info(quote.toString());
        chc.writeAndFlush(nextQuote());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        log.info(msg.toString());
        ctx.writeAndFlush(nextQuote());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

}

客户：包 net.bounceme.dur.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.io.IOException;
import java.util.logging.Logger;

public final class QuoteClient {

    private static final Logger log = Logger.getLogger(QuoteClient.class.getName());

    public static void main(String... args) throws InterruptedException, IOException {
        new QuoteClient().connect();
    }

    public void connect() throws InterruptedException, IOException {
        MyProps p = new MyProps();
        String host = p.getHost();
        int port = p.getServerPort();
        pingPong(host, port);
    }

    public void pingPong(String host, int port) throws InterruptedException, IOException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new QuoteClientInitializer());
            ChannelFuture cf = b.connect(host, port);
            cf.sync().channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

客户端初始化程序：包 net.bounceme.dur.netty;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import java.util.logging.Logger;

public class QuoteClientInitializer extends ChannelInitializer<SocketChannel> {

    private static final Logger log = Logger.getLogger(QuoteClientInitializer.class.getName());

    public QuoteClientInitializer() {
        log.info("..initializing..");
    }

    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new QuoteDecoder());
        pipeline.addLast(new QuoteEncoder());
        pipeline.addLast(new QuoteClientHandler());
    }
}

客户端处理程序：包 net.bounceme.dur.netty;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.util.logging.Logger;
import net.bounceme.dur.jdbc.Quote;

@Sharable
public class QuoteClientHandler extends SimpleChannelInboundHandler<Quote> {

    private static final Logger log = Logger.getLogger(QuoteClient.class.getName());

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("..sending new client Quote..");
        ctx.writeAndFlush(new Quote("client"));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext chc, Quote quote) throws Exception {
        log.info(quote.toString());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        log.info(msg.toString());
        ctx.writeAndFlush(new Quote("client"));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.fireChannelReadComplete();
    }

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

}

解码器：包 net.bounceme.dur.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import java.util.List;
import java.util.logging.Logger;
import net.bounceme.dur.jdbc.Quote;

public class QuoteDecoder extends MessageToMessageDecoder<Quote> {

    private static final Logger log = Logger.getLogger(QuoteDecoder.class.getName());

    @Override
    protected void decode(ChannelHandlerContext chc, Quote quote, List<Object> list) throws Exception {
        log.info(quote.toString());
        list.add(quote);
    }
}

编码器：包 net.bounceme.dur.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;
import java.util.List;
import java.util.logging.Logger;
import net.bounceme.dur.jdbc.Quote;

public class QuoteEncoder extends MessageToMessageEncoder<Quote> {

            private static final Logger log = Logger.getLogger(QuoteEncoder.class.getName());


    @Override
    protected void encode(ChannelHandlerContext chc, Quote quote, List<Object> list) throws Exception {
        log.info(quote.toString());
        list.add(quote);
    }
}

值得注意的是，en/de-code 方法从不登录到控制台。

【问题讨论】：

请添加您的 QuoteDecoder 和 Encoder 的代码。
我添加了它们，但请记住，它们从不执行——或者至少它们从不将输出记录到控制台，如果它们的 en/de-代码方法运行。

标签： java io client-server netty nio

【解决方案1】：

如果您将 QuoteServerHandler 的 channelActive 方法编辑为以下内容：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    log.info("..sending new server Quote..");
    ChannelFuture cf = ctx.writeAndFlush(nextQuote());
    if (!cf.isSuccess()){
        log.log(Level.SEVERE, cf.toString());
    }
    ctx.fireChannelActive();
}

那么您很可能会收到一条消息：unsupported message type: Quote

您的编码器需要将其编码为受支持的内容。我现在不知道那会是什么。

我建议使用ObjectEncoder，它可以将您的Quote 编码为ByteBuf。

在接收站点上，您需要一个ObjectDecoder。之后，您可以将您在 ClientHandler 中收到的 msg 转换回 Quote。

【讨论】：