首先，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){ ("关闭"+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(); (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(); (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(); (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 { (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); } } }

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