netty入门篇(1)

时间：2022-05-27 03:20:16

上一篇 nio简介 　　下一篇 netty中级篇(2)

一、为什么选择Netty

Netty是最流行的框架之一、健壮性、功能、性能、可定制性和可扩展性在同类框架中首屈一指，因此被大规模使用，例如ROCKETMQ的NameSRV，例如Hadoop的Avro，例如Dubbo中的RPC通信等等。。

为什么选择Netty?

  • API简单；
  • 功能强大，预置了选多的编码功能，支持多种主流协议；
  • 定制能力强，通过ChannelHandler对通信框架进行灵活的扩展；
  • 性能强；
  • 成熟，修改已发现的JDK nio BUG
  • 社区活跃
  • 经过大规模的商业应用考验，质量得到验证。

二、使用Netty开发TimeServer

环境准备: pom.xml

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>demo</groupId>

    <artifactId>netty</artifactId>

    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <!-- https://mvnrepository.com/artifact/io.netty/netty-all -->

        <dependency>

            <groupId>io.netty</groupId>

            <artifactId>netty-all</artifactId>

            <version>4.1.5.Final</version>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-compiler-plugin</artifactId>

                <version>3.1</version>

                <configuration>

                    <source>1.8</source>

                    <target>1.8</target>

                </configuration>

            </plugin>

        </plugins>

    </build>

</project>

1. Netty TimeServer

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

public class TimeServer {

    public void bind(int port) throws Exception {

        // 配置服务端的NIO线程组

        EventLoopGroup bossGroup = new NioEventLoopGroup();

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {

            //Netty启动Nio服务端的辅助类

            ServerBootstrap b = new ServerBootstrap();

            b.group(bossGroup, workerGroup)

                    .channel(NioServerSocketChannel.class)

                    .option(ChannelOption.SO_BACKLOG, 1024) //设置服务端tcp参数

                    .childHandler(new ChildChannelHandler());

            // 绑定端口，同步等待成功

            ChannelFuture f = b.bind(port).sync();

            // 进行阻塞，等待服务端监听端口关闭

            f.channel().closeFuture().sync();

        } finally {

            // 优雅退出，释放线程池资源

            System.out.println("服务器关闭...");

            bossGroup.shutdownGracefully();

            workerGroup.shutdownGracefully();

        }

    }

    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {

        @Override

        protected void initChannel(SocketChannel arg0) throws Exception {

            arg0.pipeline().addLast(new TimeServerHandler());

        }

    }

    /**

     * @param args

     * @throws Exception

     */

    public static void main(String[] args) throws Exception {

        int port = 8080;

        if (args != null && args.length > 0) {

            try {

                port = Integer.valueOf(args[0]);

            } catch (NumberFormatException e) {

                // 采用默认值

            }

        }

        new TimeServer().bind(port);

    }

}
  • EventLoopGroup Reactor线程组 2个
  • ServerBootstrap：Server端辅助工具
  • 设置channel: NioServerSocketChannel
  • option: 服务端tcp option设置，这里以backlog 1024为例..
  • 增加childHandler
  • f.channel().closeFuture().sync()表示进行阻塞，等待服务器端链路关闭之后main函数才退出

2. TimeServerHandler

 import io.netty.buffer.ByteBuf;

 import io.netty.buffer.Unpooled;

 import io.netty.channel.ChannelHandlerAdapter;

 import io.netty.channel.ChannelHandlerContext;

 import io.netty.channel.ChannelInboundHandlerAdapter;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeServerHandler extends ChannelInboundHandlerAdapter {

     @Override

     public void channelRead(ChannelHandlerContext ctx, Object msg)

             throws Exception {

         ByteBuf buf = (ByteBuf) msg;

         byte[] req = new byte[buf.readableBytes()];

         buf.readBytes(req);

         String body = new String(req, "UTF-8");

         System.out.println("The time server receive order : " + body);

         String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(

                 System.currentTimeMillis()).toString() : "BAD ORDER";

         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());

         ctx.write(resp);

     }

     @Override

     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

         ctx.flush();

     }

     @Override

     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

         ctx.close();

     }

 }

