Netty之实现自定义简单的编解码器一（MessageToByteEncoder<Integer>和ByteToMessageDecoder）

红太狼 2022-07-14 15:16 316阅读 0赞

1、关于自定义编码器的简介

在这里实现的编解码器很简单。**编码器的功能实现的是，int--->byets的编码；解码器实现的是，bytes--->int的解码。**

2、编码器的实现

import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    public class IntegerToByteEncoder extends MessageToByteEncoder<Integer> {
    
    	@Override
    	protected void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out)
    			throws Exception {
    		System.out.println("IntegerToByteEncoder encode msg is " + msg);
    		out.writeInt(msg);
    	}
    }

3、解码器的实现

import java.util.List;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    public class ByteToIntegerDecoder extends ByteToMessageDecoder {
    	@Override
    	protected void decode(ChannelHandlerContext ctx, ByteBuf in,
    			List<Object> out) throws Exception {
    		// Check if there are at least 4 bytes readable
    		if (in.readableBytes() >= 4) {
    			int n = in.readInt();
    			System.out.println("ByteToIntegerDecoder decode msg is " + n);
    			// Read integer from inbound ByteBuf
    			// add to the List of decodec messages
    			out.add(n);
    		}
    	}
    }

4、使用编解码器的步骤

4.1  加入编解码器

![Center][]

4.2  发送信息，编码方式的改变

因为加入了int--->bytes的编码器。所以，再发送数据的时候，可以直接发送Integer类型的数据。不需要在手动的，将Integer利用Unpooled工具类转换为ByteBuf类型在发送。

4.2.1  使用编码器之前的代码方式

![Center 1][]

4.2.1  使用编码器之后的代码方式

![Center 2][]

4.3  接受信息，编码方式的改变

因为加入了bytes--->int的解码器。所以，可以将消息转换为Integer类型处理。

4.3.1  没加入解码器前的代码方式

![20161214202810985][]

4.3.2  加入了解码器后的代码方式

![20161214202830470][]

5、服务端的实现

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;
    
    public class Server {
    	public void bind(int port) throws Exception {
    		// 服务器线程组 用于网络事件的处理 一个用于服务器接收客户端的连接
    		// 另一个线程组用于处理SocketChannel的网络读写
    		EventLoopGroup bossGroup = new NioEventLoopGroup();
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    		try {
    			// NIO服务器端的辅助启动类 降低服务器开发难度
    			ServerBootstrap serverBootstrap = new ServerBootstrap();
    			serverBootstrap.group(bossGroup, workerGroup)
    					.channel(NioServerSocketChannel.class)// 类似NIO中serverSocketChannel
    					.option(ChannelOption.SO_BACKLOG, 1024)// 配置TCP参数
    					.option(ChannelOption.SO_BACKLOG, 1024) // 设置tcp缓冲区
    					.option(ChannelOption.SO_SNDBUF, 32 * 1024) // 设置发送缓冲大小
    					.option(ChannelOption.SO_RCVBUF, 32 * 1024) // 这是接收缓冲大小
    					.option(ChannelOption.SO_KEEPALIVE, true) // 保持连接
    					.childHandler(new ChildChannelHandler());// 最后绑定I/O事件的处理类
    																// 处理网络IO事件
    
    			// 服务器启动后 绑定监听端口 同步等待成功 主要用于异步操作的通知回调 回调处理用的ChildChannelHandler
    			ChannelFuture f = serverBootstrap.bind(port).sync();
    			System.out.println("Server启动");
    			// 等待服务端监听端口关闭
    			f.channel().closeFuture().sync();
    
    		} finally {
    			// 优雅退出 释放线程池资源
    			bossGroup.shutdownGracefully();
    			workerGroup.shutdownGracefully();
    			System.out.println("服务器优雅的释放了线程资源...");
    		}
    
    	}
    
    	/**
    	 * 网络事件处理器
    	 */
    	private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
    		@Override
    		protected void initChannel(SocketChannel ch) throws Exception {
    			// 增加自定义的编码器和解码器
    			ch.pipeline().addLast(new IntegerToByteEncoder());
    			ch.pipeline().addLast(new ByteToIntegerDecoder());
    			// 服务端的处理器
    			ch.pipeline().addLast(new ServerHandler());
    		}
    	}
    
    	public static void main(String[] args) throws Exception {
    		int port = 9998;
    		new Server().bind(port);
    	}
    }

