Netty之服务端/客户端

た入场券 2023-06-16 10:57 12阅读 0赞

项目总体结构，分为两个项目，服务端和客户端

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70][]

我们先来看下服务端这个项目：

MyServer

package com.yj.server.service;
    
    import javax.annotation.PostConstruct;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    @Component
    public class MyServer {
    
    	@Value("${netty.port}")
    	private int port;
    
    	@PostConstruct
    	public void start() throws Exception {
                    //如果只是监听一个端口的话,这里的个数写几没有什么意义,只会有一个线程做accept
    		EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    		try {
    			ServerBootstrap serverBootstrap = new ServerBootstrap();
    			serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
    					.option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列等待的连接个数
    					.childOption(ChannelOption.SO_KEEPALIVE, true) // 设置保持活动连接状态
    					.childHandler(new ServerInitializer());
    			
    			ChannelFuture cf = serverBootstrap.bind(port).sync();
    			System.out.println("Netty服务端启动成功,开始监听端口:" + port);
    
    			cf.addListener(new ChannelFutureListener() {
    				@Override
    				public void operationComplete(ChannelFuture future) throws Exception {
    					if (cf.isSuccess()) {
    						System.out.println("监听端口成功");
    					} else {
    						System.out.println("监听端口失败");
    					}
    				}
    			});
    			cf.channel().closeFuture().sync();
    		} finally {
    			bossGroup.shutdownGracefully().sync();
    			workerGroup.shutdownGracefully().sync();
    		}
    	}
    }

ServerInitializer

package com.yj.server.service;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    
    public class ServerInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new ServerHandler());
        }
    }

ServerHandler

package com.yj.server.service;
    
    import java.util.concurrent.TimeUnit;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    public class ServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    
            //将 msg 转成一个 ByteBuf
            //ByteBuf 是 Netty 提供的，不是 NIO 的 ByteBuffer.
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("服务端收到消息:" + buf.toString(CharsetUtil.UTF_8));
            ctx.writeAndFlush(buf);
        	
            //比如这里我们有一个非常耗时长的业务
            //解决方案1 用户程序自定义的普通任务
            ctx.channel().eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(3 * 1000);
                        ctx.writeAndFlush(Unpooled.copiedBuffer("taskQueue中的业务1处理完毕", CharsetUtil.UTF_8));
                    } catch (Exception ex) {
                        System.out.println("发生异常" + ex.getMessage());
                    }
                }
            });
    
            ctx.channel().eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(3 * 1000);
                        ctx.writeAndFlush(Unpooled.copiedBuffer("taskQueue中的业务2处理完毕", CharsetUtil.UTF_8));
                        System.out.println("channel code=" + ctx.channel().hashCode());
                    } catch (Exception ex) {
                        System.out.println("发生异常" + ex.getMessage());
                    }
                }
            });
    
            //解决方案2 : 用户自定义定时任务 -》 该任务是提交到 scheduleTaskQueue中
            ctx.channel().eventLoop().schedule(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(3 * 1000);
                        ctx.writeAndFlush(Unpooled.copiedBuffer("scheduleTaskQueue中的业务处理完毕", CharsetUtil.UTF_8));
                        System.out.println("channel code=" + ctx.channel().hashCode());
                    } catch (Exception ex) {
                        System.out.println("发生异常" + ex.getMessage());
                    }
                }
            }, 5, TimeUnit.SECONDS);
        }
    
        //数据读取完毕
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            //writeAndFlush 是 write + flush
            //将数据写入到缓存，并刷新
            //一般讲，我们对这个发送的数据进行编码
            ctx.writeAndFlush(Unpooled.copiedBuffer("服务端数据读取完毕", CharsetUtil.UTF_8));
        }
    
        //处理异常, 一般是需要关闭通道
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

客户端这边

MyClient

package com.yj.client.service;
    
    import javax.annotation.PostConstruct;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    @Component
    public class MyClient {
    	
    	@Value("${netty.host}")
    	private String host;
    	@Value("${netty.port}")
    	private int port;
    
    	@PostConstruct
    	public void start() throws Exception {
    		EventLoopGroup group = new NioEventLoopGroup();
    		try {
    			Bootstrap b = new Bootstrap();
    			b.group(group).channel(NioSocketChannel.class)
    					.handler(new ClientInitializer());
    			ChannelFuture cf = b.connect(host, port).sync();
    			cf.channel().closeFuture().sync();
    		} finally {
    			group.shutdownGracefully().sync();
    		}
    	}
    }

ClientInitializer

package com.yj.client.service;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    
    public class ClientInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new ClientHandler());
    
        }
    }

ClientHandler

package com.yj.client.service;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandler.Sharable;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    @Sharable
    public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) {
    		ctx.writeAndFlush(Unpooled.copiedBuffer("你好,服务端", CharsetUtil.UTF_8));
    	}
    
    	@Override
    	protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
    		System.out.println("客户端接收到数据:" + msg.toString(CharsetUtil.UTF_8));
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    		cause.printStackTrace();
    		ctx.close();
    	}
    }

主要看下客户端这边的日志输出结果：

客户端接收到数据:你好,服务端
    客户端接收到数据:服务端数据读取完毕
    客户端接收到数据:taskQueue中的业务1处理完毕
    客户端接收到数据:taskQueue中的业务2处理完毕
    客户端接收到数据:scheduleTaskQueue中的业务处理完毕

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191202231010614.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1,size_16,color_FFFFFF,t_70