(1) 18行做类型转换，将msg转换为Netty的ByteBuf对象，这个对象比ByteBuffer更加强大和灵活。

(2) 19行到20行通过ByteBuf的readableBytes获取缓冲区可读的字节数，根据可读的字节数创建byte数组。将缓冲区的内容读取到byte数组中。

(3) 31行发现了flush方法，其作用是将消息发送队列中的消息写入到SocketChannel中发送给对方。从性能上考虑，为了防止频繁唤醒Selector进行消息发送，Netty的write方法不直接写入到SocketChannel中，调用write方法只会写入到缓冲数组中，调用flush方法，才会写入到SocketChannel中。

(4) 36行的close()是在发生异常后释放资源

总结: 就是比NIO舒服太多了.

3. Time Client

 import io.netty.bootstrap.Bootstrap;

 import io.netty.channel.ChannelFuture;

 import io.netty.channel.ChannelInitializer;

 import io.netty.channel.ChannelOption;

 import io.netty.channel.EventLoopGroup;

 import io.netty.channel.nio.NioEventLoopGroup;

 import io.netty.channel.socket.SocketChannel;

 import io.netty.channel.socket.nio.NioSocketChannel;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeClient {

     public void connect(int port, String host) throws Exception {

         // 配置客户端NIO线程组

         EventLoopGroup group = new NioEventLoopGroup();

         try {

             Bootstrap b = new Bootstrap();

             b.group(group).channel(NioSocketChannel.class)

                     .option(ChannelOption.TCP_NODELAY, true)

                     .handler(new ChannelInitializer<SocketChannel>() {

                         @Override

                         public void initChannel(SocketChannel ch)

                                 throws Exception {

                             ch.pipeline().addLast(new TimeClientHandler());

                         }

                     });

             // 发起异步连接操作

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

             // 当代客户端链路关闭

             f.channel().closeFuture().sync();

         } finally {

             // 优雅退出，释放NIO线程组

             group.shutdownGracefully();

         }

     }

     /**

      * @param args

      * @throws Exception

      */

     public static void main(String[] args) throws Exception {

         int port = 8080;

         if (args != null && args.length > 0) {

             try {

                 port = Integer.valueOf(args[0]);

             } catch (NumberFormatException e) {

                 // 采用默认值

             }

         }

         new TimeClient().connect(port, "127.0.0.1");

     }

 }

(1) 19行创建客户端处理I/O读写的NioEventLoopGroup线程组，然后继续创建辅助类Bootstrap，并且对其配置，此处配置为 NioSocketChannel，然后为其添加Handler。

(2) 这里Handler直接使用匿名内部类

(3) 33行的connect发送异步连接请求，然后阻塞直到关闭。

4. TimeClientHandler

 import io.netty.buffer.ByteBuf;

 import io.netty.buffer.Unpooled;

 import io.netty.channel.ChannelHandlerAdapter;

 import io.netty.channel.ChannelHandlerContext;

 import io.netty.channel.ChannelInboundHandlerAdapter;

 import java.util.logging.Logger;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeClientHandler extends ChannelInboundHandlerAdapter {

     private static final Logger logger = Logger

             .getLogger(TimeClientHandler.class.getName());

     private final ByteBuf firstMessage;

     /**

      * Creates a client-side handler.

      */

     public TimeClientHandler() {

         byte[] req = "QUERY TIME ORDER".getBytes();

         firstMessage = Unpooled.buffer(req.length);

         firstMessage.writeBytes(req);

     }

     @Override

     public void channelActive(ChannelHandlerContext ctx) {

         ctx.writeAndFlush(firstMessage);

     }

     @Override

     public void channelRead(ChannelHandlerContext ctx, Object msg)

             throws Exception {

         ByteBuf buf = (ByteBuf) msg;

         byte[] req = new byte[buf.readableBytes()];

         buf.readBytes(req);

         String body = new String(req, "UTF-8");

         System.out.println("Now is : " + body);

     }

     @Override

     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

         // 释放资源

         logger.warning("Unexpected exception from downstream : "

                 + cause.getMessage());

         ctx.close();

     }

 }

