Netty框架实现tcp通信

柔情只为你懂 2022-05-11 04:50 333阅读 0赞

启动服务端

*  创建两个NIO线程组，一个专门用于接收来自客户端的连接，另一个则用于处理已经被接收的连接。
    *  创建一个ServerBootstrap对象，配置Netty的一系列参数，例如接受传出数据的缓存大小等。
    *  创建一个用于实际处理数据的类ChannelInitializer，进行初始化的准备工作，比如设置接受传出数据的字符集、格式以及实际处理数据的接口。
    *  绑定端口，执行同步阻塞方法等待服务器端启动即可。

package com.zc.netty;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    public class NettyServer {
        private static final int port = 8098;
        public void run() throws Exception{
            //NioEventLoopGroup是用来处理IO操作的多线程事件循环器
            EventLoopGroup bossGroup  = new NioEventLoopGroup();  // 用来接收进来的连接
            EventLoopGroup workerGroup  = new NioEventLoopGroup();// 用来处理已经被接收的连接
            try{
                ServerBootstrap server =new ServerBootstrap();//是一个启动NIO服务的辅助启动类
                server.group(bossGroup,workerGroup )
                        .channel(NioServerSocketChannel.class)  // 这里告诉Channel如何接收新的连接
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                // 自定义处理类
                                ch.pipeline().addLast(new NettyServerHandler());
                            }
                        });
                server.option(ChannelOption.SO_BACKLOG,128);
                server.childOption(ChannelOption.SO_KEEPALIVE, true);
                ChannelFuture f = server.bind(port).sync();// 绑定端口，开始接收进来的连接
                System.out.println("服务端启动成功...");
                // 监听服务器关闭监听
                f.channel().closeFuture().sync();
            }finally {
                bossGroup.shutdownGracefully(); 关闭EventLoopGroup，释放掉所有资源包括创建的线程
                workerGroup.shutdownGracefully();
            }
    
            // 服务器绑定端口监听
    
    
        }
        public static void main(String[] args) throws Exception {
            new NettyServer().run();
        }
    }

import io.netty.channel.ChannelInitializer;
    import io.netty.channel.socket.SocketChannel;
    
    /**
     * @创建人：zc
     * @时间：2018/9/20 0020
     * @描述：
     */
    public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            // 自定义处理类
            ch.pipeline().addLast(new NettyServerHandler());
        }
    }

上面这段代码为什么要这么写，建议大家先不考虑这个问题，就先这么写。作为一个server端的编写，我们主要在乎的是端口。无论什么框架，什么模型，socket的主要都是port，唯一我们需要写的业务就是NettyServerHandler

package com.zc.module.netty;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    import io.netty.util.ReferenceCountUtil;
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import java.net.InetAddress;
    
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
        private Log logger = LogFactory.getLog(NettyServerHandler.class);
    
        //收到数据时调用
        @Override
        public  void channelRead(ChannelHandlerContext ctx, Object  msg) throws Exception {
            try {
                ByteBuf in = (ByteBuf)msg;
                int readableBytes = in.readableBytes();
                byte[] bytes =new byte[readableBytes];
                in.readBytes(bytes);
                System.out.println(new String(bytes));
                //System.out.print(in.toString(CharsetUtil.UTF_8));
    
                logger.error("服务端接受的消息 : " + msg);
            }finally {
                // 抛弃收到的数据
                ReferenceCountUtil.release(msg);
            }
        }
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            // 当出现异常就关闭连接
            cause.printStackTrace();
            ctx.close();
        }
        /*
         * 建立连接时，返回消息
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //System.out.println("连接的客户端地址:" + ctx.channel().remoteAddress());
            logger.error("连接的客户端地址:" + ctx.channel().remoteAddress());
            logger.error("连接的客户端ID:" + ctx.channel().id());
            ctx.writeAndFlush("client"+ InetAddress.getLocalHost().getHostName() + "success connected！ \n");
            System.out.println("connection");
            //StaticVar.ctxList.add(ctx);
            //StaticVar.chc = ctx;
            super.channelActive(ctx);
        }
    }