赞
踩
转载:Springboot+Netty搭建基于TCP/IP协议的服务端(一)
<!-- netty依赖 springboot2.x自动导入版本 -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
NettyApplication.java
package com.mingshine.cloud; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; @SpringBootApplication @EnableAsync @Configuration public class NettyApplication { public static void main(String[] args) { SpringApplication.run(NettyApplication.class, args); } // @Async // @Override // public void run(String... args) throws Exception { // new BootNettyServer().bind(8888); // } @Bean public BootNettyServer mBootNettyServer() throws Exception { BootNettyServer bootNettyServer = new BootNettyServer(); bootNettyServer.bind(8888); return bootNettyServer; } }
BootNettyChannelInboundHandlerAdapter.java
package com.mingshine.cloud; import java.io.IOException; import java.net.InetSocketAddress; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; /** * I/O数据读写处理类 */ public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter { /** * 从客户端收到新的数据时,这个方法会在收到消息时被调用 * * @param ctx * @param msg */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException { System.out.println("channelRead:read msg:" + msg.toString()); //回应客户端 ctx.write("I got it"); } /** * 从客户端收到新的数据、读取完成时调用 * * @param ctx */ @Override public void channelReadComplete(ChannelHandlerContext ctx) throws IOException { System.out.println("channelReadComplete"); ctx.flush(); } /** * 当出现 Throwable 对象才会被调用,即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时 * * @param ctx * @param cause */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException { System.out.println("exceptionCaught"); cause.printStackTrace(); ctx.close();//抛出异常,断开与客户端的连接 } /** * 客户端与服务端第一次建立连接时 执行 * * @param ctx * @throws Exception */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException { super.channelActive(ctx); ctx.channel().read(); InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress(); String clientIp = insocket.getAddress().getHostAddress(); //此处不能使用ctx.close(),否则客户端始终无法与服务端建立连接 System.out.println("channelActive:" + clientIp + ctx.name()); } /** * 客户端与服务端 断连时 执行 * * @param ctx * @throws Exception */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException { super.channelInactive(ctx); InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress(); String clientIp = insocket.getAddress().getHostAddress(); ctx.close(); //断开连接时,必须关闭,否则造成资源浪费,并发量很大情况下可能造成宕机 System.out.println("channelInactive:" + clientIp); } /** * 服务端当read超时, 会调用这个方法 * * @param ctx * @param evt * @throws Exception */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException { super.userEventTriggered(ctx, evt); InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress(); String clientIp = insocket.getAddress().getHostAddress(); ctx.close();//超时时断开连接 System.out.println("userEventTriggered:" + clientIp); } @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { System.out.println("channelRegistered"); } @Override public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { System.out.println("channelUnregistered"); } @Override public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { System.out.println("channelWritabilityChanged"); } }
BootNettyChannelInitializer.java
package com.mingshine.cloud; import io.netty.channel.Channel; import io.netty.channel.ChannelInitializer; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; /** * 通道初始化 */ public class BootNettyChannelInitializer<SocketChannel> extends ChannelInitializer<Channel> { @Override protected void initChannel(Channel ch) throws Exception { // ChannelOutboundHandler,依照逆序执行 ch.pipeline().addLast("encoder", new StringEncoder()); // 属于ChannelInboundHandler,依照顺序执行 ch.pipeline().addLast("decoder", new StringDecoder()); /** * 自定义ChannelInboundHandlerAdapter */ ch.pipeline().addLast(new BootNettyChannelInboundHandlerAdapter()); } }
BootNettyServer.java
package com.mingshine.cloud; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.AdaptiveRecvByteBufAllocator; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; /** * netty的server * */ public class BootNettyServer { public void bind(int port) throws Exception { /** * 配置服务端的NIO线程组 * NioEventLoopGroup 是用来处理I/O操作的Reactor线程组 * bossGroup:用来接收进来的连接,workerGroup:用来处理已经被接收的连接,进行socketChannel的网络读写, * bossGroup接收到连接后就会把连接信息注册到workerGroup * workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍 */ EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { /** * ServerBootstrap 是一个启动NIO服务的辅助启动类 */ ServerBootstrap serverBootstrap = new ServerBootstrap(); /** * 设置group,将bossGroup, workerGroup线程组传递到ServerBootstrap */ serverBootstrap = serverBootstrap.group(bossGroup, workerGroup); /** * ServerSocketChannel是以NIO的selector为基础进行实现的,用来接收新的连接,这里告诉Channel通过NioServerSocketChannel获取新的连接 */ serverBootstrap = serverBootstrap.channel(NioServerSocketChannel.class); /** * option是设置 bossGroup,childOption是设置workerGroup * netty 默认数据包传输大小为1024字节, 设置它可以自动调整下一次缓冲区建立时分配的空间大小,避免内存的浪费 最小 初始化 最大 (根据生产环境实际情况来定) * 使用对象池,重用缓冲区 */ serverBootstrap = serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576)); serverBootstrap = serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576)); /** * 设置 I/O处理类,主要用于网络I/O事件,记录日志,编码、解码消息 */ serverBootstrap = serverBootstrap.childHandler(new BootNettyChannelInitializer<SocketChannel>()); System.out.println("netty server start success!"); /** * 绑定端口,同步等待成功 */ ChannelFuture f = serverBootstrap.bind(port).sync(); /** * 等待服务器监听端口关闭 */ f.channel().closeFuture().sync(); } catch (InterruptedException e) { } finally { /** * 退出,释放线程池资源 */ bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。