这里重点关注3个方法: channelActive channelRead和exceptionCaught。

(1) 当客户端和服务器端成功创建链路，调用channelActive方法，发送查询时间的指令给服务端，调用writeAndFlush方法发送数据。

(2) 39行开始调用channelRead，读取数据，49行处理异常时释放资源即可。

三、TCP 粘包/拆包问题的解决之道

1、TCP得粘包和拆包问题

  • TCP是一个流协议
  • TCP底层不了解业务数据含义，即不知道多少个字节算是业务上的一整体数据
  • 因此业务上认为，一个完整的包会被TCP拆分为多个包进行发送，也有可能将多个小的包封装成一个大包进行发送、

用下图进行描述，假设client发送了2个包，D1和D2，服务器端读到的数据是不确定的

netty入门篇(1)

存在4种可能：

server 分2次，分别读到D1，D2，完美巧合，没有粘包和拆包

server一次读到D1，D2，D1和D2粘在一起，称为粘包

server分2次，第一次读到D1和D2的部分内容，第二次读到了D2的剩余内容，拆包

server分2次，第一次读到D1的部分内容D1_1，第二次读到D1剩下的内容D1_2和完整的D2。

如果此时服务器端TCP接收的滑窗非常的小、而且数据包D1和D2都比较大，很有可能发生第5种可能性，服务器端多次才能将D1和D2接收完全，期间发生多次拆包...即上4种情况的多次组合...

下面我们来分析一下原因:

3个原因:

(1) 应用程序write写入的字节大于套接口(scoket)发送缓冲的大小;

(2) 进行MSS大小的TCP分段;

(3) 以太网帧的payload大于MTU进行IP分片

netty入门篇(1)

总结就是：不可避免...

解决思路：

(1) 定长数据，例如每个报文200bytes，不够空格补充...

(2) 在包围增加回车换行符或者其他的特殊字符进行分割，例如FTP协议

(3) 将消息分为消息头和消息体，消息头中包含消息总长度(或者消息体长度)content-length，通常的设计思路为消息头的第一个字段用int32来表示消息的总长度；

(4) 更复杂的应用层协议

2. 下面我们来模拟未考虑TCP粘包问题导致功能异常

修改上面的代码：

修改TimeServerHandler

 import io.netty.buffer.ByteBuf;

 import io.netty.buffer.Unpooled;

 import io.netty.channel.ChannelHandlerContext;

 import io.netty.channel.ChannelInboundHandlerAdapter;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeServerHandler extends ChannelInboundHandlerAdapter {

     private int counter;

     @Override

     public void channelRead(ChannelHandlerContext ctx, Object msg)

             throws Exception {

         ByteBuf buf = (ByteBuf) msg;

         byte[] req = new byte[buf.readableBytes()];

         buf.readBytes(req);

         String body = new String(req, "UTF-8").substring(0, req.length

                 - System.getProperty("line.separator").length());

         System.out.println("The time server receive order : " + body

                 + " ; the counter is : " + ++counter);

         String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(

                 System.currentTimeMillis()).toString() : "BAD ORDER";

         currentTime = currentTime + System.getProperty("line.separator");

         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());

         ctx.writeAndFlush(resp);

     }

     @Override

     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

         ctx.close();

     }

 }

主要是增加了一个counter进行计数..

修改TimeClientHandler

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.logging.Logger;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private static final Logger logger = Logger

            .getLogger(TimeClientHandler.class.getName());

    private final ByteBuf firstMessage;

    /**

     * Creates a client-side handler.

     */

    public TimeClientHandler() {

        byte[] req = "QUERY TIME ORDER".getBytes();

        firstMessage = Unpooled.buffer(req.length);

        firstMessage.writeBytes(req);

    }

    @Override

    public void channelActive(ChannelHandlerContext ctx) {

        ctx.writeAndFlush(firstMessage);

    }

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg)

            throws Exception {

        ByteBuf buf = (ByteBuf) msg;

        byte[] req = new byte[buf.readableBytes()];

        buf.readBytes(req);

        String body = new String(req, "UTF-8");

        System.out.println("Now is : " + body);

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        // 释放资源

        logger.warning("Unexpected exception from downstream : "

                + cause.getMessage());

        ctx.close();

    }

}

