Netty是建立在NIO基础之上，Netty在NIO之上又提供了更高层次的抽象。
在Netty里面，Accept连接可以使用单独的线程池去处理，读写操作又是另外的线程池来处理。
Accept连接和读写操作也可以使用同一个线程池来进行处理。而请求处理逻辑既可以使用单独的线程池进行处理，也可以跟放在读写线程一块处理。线程池中的每一个线程都是NIO线程。用户可以根据实际情况进行组装，构造出满足系统需求的并发模型。
Netty提供了内置的常用编解码器，包括行编解码器［一行一个请求］，前缀长度编解码器［前N个字节定义请求的字节长度］，可重放解码器［记录半包消息的状态］，HTTP编解码器，WebSocket消息编解码器等等
Netty提供了一些列生命周期回调接口，当一个完整的请求到达时，当一个连接关闭时，当一个连接建立时，用户都会收到回调事件，然后进行逻辑处理。
Netty可以同时管理多个端口，可以使用NIO客户端模型，这些对于RPC服务是很有必要的。
Netty除了可以处理TCP Socket之外，还可以处理UDP Socket。
在消息读写过程中，需要大量使用ByteBuffer，Netty对ByteBuffer在性能和使用的便捷性上都进行了优化和抽象。
服务端：
package ty.server;import ty.bootstrap.ServerBootstrap;import ty.channel.Channel;import ty.channel.ChannelOption;import ty.channel.EventLoopGroup;import ty.channel.nio.NioEventLoopGroup;import ty.channel.socket.nio.NioServerSocketChannel;public class NettyServer {/*** 端口*/private int port;public NettyServer(int port) {this.port = port;}public void run() {//EventLoopGroup是用来处理IO操作的多线程事件循环器//负责接收客户端连接线程EventLoopGroup bossGroup = new NioEventLoopGroup();//负责处理客户端i/o事件、task任务、监听任务组EventLoopGroup workerGroup = new NioEventLoopGroup();//启动 NIO 服务的辅助启动类ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup);//配置 Channelbootstrap.channel(NioServerSocketChannel.class);bootstrap.childHandler(new ServerIniterHandler());//BACKLOG用于构造服务端套接字ServerSocket对象，// 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度bootstrap.option(ChannelOption.SO_BACKLOG, 1024);//是否启用心跳保活机制bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);try {//绑定服务端口监听Channel channel = bootstrap.bind(port).sync().channel();System.out.println("server run in port " + port);//服务器关闭监听/*channel.closeFuture().sync()实际是如何工作:channel.closeFuture()不做任何操作，只是简单的返回channel对象中的closeFuture对象，对于每个Channel对象，都会有唯一的一个CloseFuture，用来表示关闭的Future，所有执行channel.closeFuture().sync()就是执行的CloseFuturn的sync方法，从上面的解释可以知道，这步是会将当前线程阻塞在CloseFuture上*/channel.closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();} finally {//关闭事件流组 bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {new NettyServer(8899).run();}}
服务端业务逻辑处理：
package ty.server;import ty.channel.Channel;import ty.channel.ChannelHandlerContext;import ty.channel.SimpleChannelInboundHandler;import ty.channel.group.ChannelGroup;import ty.channel.group.DefaultChannelGroup;import ty.util.concurrent.GlobalEventExecutor;public class ServerHandler extends SimpleChannelInboundHandler<String> {/*** 所有的活动用户*/public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 读取消息通道** @param context* @param s* @throws Exception*/@Overrideprotected void channelRead0(ChannelHandlerContext context, String s)throws Exception {Channel channel = context.channel();//当有用户发送消息的时候，对其他的用户发送消息for (Channel ch : group) {if (ch == channel) {ch.writeAndFlush("[you]: " + s + "\n");} else {ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");}}System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");}/*** 处理新加的消息通道** @param ctx* @throws Exception*/@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();for (Channel ch : group) {if (ch == channel) {ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");}}group.add(channel);}/*** 处理退出消息通道** @param ctx* @throws Exception*/@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();for (Channel ch : group) {if (ch == channel) {ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");}}group.remove(channel);}/*** 在建立连接时发送消息** @param ctx* @throws Exception*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();boolean active = channel.isActive();if (active) {System.out.println("[" + channel.remoteAddress() + "] is online");} else {System.out.println("[" + channel.remoteAddress() + "] is offline");}ctx.writeAndFlush("[server]: welcome");}/*** 退出时发送消息** @param ctx* @throws Exception*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();if (!channel.isActive()) {System.out.println("[" + channel.remoteAddress() + "] is offline");} else {System.out.println("[" + channel.remoteAddress() + "] is online");}}/*** 异常捕获** @param ctx* @param e* @throws Exception*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {Channel channel = ctx.channel();System.out.println("[" + channel.remoteAddress() + "] leave the room");ctx.close().sync();}}
服务端处理器注册：
package ty.server;import ty.channel.ChannelInitializer;import ty.channel.ChannelPipeline;import ty.channel.socket.SocketChannel;import ty.handler.codec.string.StringDecoder;import ty.handler.codec.string.StringEncoder;public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//管道注册handlerChannelPipeline pipeline = socketChannel.pipeline();//编码通道处理pipeline.addLast("decode", new StringDecoder());//转码通道处理pipeline.addLast("encode", new StringEncoder());//聊天服务通道处理pipeline.addLast("chat", new ServerHandler());}}
客户端：
package ty.client;import ty.bootstrap.Bootstrap;import ty.channel.Channel;import ty.channel.EventLoopGroup;import ty.channel.nio.NioEventLoopGroup;import ty.channel.socket.nio.NioSocketChannel;import mons.lang3.StringUtils;import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;public class NettyClient {private String ip;private int port;private boolean stop = false;public NettyClient(String ip, int port) {this.ip = ip;this.port = port;}public void run() throws IOException {//设置一个多线程循环器EventLoopGroup workerGroup = new NioEventLoopGroup();//启动附注类Bootstrap bootstrap = new Bootstrap();bootstrap.group(workerGroup);//指定所使用的NIO传输channelbootstrap.channel(NioSocketChannel.class);//指定客户端初始化处理bootstrap.handler(new ClientIniterHandler());try {//连接服务Channel channel = bootstrap.connect(ip, port).sync().channel();while (true) {//向服务端发送内容BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));String content = reader.readLine();if (StringUtils.isNotEmpty(content)) {if (StringUtils.equalsIgnoreCase(content, "q")) {System.exit(1);}channel.writeAndFlush(content);}}} catch (InterruptedException e) {e.printStackTrace();System.exit(1);} finally {workerGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new NettyClient("127.0.0.1", 8899).run();}}
客户端逻辑处理：
package ty.client;import ty.channel.ChannelHandlerContext;import ty.channel.SimpleChannelInboundHandler;public class ClientHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {//打印服务端的发送数据 System.out.println(s);}}
客户端处理器注册：
package ty.client;import ty.channel.ChannelInitializer;import ty.channel.ChannelPipeline;import ty.channel.socket.SocketChannel;import ty.handler.codec.http.HttpClientCodec;import ty.handler.codec.string.StringDecoder;import ty.handler.codec.string.StringEncoder;public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//注册管道ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast("decoder", new StringDecoder());pipeline.addLast("encoder", new StringEncoder());pipeline.addLast("http", new HttpClientCodec());pipeline.addLast("chat", new ClientHandler());}}
测试时先启动服务端，再启动客户端。。。
声明： 本文借鉴网络资源，来自：/kingsonfu/p/8635064.html
如果侵犯您的利益， 请联系本人删除
