java - StringEncoder 和 StringDecoder 在 netty 服务器中的正确用法是什么？

Question

下面基于字符串的服务器在功能上等同于它的数据报服务器兄弟吗？

我看到的唯一显着差异，也是我试图实现的唯一差异，是从 NioDatagramChannel 到 nioServerSocketChannel。处理程序之间存在一些细微差别，但是两个处理程序都会使用 nextQuote() 响应“QOTM”吗？

为了简洁和理智，我不能包括客户。我不熟悉 netty，而且我能找到的关于这个主题的 4.x 文档并不多。 Netty in Action确实说:

7.2.4 MessageToMessageDecoder – Decode POJO’s on the fly If you want to decode a message to another type of message MessageToMessageDecoder is the way to go. It allows an easy way to do so. The semantic is quite the same as for all the other decoders we explained before.

但为了简单起见，我暂时只是尝试使用字符串编码/解码器。我在服务器中正确使用它们吗？

另请参阅:

http://seeallhearall.blogspot.ca/2012/05/netty-tutorial-part-1-introduction-to.html

数据报服务器:

package net.bounceme.dur.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import java.util.logging.Logger;
import net.bounceme.dur.netty.datagram.DatagramServerInitializer;

public final class DatagramServer {

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

    public void start() throws InterruptedException {
        MyProps p = new MyProps();
        int port = p.getServerPort();
        pingPong(port);
    }


    private void pingPong(int port) throws InterruptedException {
        log.fine("which handler?");
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new DatagramServerInitializer());
            b.bind(port).sync().channel().closeFuture().await();
        } finally {
            group.shutdownGracefully();
        }
    }
}   

字符串服务器:

package net.bounceme.dur.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.util.logging.Logger;
import net.bounceme.dur.netty.string.StringServerInitializer;

public final class StringServer {

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

    public void start() throws InterruptedException {
        MyProps p = new MyProps();
        int port = p.getServerPort();
        pingPong(port);
    }

    private void pingPong(int port) throws InterruptedException {
        log.fine("which handler?");
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new StringServerInitializer());
            b.bind(port).sync().channel().closeFuture().await();
        } finally {
            group.shutdownGracefully();
        }
    }
}

数据报服务器处理程序:

package net.bounceme.dur.netty.datagram;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.CharsetUtil;
import java.util.Random;
import java.util.logging.Logger;

public class DatagramServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {

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

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

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        ctx.writeAndFlush(nextQuote());
    }

    // 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 String nextQuote() {
        int quoteId;
        synchronized (random) {
            quoteId = random.nextInt(quotes.length);
        }
        return quotes[quoteId];
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        log.info(packet.toString());
        if ("QOTM?".equals(packet.content().toString(CharsetUtil.UTF_8))) {
            ctx.write(new DatagramPacket(
                    Unpooled.copiedBuffer("QOTM: " + nextQuote(), CharsetUtil.UTF_8), packet.sender()));
        }
    }

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

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

数据报服务器初始化程序:

package net.bounceme.dur.netty.datagram;

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

public class DatagramServerInitializer extends ChannelInitializer<NioDatagramChannel> {

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

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

    @Override
    protected void initChannel(NioDatagramChannel c) throws Exception {
        log.info("..adding to pipeline..");
        ChannelPipeline pipeline = c.pipeline();
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new DatagramServerHandler());
    }
}

字符串服务器处理程序:

package net.bounceme.dur.netty.string;

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

public class StringServerHandler extends SimpleChannelInboundHandler<String> {

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

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

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        ctx.writeAndFlush(nextQuote());
    }

    // 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 String nextQuote() {
        int quoteId;
        synchronized (random) {
            quoteId = random.nextInt(quotes.length);
        }
        return quotes[quoteId];
    }

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

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

    @Override
    protected void channelRead0(ChannelHandlerContext chc, String msg) throws Exception {
        System.err.println(msg);
        if ("QOTM?".equals(msg)) {
            chc.writeAndFlush(nextQuote());
        } else {
            log.warning(msg);  //never executes
        }
    }
}

字符串服务器初始值设定项:

package net.bounceme.dur.netty.string;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.util.logging.Logger;

public class StringServerInitializer extends ChannelInitializer<ServerSocketChannel> {

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

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

    @Override
    protected void initChannel(ServerSocketChannel c) throws Exception {
        log.info("..adding to pipeline..");
        ChannelPipeline pipeline = c.pipeline();
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new StringEncoder());
        pipeline.addLast(new StringServerHandler());
    }
}

Answer 1

您似乎在 TCP 版本中正确使用了字符串编码器。

但是 TCP 和 UDP 并不相同。通过 TCP 连接，管道中的第一个传入处理程序将接收一个缓冲区，其中包含从套接字缓冲区读取的数据。套接字(以及 channel )绑定(bind)到远程对等点上的套接字，因此 channel 包含与远程对等点通信所需的所有信息。这包括远程地址和端口。 Netty 还可以创建附加到该 channel 的单独管道实例(取决于您如何设置它)。

虽然 API 相似，但 UDP 的行为却截然不同。它不是流，而是一组离散的数据报包，单个 channel 将接收来自许多不同发送者的数据报包。管道中的第一个传入处理程序将接收至少包含您的有效负载(您的消息)和发送者的 DatagramPacket。 DelimiterBasedFrameDecoder 无法在 DatagramPacket 上工作 - 它需要一个 ByteBuf，并且设计用于处理可以以多个 block 读取消息的流(请参阅用户指南中的 Dealing with a Stream-based Transport)

对于 UDP，您将其过于复杂化了。除非您将大消息分成多个 DatagramPacket，否则 DatagramPacket 应包含整个消息。直接从 DatagramPacket.content() 解码消息。请参阅QuoteOfTheMomentServerHandler有关从 DatagramPacket 读取字符串的示例。

另请参阅相同的示例，了解如何发送响应。使用消息创建新的 DatagramPacket，并将接收到的数据报的发送者用作新数据报的接收者。这很重要，因为 channel 没有绑定(bind)到特定的远程对等点。