主要是进行100次连续的发送数据...

由于tcp粘包拆包有一定的随机性，所以每次的结果可能不同，其中一次结果大致上是：

Server端打印：

QUERY TIME ORDER

....

the counter is : 2

Client端打印:

Now is : Thu Dec 15 15:11:22 CST 2016

BAD ORDER

BAD ORDER

 ; the counter is : 1

结果表明：client发送了100条消息，但是server是按照2次接收，只返回2条应答，但是client上的counter为1表明只client也接收了一次，说明这2条也进行了粘包。

3. 解决TCP粘包的TimeServer

TimeServer

 import io.netty.bootstrap.ServerBootstrap;

 import io.netty.channel.ChannelFuture;

 import io.netty.channel.ChannelInitializer;

 import io.netty.channel.ChannelOption;

 import io.netty.channel.EventLoopGroup;

 import io.netty.channel.nio.NioEventLoopGroup;

 import io.netty.channel.socket.SocketChannel;

 import io.netty.channel.socket.nio.NioServerSocketChannel;

 import io.netty.handler.codec.LineBasedFrameDecoder;

 import io.netty.handler.codec.string.StringDecoder;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeServer {

     public void bind(int port) throws Exception {

         // 配置服务端的NIO线程组

         EventLoopGroup bossGroup = new NioEventLoopGroup();

         EventLoopGroup workerGroup = new NioEventLoopGroup();

         try {

             ServerBootstrap b = new ServerBootstrap();

             b.group(bossGroup, workerGroup)

                     .channel(NioServerSocketChannel.class)

                     .option(ChannelOption.SO_BACKLOG, 1024)

                     .childHandler(new ChildChannelHandler());

             // 绑定端口，同步等待成功

             ChannelFuture f = b.bind(port).sync();

             // 等待服务端监听端口关闭

             f.channel().closeFuture().sync();

         } finally {

             // 优雅退出，释放线程池资源

             bossGroup.shutdownGracefully();

             workerGroup.shutdownGracefully();

         }

     }

     private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {

         @Override

         protected void initChannel(SocketChannel arg0) throws Exception {

             arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));

             arg0.pipeline().addLast(new StringDecoder());

             arg0.pipeline().addLast(new TimeServerHandler());

         }

     }

     /**

      * @param args

      * @throws Exception

      */

     public static void main(String[] args) throws Exception {

         int port = 8080;

         if (args != null && args.length > 0) {

             try {

                 port = Integer.valueOf(args[0]);

             } catch (NumberFormatException e) {

                 // 采用默认值

             }

         }

         new TimeServer().bind(port);

     }

 }

重点看44行，增加了2个解码器: LineBasedFrameDecoder和StringDecoder。

继续看TimeServerHandler

 import io.netty.buffer.ByteBuf;

 import io.netty.buffer.Unpooled;

 import io.netty.channel.ChannelHandlerContext;

 import io.netty.channel.ChannelInboundHandlerAdapter;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeServerHandler extends ChannelInboundHandlerAdapter {

     private int counter;

     @Override

     public void channelRead(ChannelHandlerContext ctx, Object msg)

             throws Exception {

         String body = (String) msg;

         System.out.println("The time server receive order : " + body

                 + " ; the counter is : " + ++counter);

         String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(

                 System.currentTimeMillis()).toString() : "BAD ORDER";

         currentTime = currentTime + System.getProperty("line.separator");

         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());

         ctx.writeAndFlush(resp);

     }

     @Override

     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

         ctx.close();

     }

 }

看18行，直接获取之后不是ByteBuf，而直接是一个String对象，代码非常简洁。

