搭建netty服务器

系统管理员 2022-05-19 00:46 275阅读 0赞

首先，jar包。

<dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>5.0.0.Alpha2</version>
    </dependency>

简单点，粗暴点，直接all。

然后Server端代码：

public class TPRSTCPServer implements Runnable{
    	//间隔时间
    	protected static int readerIdleTime = 10;
    	
    	private static Logger logger = Logger.getLogger(TPRSTCPServer.class);
    
    	private static final int PORT = ;
    	
    	private TPRSTCPServer(){
    		
    	}
    	private static TPRSTCPServer tcpServer = new TPRSTCPServer();
    	public static TPRSTCPServer getInstance(){
    		return tcpServer;
    	}
    	
    	@Override
    	public void run() {
    		int port = PORT;
    		try {
    			new NettyServer(port).run();
    			System.out.println("server:run()");
    		} catch (Exception e) {
    			logger.error(e);
    		}
    		
    	}
    	
    }
    class NettyServer {
    	private int port;
    
    	public NettyServer(int port) {
    		super();
    		this.port = port;
    	}
    
    	public void run() throws Exception {
    
    		/***
    		 * NioEventLoopGroup 是用来处理I/O操作的多线程事件循环器，
    		 * Netty提供了许多不同的EventLoopGroup的实现用来处理不同传输协议。 在这个例子中我们实现了一个服务端的应用，
    		 * 因此会有2个NioEventLoopGroup会被使用。 第一个经常被叫做‘boss’，用来接收进来的连接。
    		 * 第二个经常被叫做‘worker’，用来处理已经被接收的连接， 一旦‘boss’接收到连接，就会把连接信息注册到‘worker’上。
    		 * 如何知道多少个线程已经被使用，如何映射到已经创建的Channels上都需要依赖于EventLoopGroup的实现，
    		 * 并且可以通过构造函数来配置他们的关系。
    		 */
    		EventLoopGroup bossGroup = new NioEventLoopGroup();
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    		System.out.println("运行端口：" + port);
    		try {
    			/**
    			 * ServerBootstrap 是一个启动NIO服务的辅助启动类 你可以在这个服务中直接使用Channel
    			 */
    			ServerBootstrap b = new ServerBootstrap();
    			/**
    			 * 这一步是必须的，如果没有设置group将会报java.lang.IllegalStateException: group not
    			 * set异常
    			 */
    			b = b.group(bossGroup, workerGroup);
    			/***
    			 * ServerSocketChannel以NIO的selector为基础进行实现的，用来接收新的连接
    			 * 这里告诉Channel如何获取新的连接.
    			 */
    			b = b.channel(NioServerSocketChannel.class);
    			/***
    			 * 这里的事件处理类经常会被用来处理一个最近的已经接收的Channel。 ChannelInitializer是一个特殊的处理类，
    			 * 他的目的是帮助使用者配置一个新的Channel。
    			 * 也许你想通过增加一些处理类比如NettyServerHandler来配置一个新的Channel
    			 * 或者其对应的ChannelPipeline来实现你的网络程序。 当你的程序变的复杂时，可能你会增加更多的处理类到pipline上，
    			 * 然后提取这些匿名类到最顶层的类上。
    			 */
    			b = b.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
    				@Override
    				public void initChannel(SocketChannel ch) throws Exception {
    					
    					InetSocketAddress insocket = ch.remoteAddress();
    					String ip = insocket.getAddress().getHostAddress();
    					System.out.println(ip+" is connect");
    					
    					
    					TPRSGatewayService.addGatewayChannel(ip, ch.pipeline());
    					//ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
    					ch.pipeline().addLast(new IdleStateHandler(TPRSTCPServer.readerIdleTime, 0, 0, TimeUnit.SECONDS));
    					//ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
    					ch.pipeline().addLast(new TPRSDiscardServerHandler());// demo1.discard
    					
    					// ch.pipeline().addLast(new
    					// ResponseServerHandler());//demo2.echo
    					// ch.pipeline().addLast(new
    					// TimeServerHandler());//demo3.time
    				}
    			});
    			/***
    			 * 你可以设置这里指定的通道实现的配置参数。 我们正在写一个TCP/IP的服务端，
    			 * 因此我们被允许设置socket的参数选项比如tcpNoDelay和keepAlive。
    			 * 请参考ChannelOption和详细的ChannelConfig实现的接口文档以此可以对ChannelOptions的有一个大概的认识。
    			 */
    			b = b.option(ChannelOption.SO_BACKLOG, 128);
    			/***
    			 * option()是提供给NioServerSocketChannel用来接收进来的连接。
    			 * childOption()是提供给由父管道ServerChannel接收到的连接，
    			 * 在这个例子中也是NioServerSocketChannel。
    			 */
    			b = b.childOption(ChannelOption.SO_KEEPALIVE, true);
    			/***
    			 * 绑定端口并启动去接收进来的连接
    			 */
    			ChannelFuture f = b.bind(port).sync();
    			/**
    			 * 这里会一直等待，直到socket被关闭
    			 */
    			f.channel().closeFuture().sync();
    		} finally {
    			/***
    			 * 关闭
    			 */
    			workerGroup.shutdownGracefully();
    			bossGroup.shutdownGracefully();
    		}
    	}
    
    	
    }