6、 服务端Handler的实现

import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class ServerHandler extends ChannelHandlerAdapter {
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg)
    			throws Exception {
    		// 接受客户端的数据
    		Integer body = (Integer) msg;
    		System.out.println("Client :" + body.toString());
    		// 服务端，回写数据给客户端
    		// 直接回写整形的数据
    		ctx.writeAndFlush(33);
    
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    			throws Exception {
    		ctx.close();
    	}
    
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		ctx.flush();
    	}
    }

7、客户端的实现

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;
    
    public class Client {
    	/**
    	 * 连接服务器
    	 * 
    	 * @param port
    	 * @param host
    	 * @throws Exception
    	 */
    	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 ClientChannelHandler());//
    			// 异步链接服务器 同步等待链接成功
    			ChannelFuture f = b.connect(host, port).sync();
    
    			// 发送消息
    			Thread.sleep(1000);
    			f.channel().writeAndFlush(777);
    			f.channel().writeAndFlush(666);
    			Thread.sleep(2000);
    			f.channel().writeAndFlush(888);
    
    			// 等待链接关闭
    			f.channel().closeFuture().sync();
    
    		} finally {
    			group.shutdownGracefully();
    			System.out.println("客户端优雅的释放了线程资源...");
    		}
    
    	}
    
    	/**
    	 * 网络事件处理器
    	 */
    	private class ClientChannelHandler extends
    			ChannelInitializer<SocketChannel> {
    		@Override
    		protected void initChannel(SocketChannel ch) throws Exception {
    			// 增加自定义的编码器和解码器
    			ch.pipeline().addLast(new IntegerToByteEncoder());
    			ch.pipeline().addLast(new ByteToIntegerDecoder());
    			// 客户端的处理器
    			ch.pipeline().addLast(new ClientHandler());
    		}
    
    	}
    
    	public static void main(String[] args) throws Exception {
    		new Client().connect(9998, "127.0.0.1");
    
    	}
    }

8、客户端Handler的实现

import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.util.ReferenceCountUtil;
    
    public class ClientHandler extends ChannelHandlerAdapter {
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg)
    			throws Exception {
    		try {
    			Integer body = (Integer) msg;
    			System.out.println("Client :" + body.toString());
    
    			// 只是读数据，没有写数据的话
    			// 需要自己手动的释放的消息
    
    		} finally {
    			ReferenceCountUtil.release(msg);
    		}
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    			throws Exception {
    		ctx.close();
    	}
    
    }

9、程序的运行结果

Server端

Server启动
    ByteToIntegerDecoder decode msg is 777
    ByteToIntegerDecoder decode msg is 666
    Client :777
    IntegerToByteEncoder encode msg is 33
    Client :666
    IntegerToByteEncoder encode msg is 33
    ByteToIntegerDecoder decode msg is 888
    Client :888
    IntegerToByteEncoder encode msg is 33

Client端

IntegerToByteEncoder encode msg is 777
    IntegerToByteEncoder encode msg is 666
    ByteToIntegerDecoder decode msg is 33
    ByteToIntegerDecoder decode msg is 33
    Client :33
    Client :33
    IntegerToByteEncoder encode msg is 888
    ByteToIntegerDecoder decode msg is 33
    Client :33

10、[源码下载][Link 1]

[Center]: /images/20220714/f4d33ff8e56042d29a7390a1768025ec.png
[Center 1]: /images/20220714/247c51b3ce1e4fe1bc9b0fc412447748.png
[Center 2]: /images/20220714/46a529cb56ab4c53a2b872a8b5c2c545.png
[20161214202810985]: /images/20220714/8d6f93b23b4a4d459fafe9029cc3a106.png
[20161214202830470]: /images/20220714/6958d25cfa4f4b5a8596758ef160fbae.png
[Link 1]: http://pan.baidu.com/s/1pLIZLAb