TimeClient

 import io.netty.bootstrap.Bootstrap;

 import io.netty.channel.ChannelFuture;

 import io.netty.channel.ChannelInitializer;

 import io.netty.channel.ChannelOption;

 import io.netty.channel.EventLoopGroup;

 import io.netty.channel.nio.NioEventLoopGroup;

 import io.netty.channel.socket.SocketChannel;

 import io.netty.channel.socket.nio.NioSocketChannel;

 import io.netty.handler.codec.LineBasedFrameDecoder;

 import io.netty.handler.codec.string.StringDecoder;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class TimeClient {

     public void connect(int port, String host) throws Exception {

         // 配置客户端NIO线程组

         EventLoopGroup group = new NioEventLoopGroup();

         try {

             Bootstrap b = new Bootstrap();

             b.group(group).channel(NioSocketChannel.class)

                     .option(ChannelOption.TCP_NODELAY, true)

                     .handler(new ChannelInitializer<SocketChannel>() {

                         @Override

                         public void initChannel(SocketChannel ch)

                                 throws Exception {

                             ch.pipeline().addLast(

                                     new LineBasedFrameDecoder(1024));

                             ch.pipeline().addLast(new StringDecoder());

                             ch.pipeline().addLast(new TimeClientHandler());

                         }

                     });

             // 发起异步连接操作

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

             // 当代客户端链路关闭

             f.channel().closeFuture().sync();

         } finally {

             // 优雅退出，释放NIO线程组

             group.shutdownGracefully();

         }

     }

     /**

      * @param args

      * @throws Exception

      */

     public static void main(String[] args) throws Exception {

         int port = 8080;

         if (args != null && args.length > 0) {

             try {

                 port = Integer.valueOf(args[0]);

             } catch (NumberFormatException e) {

                 // 采用默认值

             }

         }

         new TimeClient().connect(port, "127.0.0.1");

     }

 }

类似TimeServer增加了2个解码器

再看TimeClientHandler

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.logging.Logger;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private static final Logger logger = Logger

            .getLogger(TimeClientHandler.class.getName());

    private int counter;

    private byte[] req;

    /**

     * Creates a client-side handler.

     */

    public TimeClientHandler() {

        req = ("QUERY TIME ORDER" + System.getProperty("line.separator"))

                .getBytes();

    }

    @Override

    public void channelActive(ChannelHandlerContext ctx) {

        ByteBuf message = null;

        for (int i = 0; i < 100; i++) {

            message = Unpooled.buffer(req.length);

            message.writeBytes(req);

            ctx.writeAndFlush(message);

        }

    }

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg)

            throws Exception {

        String body = (String) msg;

        System.out.println("Now is : " + body + " ; the counter is : "

                + ++counter);

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        // 释放资源

        logger.warning("Unexpected exception from downstream : "

                + cause.getMessage());

        ctx.close();

    }

}

直接运行发现完全符合我们需求

4. 分析LineBaseFrameDecoder和StringDecoder

LineBasedFrameDecoder的工作原理非常简单:

(1) 遍历ByteBuf中的可读字节，判断看是否有"\n"或者"\r\n"，如果有，就以此为结束位置，从可读索引到结束位置区间的字节就组成了一行

(2) 是一个以换行符为结束标志的解码器，支持携带结束符或者不携带结束符2种方式，同时支持配置单行的最大长度。

(3) 超过单行最大长度直接抛异常

StringDecoder的非常简单：

(1) 将接收的对象转换为字符串

(2) 继续调用后面的Handler

因此:

LineBasedFrameDecoder和StringDecoder组合在一起就是行切换文件解码器。

四、分割符解码器的应用

使用DelimiterBasedFrameDecoder即可...