readerIdleTime理解为心跳间隔时间。端口自行设置，其他已有注解。

handler类，消息处理：

public class TPRSDiscardServerHandler extends ChannelHandlerAdapter {
    	
    	static Logger logger = Logger.getLogger(TPRSDiscardServerHandler.class);
    	private static List<String> program1List = new ArrayList<String>();
        private static List<String> program2List = new ArrayList<String>();
    	
    	private int lossConnectCount = 0;
    	private int outTimes = 10;
    	/**
    	 * 心跳，时间设置在IdleStateHandler中
    	 * 总的时间 lossConnectCount*time 超时则关闭通道
    	 */
    	@Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    		String ip = insocket.getAddress().getHostAddress();
            if (evt instanceof IdleStateEvent){
                IdleStateEvent event = (IdleStateEvent)evt;
                if (event.state()== IdleState.READER_IDLE){
                    lossConnectCount++;
                    if (lossConnectCount>outTimes){
                    	logger.info("关闭"+ip+"不活跃通道!");
                        close(ctx, null);
                        
                    }
                }
            }else {
                super.userEventTriggered(ctx,evt);
            }
        }
    	
    	/**
    	 * 这里我们覆盖了chanelRead()事件处理方法。 每当从客户端收到新的数据时， 这个方法会在收到消息时被调用，
    	 * 这个例子中，收到的消息的类型是ByteBuf
    	 * 
    	 * @param ctx
    	 *            通道处理的上下文信息
    	 * @param msg
    	 *            接收的消息
    	 * @throws UnsupportedEncodingException 
    	 */
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws UnsupportedEncodingException {
    		//心跳复位，你也可以放入parseHelper中复位，这里认为一旦收到消息就算一次心跳
    		lossConnectCount = 0;
    		
    		try {
    			ByteBuf buf = (ByteBuf)msg;  
    			
                InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    			String ip = insocket.getAddress().getHostAddress();
    			//新连接存入map
    			ChannelPipeline sc = TPRSGatewayService.getGatewayChannel(ip);
    			if(sc == null){
    				TPRSGatewayService.addGatewayChannel(ip, ctx.pipeline());
    			}
    			
    			try {
    				byte[] result1 = new byte[buf.readableBytes()];
    				buf.readBytes(result1);
    				 ByteArrayInputStream is = new ByteArrayInputStream(result1);
    		         BufferedReader br = new BufferedReader(new InputStreamReader(is, "GBK"));
    		         String aline = "";
    		         while ((aline = br.readLine()) != null) {
    		        	 parseHelper(aline, ip);
    		         }
    		         
    			} catch (Exception e) {
    				logger.error(e);
    			}
    			
    		} finally {
    			/**
    			 * ByteBuf是一个引用计数对象，这个对象必须显示地调用release()方法来释放。
    			 * 请记住处理器的职责是释放所有传递到处理器的引用计数对象。
    			 */
    			// 抛弃收到的数据
    			ReferenceCountUtil.release(msg);
    		}
    
    	}
    
    	/***
    	 * 这个方法会在发生异常时触发
    	 * 
    	 * @param ctx
    	 * @param cause
    	 */
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		/**
    		 * exceptionCaught() 事件处理方法是当出现 Throwable 对象才会被调用，即当 Netty 由于 IO
    		 * 错误或者处理器在处理事件时抛出的异常时。在大部分情况下，捕获的异常应该被记录下来 并且把关联的 channel
    		 * 给关闭掉。然而这个方法的处理方式会在遇到不同异常的情况下有不 同的实现，比如你可能想在关闭连接之前发送一个错误码的响应消息。
    		 */
    		// 出现异常就关闭
    		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    		String ip = insocket.getAddress().getHostAddress();
    		logger.info(ip+" 出现异常！");
    		cause.printStackTrace();
    		close(ctx, null);
    	}
    	
    	@Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //channel失效处理,客户端下线或者强制退出等任何情况都触发这个方法
    		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    		String ip = insocket.getAddress().getHostAddress();
    		logger.info(ip+" 出现异常！");
            super.channelInactive(ctx);
            close(ctx, null);
        }
    	@Override
    	public void close(ChannelHandlerContext ctx, ChannelPromise promise)
    	throws Exception {
    		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    		String ip = insocket.getAddress().getHostAddress();
    		logger.info(ip+" close.....");
    		TPRSGatewayService.removeGatewayChannel(ip);
    		ctx.close();
    	}
    	
    	private void parseHelper(String msg,String ip) throws Exception {
    		System.out.println("接收的消息："+msg);
            //根据规则处理消息。。。。
    	}
    	
    }

