使用springboot+netty处理tcp/ip服务端编程

╰+哭是因爲堅強的太久メ 2022-02-24 04:10 448阅读 0赞

# 1.添加依赖 #

略

# 2.netty服务器启动 #

springboot的bean代码,另开一个线程启动

@Component
    public class NettyServer {
    
    	private static Logger logger = LoggerFactory.getLogger(NettyServer.class);
    
    	// 保存response的map
    	public static Map<String, Response> map = new HashMap<String, Response>();
    	// 保存客户端连接的通道引用
    	public static SocketChannel sc = null;
    
    
    	public static EventLoopGroup acceptor;
    	public static EventLoopGroup worker;
    	
    	@PostConstruct
    	public void init() throws InterruptedException {
    		new NettyServerThread().start();
    		logger.info("nettyServer启动");
    	}
    
    	@PreDestroy
    	public void exit() {
    		acceptor.shutdownGracefully();
    		worker.shutdownGracefully();
    	}
    
    }

具体启动的代码

public class NettyServerThread extends Thread {
    
    	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
    
    	@Override
    	public void run() {
    
    		EventLoopGroup acceptor = new NioEventLoopGroup();
    		EventLoopGroup worker = new NioEventLoopGroup();
    		NettyServer.acceptor = acceptor;
    		NettyServer.worker = worker;
    		ServerBootstrap bootstrap = new ServerBootstrap();
    
    		// 添加boss和worker组
    		bootstrap.group(acceptor, worker);
    		//这句是指定允许等待accept的最大连接数量,我只需要连一个客户端,这里就关掉了,java默认是50个
    		// bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
    		bootstrap.option(ChannelOption.TCP_NODELAY, true);
    		// 用于构造socketchannel工厂
    		bootstrap.channel(NioServerSocketChannel.class);
    
    		/**
    		 * 传入自定义客户端Handle（处理消息）
    		 */
    		bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    			@Override
    			public void initChannel(SocketChannel ch) throws Exception {
    				if (NettyServer.sc == null) {
    					logger.info("来自" + ch.remoteAddress() + "的新连接接入");
    					NettyServer.sc= ch;
    					// 注册handler
    					ch.pipeline().addLast(new ReadTimeoutHandler(10));
    					ch.pipeline().addLast(new MessageHandler());
    				} else {
    					ch.close();
    				}
    			}
    		});
    
    		// 绑定端口，开始接收进来的连接
    		ChannelFuture f;
    		try {
    			f = bootstrap.bind(8888).sync();
    			// 等待服务器 socket 关闭 。
    			f.channel().closeFuture().sync();
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    
    	}
    }

这里是netty启动的核心代码,通过bootstrap添加各种配置来装饰nettyserver并最终启动

# 3.group #

我们都知道nio对比bio最大的特点就是不会阻塞,具体怎么实现的呢,就在  
NioEventLoopGroup里,这里其实是新建了一个线程池去执行一些任务  
这里acceptor里的线程用来维护accept接入新连接的SelectionKey,worker里的线程用来维护客户端的SelectionKey

如果你只给一个线程池,实际上也可以使用,这种情况下acceptor和worker需要完成的工作都会使用这一个线程池中的线程

# 4.Handler #

initChannel方法就是重载accept接入后初始化通道的方法了,通道被accept之后该通道的所有SelectionKey都会通过同一个线程来维护(为了避免线程并发的问题,但他们之间并非一一对应,一个线程可以同时维护多个通道的SelectionKey)

在initChannel方法,我给通道添加了两个Handler,第一个是超时十秒会抛出异常并断开连接,第二个是我自定义的处理客户端发送信息的Handler,netty基本上大部分业务代码会在自定义Handler里编写

public class MessageHandler extends ChannelInboundHandlerAdapter {
    
    	private static Logger logger = LoggerFactory.getLogger(MessageHandler.class);
    
    	/**
    	 * 本方法用于读取客户端发送的信息
    	 * 
    	 * @param ctx
    	 * @param msg
    	 * @throws Exception
    	 */
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    
    		ByteBuf msgByteBuf = (ByteBuf) msg;
    		logger.info(msgByteBuf.toString());
    		byte[] msgBytes = new byte[msgByteBuf.readableBytes()];
    		// msg中存储的是ByteBuf类型的数据，把数据读取到byte[]中
    		msgByteBuf.readBytes(msgBytes);
    		// 释放资源
    		msgByteBuf.release();
    
    		// 可能返回到的msgByteBuf是多条信息拼起来的,把他们拆开分别处理
    		List<byte[]> list = getMsgList(msgBytes);
    
    		// 真正处理信息的方法
    		list.forEach(v -> handler(v, ctx));
    
    	}
    
    	/**
    	 * 切分信息的方法
    	 * 
    	 * @param msgBytes
    	 * @return
    	 */
    	private List<byte[]> getMsgList(byte[] msgBytes) {
    		List<byte[]> list = new ArrayList<byte[]>();
    		//具体业务代码略
    		return list;
    	}
    
    	/**
    	 * 本方法用作处理异常
    	 * 
    	 * @param ctx
    	 * @param cause
    	 * @throws Exception
    	 */
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		if (cause.getClass() == io.netty.handler.timeout.ReadTimeoutException.class) {
    			logger.info("来自" + NettyServer.sc.remoteAddress() + "的连接超时断开");
    		} else {
    			cause.printStackTrace();
    			logger.info("来自" + NettyServer.sc.remoteAddress() + "的连接异常断开");
    			ctx.close();
    		}
    		NettyServer.sc= null;
    	}
    
    	/**
    	 * 信息获取完毕后操作
    	 *
    	 * @param ctx
    	 * @throws Exception
    	 */
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		ctx.flush();
    	}
    
    	/**
    	 * 断开连接时操作
    	 *
    	 * @param ctx
    	 * @throws Exception
    	 */
    	@Override
    	public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    
    		if (NettyServer.sc!= null) {
    			logger.info("来自" + NettyServer.sc.remoteAddress() + "的连接主动断开");
    			NettyServer.sc= null;
    		}
    
    		ctx.fireChannelUnregistered();
    	}
    
    	/**
    	 * 根据信息具体操作的业务方法
    	 * 
    	 * @param msgBytes
    	 * @param ctx
    	 */
    	private void handler(byte[] msgBytes, ChannelHandlerContext ctx) {
    		// 具体业务代码略,可以通过ctx的write和flush方法回应客户端的信息
    	}
    
    
    }

真正重要的是重写的几个方法,下面逐一介绍

### 1.channelUnregistered ###

在客户端断开通道(或其他原因,总之触发了Unregistered这个SelectionKey)时,记录日志,调用ctx.fireChannelUnregistered();做netty关闭通道的一些处理,并把连入的客户端置空

### 2.exceptionCaught ###

处理异常,因为前面加Handler的顺序这个在ReadTimeoutHandler后面,所以ReadTimeoutHandler抛出的异常可以在这里被处理  
如果是ReadTimeoutException,则记录超时断开的日志,否则打印出具体异常,关闭通道,并记录异常断开的日志

### 3.channelReadComplete ###

执行ctx.flush();  
为什么要这么做,因为有时候客户端发送信息不会在发送后清空管道,这样就没有结束标识,read的SelectionKey不会触发,我们这里执行一下刷新

### 4.channelRead ###

这里就是接收到的具体信息,需要注意的是,这里可能是客户端发送的多条信息连起来,所以要按照业务的逻辑切分开分别处理,  
可以在处理后通过传入的ctx写入你的回复