1. EohoServer

 import io.netty.bootstrap.ServerBootstrap;

 import io.netty.buffer.ByteBuf;

 import io.netty.buffer.Unpooled;

 import io.netty.channel.ChannelFuture;

 import io.netty.channel.ChannelInitializer;

 import io.netty.channel.ChannelOption;

 import io.netty.channel.EventLoopGroup;

 import io.netty.channel.nio.NioEventLoopGroup;

 import io.netty.channel.socket.SocketChannel;

 import io.netty.channel.socket.nio.NioServerSocketChannel;

 import io.netty.handler.codec.DelimiterBasedFrameDecoder;

 import io.netty.handler.codec.string.StringDecoder;

 import io.netty.handler.logging.LogLevel;

 import io.netty.handler.logging.LoggingHandler;

 /**

  * @author lilinfeng

  * @version 1.0

  * @date 2014年2月14日

  */

 public class EchoServer {

     public void bind(int port) throws Exception {

         // 配置服务端的NIO线程组

         EventLoopGroup bossGroup = new NioEventLoopGroup();

         EventLoopGroup workerGroup = new NioEventLoopGroup();

         try {

             ServerBootstrap b = new ServerBootstrap();

             b.group(bossGroup, workerGroup)

                     .channel(NioServerSocketChannel.class)

                     .option(ChannelOption.SO_BACKLOG, 100)

                     .handler(new LoggingHandler(LogLevel.INFO))

                     .childHandler(new ChannelInitializer<SocketChannel>() {

                         @Override

                         public void initChannel(SocketChannel ch)

                                 throws Exception {

                             ByteBuf delimiter = Unpooled.copiedBuffer("$_"

                                     .getBytes());

                             ch.pipeline().addLast(

                                     new DelimiterBasedFrameDecoder(1024,

                                             delimiter));

                             ch.pipeline().addLast(new StringDecoder());

                             ch.pipeline().addLast(new EchoServerHandler());

                         }

                     });

             // 绑定端口，同步等待成功

             ChannelFuture f = b.bind(port).sync();

             // 等待服务端监听端口关闭

             f.channel().closeFuture().sync();

         } finally {

             // 优雅退出，释放线程池资源

             bossGroup.shutdownGracefully();

             workerGroup.shutdownGracefully();

         }

     }

     public static void main(String[] args) throws Exception {

         int port = 8080;

         if (args != null && args.length > 0) {

             try {

                 port = Integer.valueOf(args[0]);

             } catch (NumberFormatException e) {

                 // 采用默认值

             }

         }

         new EchoServer().bind(port);

     }

 }

(1) 重点在于38行的DelimiterBasedFrameDecoder， 与上面的换行分割符类似，只是可以自定义特殊符号

(2) DelimiterBasedFrameDecoder有2个参数，一个为单行最大长度，一个为自定义符号对象

(3) 如果到达长度仍然没有查询到，就抛出TooLongFrameException异常

2. EchoServerHandler

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelHandler.Sharable;

import io.netty.channel.ChannelHandlerAdapter;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

@Sharable

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    int counter = 0;

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg)

            throws Exception {

        String body = (String) msg;

        System.out.println("This is " + ++counter + " times receive client : ["

                + body + "]");

        body += "$_";

        ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());

        ctx.writeAndFlush(echo);

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        cause.printStackTrace();

        ctx.close();// 发生异常，关闭链路

    }

}

非常简单直接打印再write即可... 由此也可以看出netty框架比较干净的分离出来了业务逻辑代码。

3. Client端和ClientHandler基本类似

import io.netty.bootstrap.Bootstrap;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioSocketChannel;

import io.netty.handler.codec.DelimiterBasedFrameDecoder;

import io.netty.handler.codec.string.StringDecoder;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

public class EchoClient {

    public void connect(int port, String host) throws Exception {

        // 配置客户端NIO线程组

        EventLoopGroup group = new NioEventLoopGroup();

        try {

            Bootstrap b = new Bootstrap();

            b.group(group).channel(NioSocketChannel.class)

                    .option(ChannelOption.TCP_NODELAY, true)

                    .handler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        public void initChannel(SocketChannel ch)

                                throws Exception {

                            ByteBuf delimiter = Unpooled.copiedBuffer("$_"

                                    .getBytes());

                            ch.pipeline().addLast(

                                    new DelimiterBasedFrameDecoder(1024,

                                            delimiter));

                            ch.pipeline().addLast(new StringDecoder());

                            ch.pipeline().addLast(new EchoClientHandler());

                        }

                    });

