今夜不加班,从紧张的工作节奏中释放了下,打开微信跟之前美女同事(未婚)唠嗑了两下,哎呦没话题,那搞点啥呢?无聊中..........- 算了 记录下这段时间的辛酸旅程,总结下物联网的基础,以此纪念9年的编程生活。
(springboot版)netty服务端创建:
开启netty线程
public class BootNettyServer {
public void bind(int port) {
/**
* 配置服务端的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());
/**
* 绑定端口,同步等待成功
*/
ChannelFuture f = serverBootstrap.bind(port).sync();
/**
* 等待服务器监听端口关闭
*/
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
} finally {
/**
* 退出,释放线程池资源
*/
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
/**
* 通道初始化
* @author Administrator
*
*/
public class BootNettyChannelInitializer extends ChannelInitializer<channel>{/<channel>
@Override
protected void initChannel(Channel ch) throws Exception {
// TODO Auto-generated method stub
// ChannelOutboundHandler,依照逆序执行
ch.pipeline().addLast("encoder", new StringEncoder());
// 属于ChannelInboundHandler,依照顺序执行
ch.pipeline().addLast("decoder", new StringDecoder());
/**
* 自定义ChannelInboundHandlerAdapter
*/
ch.pipeline().addLast(new BootNettyChannelInboundHandlerAdapter());
}
}
/**
* handler 过滤器
* @author Administrator
*
*/
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);
}
}
以上是netty的服务端创建内容,应用场景: 当多传感器(支持多规约)向服务器端传递,接收数据类。下一篇将讲述netty下 如果处理 tcp 拆包的问题。
閱讀更多 編程俠 的文章