然后是保存channel的辅助类：

public class TPRSGatewayService {
    	private static Logger logger = Logger.getLogger(TPRSGatewayService.class);
    	private static Map<String, ChannelPipeline> map = new ConcurrentHashMap<>();
    	
    	
    	
    	
    	public static void addGatewayChannel(String ip, ChannelPipeline channelPipeline) {
    		map.put(ip, channelPipeline);
    	}
    	
    	
    
    	public static Map<String, ChannelPipeline> getChannels() {
    		return map;
    	}
    	
    	
    
    	public static ChannelPipeline getGatewayChannel(String ip) {
    		try {
    			return map.get(ip);
    		} catch (Exception e) {
    			logger.error(e);
    			return null;
    		}
    	}
    	
    	
    
    	public static void removeGatewayChannel(String ip) {
    		try {
    			System.out.println(ip+"移除");
    			map.remove(ip);
    			
    		} catch (Exception e) {
    			logger.error(e);
    		}
    	}
    	
    	
    	
    	public static boolean sendToClient(String msg,String ip){
    		ChannelPipeline ch = map.get(ip);
    		if(ch!=null){
    			
    			try {
    				logger.info(ip+" : "+msg);
    				byte[] bytes = msg.getBytes("gbk");
    				ch.writeAndFlush(Unpooled.copiedBuffer(bytes));
    				return true;
    			} catch (Exception e) {
    				map.remove(ip);
    				return false;
    			}
    		}else{
    			System.out.println(" 该 链 接 不 存 在 ！");
    
    		}
    		return false;
    	}
    	
    	
    }

byte\[\] bytes = msg.getBytes("gbk");这里根据终端能接收的字符集类型设定，本文中原项目类型是utf8，但是终端是只能识别gbk，所以要getBytes("gbk")。

然后就可以测试了：

// 将规则跑起来
    		public static void main(String[] args) {
    			TPRSTCPServer server = TPRSTCPServer.getInstance();
    			Thread t = new Thread(server);// 创建客户端处理线程
    			t.start();// 启动线程
    			Scanner scan = new Scanner(System.in);
    			while (true) {
    				System.out.println("输入操作：");
    				String op = scan.nextLine();
    				if ("1".equals(op)) {
    					for (String key : TPRSGatewayService.getChannels().keySet()) {
    						System.out.println("key = " + key);
    					}
    				} else{
    					System.out.println("输入地址");
    					String ClientIp = scan.nextLine();
    					System.out.println("输入地址是：" + ClientIp);
    //				    String ClientIp = "192.168.1.135";
    					System.out.println("输入消息");
    					String message = scan.nextLine();
    					System.out.println("输入消息：" + message);
    
    					boolean s = TPRSGatewayService.sendToClient(message, ClientIp);
    					System.out.println("消息发送状态：" + s);
    				}
    
    			
    			}
    			
    		}

累了困了写篇博客，提神醒脑。