            // 发起异步连接操作

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

            // 当代客户端链路关闭

            f.channel().closeFuture().sync();

        } finally {

            // 优雅退出，释放NIO线程组

            group.shutdownGracefully();

        }

    }

    /**

     * @param args

     * @throws Exception

     */

    public static void main(String[] args) throws Exception {

        int port = 8080;

        if (args != null && args.length > 0) {

            try {

                port = Integer.valueOf(args[0]);

            } catch (NumberFormatException e) {

                // 采用默认值

            }

        }

        new EchoClient().connect(port, "127.0.0.1");

    }

}
import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelHandlerAdapter;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

public class EchoClientHandler extends ChannelInboundHandlerAdapter {

    private int counter;

    static final String ECHO_REQ = "Hi, Lilinfeng. Welcome to Netty.$_";

    /**

     * Creates a client-side handler.

     */

    public EchoClientHandler() {

    }

    @Override

    public void channelActive(ChannelHandlerContext ctx) {

        // ByteBuf buf = UnpooledByteBufAllocator.DEFAULT.buffer(ECHO_REQ

        // .getBytes().length);

        // buf.writeBytes(ECHO_REQ.getBytes());

        for (int i = 0; i < 10; i++) {

            ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));

        }

    }

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg)

            throws Exception {

        System.out.println("This is " + ++counter + " times receive server : ["

                + msg + "]");

    }

    @Override

    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

        ctx.flush();

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        cause.printStackTrace();

        ctx.close();

    }

}

运行代码，符合预期..

五、定长解码器

1. 开发服务端

非常简单，直接上代码:

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

import io.netty.handler.codec.FixedLengthFrameDecoder;

import io.netty.handler.codec.string.StringDecoder;

import io.netty.handler.logging.LogLevel;

import io.netty.handler.logging.LoggingHandler;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

public class EchoServer {

    public void bind(int port) throws Exception {

        // 配置服务端的NIO线程组

        EventLoopGroup bossGroup = new NioEventLoopGroup();

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap b = new ServerBootstrap();

            b.group(bossGroup, workerGroup)

                    .channel(NioServerSocketChannel.class)

                    .option(ChannelOption.SO_BACKLOG, 100)

                    .handler(new LoggingHandler(LogLevel.INFO))

                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        public void initChannel(SocketChannel ch)

                                throws Exception {

                            ch.pipeline().addLast(

                                    new FixedLengthFrameDecoder(20));

                            ch.pipeline().addLast(new StringDecoder());

                            ch.pipeline().addLast(new EchoServerHandler());

                        }

                    });

            // 绑定端口，同步等待成功

            ChannelFuture f = b.bind(port).sync();

            // 等待服务端监听端口关闭

            f.channel().closeFuture().sync();

        } finally {

            // 优雅退出，释放线程池资源

            bossGroup.shutdownGracefully();

            workerGroup.shutdownGracefully();

        }

    }

    public static void main(String[] args) throws Exception {

        int port = 8080;

        if (args != null && args.length > 0) {

            try {

                port = Integer.valueOf(args[0]);

            } catch (NumberFormatException e) {

                // 采用默认值

            }

        }

        new EchoServer().bind(port);

    }

}
import io.netty.channel.ChannelHandler.Sharable;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

/**

 * @author lilinfeng

 * @version 1.0

 * @date 2014年2月14日

 */

@Sharable

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg)

            throws Exception {

        System.out.println("Receive client : [" + msg + "]");

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        cause.printStackTrace();

        ctx.close();// 发生异常，关闭链路

    }

}

2. 使用telnet进行访问

(1) 我使用的是Xshell，直接命令

(2) telnet 127.0.0.1 8080

(3) 再随便输入字符，发现每20个字符，服务端显示一次，符合预期

相关文章