3.netty和protobuf

1.ChannelGroup可以免遍历由netty提供,覆盖remove方法即可触发删除channel\

2.群聊私聊

13.群聊私聊简单原理图

3.netty心跳检测机制,客户端对服务器有没有读写(读,写空闲)
//IdleStateHandler(3,5,7,TimeUnite.SECONDS)是netty提供的检测状态的处理器,也加到pipeline,读,写,读写都没有
//并在handler类实现方法userEventTriggered处理事件

   @Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if(evt instanceof IdleStateEvent) {//将  evt 向下转型 IdleStateEventIdleStateEvent event = (IdleStateEvent) evt;String eventType = null;switch (event.state()) {case READER_IDLE:eventType = "读空闲";break;case WRITER_IDLE:eventType = "写空闲";break;case ALL_IDLE:eventType = "读写空闲";break;}System.out.println(ctx.channel().remoteAddress() + "--超时时间--" + eventType);System.out.println("服务器做相应处理..");//如果发生空闲，我们关闭通道// ctx.channel().close();}}//init注意的顺序pipeline.addLast("MyHttpServerCodec",new HttpServerCodec());

2. 增加一个自定义的handler

pipeline.addLast(new IdleStateHandler(7000,7000,10, TimeUnit.SECONDS));
pipeline.addLast("MyTestHttpServerHandler", new TestHttpServerHandler());

4.websocker实现全双工(两边可以通信,半双工只能一边通信)的长连接,
//因为大数据时http会发送多次请求获取数据,所以要升级用ws协议
//html有WebSocket对象,发送http请求然后200状态码转为101为ws协议

public class MyServer {public static void main(String[] args) throws Exception{//创建两个线程组EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLooptry {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup);serverBootstrap.channel(NioServerSocketChannel.class);serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//相当于链式编程//因为基于http协议，使用http的编码和解码器pipeline.addLast(new HttpServerCodec());//是以块方式写，添加ChunkedWriteHandler处理器pipeline.addLast(new ChunkedWriteHandler());/*说明1. http数据在传输过程中是分段, HttpObjectAggregator ，就是可以将多个段聚合2. 这就就是为什么，当浏览器发送大量数据时，就会发出多次http请求*/pipeline.addLast(new HttpObjectAggregator(8192));/*说明1. 对应websocket ，它的数据是以 帧(frame) 形式传递2. 可以看到WebSocketFrame 下面有六个子类3. 浏览器请求时 ws://localhost:7000/hello 表示请求的uri4. WebSocketServerProtocolHandler 核心功能是将 http协议升级为 ws协议 , 保持长连接5. 是通过一个 状态码 101*/pipeline.addLast(new WebSocketServerProtocolHandler("/hello2"));//自定义的handler ，处理业务逻辑pipeline.addLast(new MyTextWebSocketFrameHandler());}});//启动服务器ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();channelFuture.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {System.out.println("服务器收到消息 " + msg.text());//回复消息ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " " + msg.text()));}//当web客户端连接后， 触发方法@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {//id 表示唯一的值，LongText 是唯一的 ShortText 不是唯一System.out.println("handlerAdded 被调用" + ctx.channel().id().asLongText());System.out.println("handlerAdded 被调用" + ctx.channel().id().asShortText());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {System.out.println("handlerRemoved 被调用" + ctx.channel().id().asLongText());}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常发生 " + cause.getMessage());ctx.close(); //关闭连接}
}

5.编码(cli)—二进制–>解码(server)

1.netty提供的 StringEncoder ObjectEncoder
2.但是使用java底层序列化和反序列化效率低
3.无法跨语言,google protobuf可以解决 ,很多公司由http+json–>tcp+protobuf

6.protobuf (是RPC remote procedure call远程过程调用) 跨平台 生成的大对象包含多个小对象

1.引入依赖
2.写 xxx.proto文件,idea安装插件protobuf

      pipeline.addLast("encoder", new ProtobufEncoder()); //netty需要加netty自带protobuf解编码器//普通的写法
syntax = "proto3";option java_outer_classname="StudentPOJO"; // 外部类名, 文件名message Student {int32 id = 1;//Student类的属性string name = 2; //
}//升级版//需要注意包名,要根据自己的项目包创建,没有可不指定//版本//生成外部类文件名 message Student{ enum DataType{ StudentType =0; //enum编号从0开始WorkType =1;}DataType data_type=1; //标识传入那个类型的id WorkType oneof dataBody{  //相当于在对象中选一个作为属性Student student =2; Worker worker =3;}}  message Student{  //看文档,不要写java语言类型int32 id=1; //属性的序号string name=2;}  message Work{  //看文档,不要写java语言类型int32 id=1; //属性的序号string name=2;}  

3.打开下载好的 protoc.exe生成java文件

protoc.exe --java_out=.  Student.proto //有空格隔开,生成到当前文件夹,输入Student.proto文件,文件名在配置文件中写

4.客户端

         StudentPOJO.Student.new Builder.setId(4).setName("ss").build();//可以发送多种对象,而不是一个对象//添加编解码器if(0 == random) { //发送Student 对象myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType).setStudent(MyDataInfo.Student.newBuilder().setId(5).setName("玉麒麟 卢俊义").build()).build();} else { // 发送一个Worker 对象myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.WorkerType).setWorker(MyDataInfo.Worker.newBuilder().setAge(20).setName("老李").build()).build();}

//完整代码 StudentPOJO.proto

syntax = "proto3";
option optimize_for = SPEED; // 加快解析
option java_package="com.atguigu.netty.codec2";   //指定生成到哪个包下!!!
option java_outer_classname="MyDataInfo"; // 外部类名, 文件名//protobuf 可以使用message 管理其他的message
message MyMessage {//定义一个枚举类型enum DataType {//可以设置枚举类型,然后在java里面判断StudentType = 0; //在proto3 要求enum的编号从0开始WorkerType = 1;}//用data_type 来标识传的是哪一个枚举类型DataType data_type = 1;//表示每次枚举类型最多只能出现其中的一个, 节省空间,另外一个对象如果在java取值变为空值,oneof dataBody {Student student = 2;Worker worker = 3;}}message Student {int32 id = 1;//Student类的属性string name = 2; //
}
message Worker {string name=1;int32 age=2;
}

public class NettyServer {public static void main(String[] args) throws Exception {//创建BossGroup 和 WorkerGroup//说明//1. 创建两个线程组 bossGroup 和 workerGroup//2. bossGroup 只是处理连接请求 , 真正的和客户端业务处理，会交给 workerGroup完成//3. 两个都是无限循环//4. bossGroup 和 workerGroup 含有的子线程(NioEventLoop)的个数//   默认实际 cpu核数 * 2EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup(); //8try {//创建服务器端的启动对象，配置参数ServerBootstrap bootstrap = new ServerBootstrap();//使用链式编程来进行设置bootstrap.group(bossGroup, workerGroup) //设置两个线程组.channel(NioServerSocketChannel.class) //使用NioSocketChannel 作为服务器的通道实现.option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列得到连接个数.childOption(ChannelOption.SO_KEEPALIVE, true) //设置保持活动连接状态
//                    .handler(null) // 该 handler对应 bossGroup , childHandler 对应 workerGroup.childHandler(new ChannelInitializer<SocketChannel>() {//创建一个通道初始化对象(匿名对象)//给pipeline 设置处理器@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//在pipeline加入ProtoBufDecoder//指定对哪种对象进行解码pipeline.addLast("decoder", new ProtobufDecoder(MyDataInfo.MyMessage.getDefaultInstance()));  //!!!入站 解码.pipeline.addLast(new NettyServerHandler());}}); // 给我们的workerGroup 的 EventLoop 对应的管道设置处理器System.out.println(".....服务器 is ready...");//绑定一个端口并且同步, 生成了一个 ChannelFuture 对象//启动服务器(并绑定端口)ChannelFuture cf = bootstrap.bind(6668).sync();//给cf 注册监听器，监控我们关心的事件cf.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (cf.isSuccess()) {System.out.println("监听端口 6668 成功");} else {System.out.println("监听端口 6668 失败");}}});//对关闭通道进行监听cf.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}}

//处理

public class NettyServerHandler extends SimpleChannelInboundHandler<MyDataInfo.MyMessage> {//读取数据实际(这里我们可以读取客户端发送的消息)/*1. ChannelHandlerContext ctx:上下文对象, 含有 管道pipeline , 通道channel, 地址2. Object msg: 就是客户端发送的数据 默认Object*/@Overridepublic void channelRead0(ChannelHandlerContext ctx, MyDataInfo.MyMessage msg) throws Exception {//根据dataType 来显示不同的信息MyDataInfo.MyMessage.DataType dataType = msg.getDataType();if(dataType == MyDataInfo.MyMessage.DataType.StudentType) {MyDataInfo.Student student = msg.getStudent();System.out.println("学生id=" + student.getId() + " 学生名字=" + student.getName());} else if(dataType == MyDataInfo.MyMessage.DataType.WorkerType) {MyDataInfo.Worker worker = msg.getWorker();System.out.println("工人的名字=" + worker.getName() + " 年龄=" + worker.getAge());} else {System.out.println("传输的类型不正确");}}//    //读取数据实际(这里我们可以读取客户端发送的消息)
//    /*
//    1. ChannelHandlerContext ctx:上下文对象, 含有 管道pipeline , 通道channel, 地址
//    2. Object msg: 就是客户端发送的数据 默认Object
//     */
//    @Override
//    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//
//        //读取从客户端发送的StudentPojo.Student
//
//        StudentPOJO.Student student = (StudentPOJO.Student) msg;
//
//        System.out.println("客户端发送的数据 id=" + student.getId() + " 名字=" + student.getName());
//    }//数据读取完毕@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {//writeAndFlush 是 write + flush//将数据写入到缓存，并刷新//一般讲，我们对这个发送的数据进行编码ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端~(>^ω^<)喵1", CharsetUtil.UTF_8));}//处理异常, 一般是需要关闭通道@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}
}

//客户端

public class NettyClient {public static void main(String[] args) throws Exception {//客户端需要一个事件循环组EventLoopGroup group = new NioEventLoopGroup();try {//创建客户端启动对象//注意客户端使用的不是 ServerBootstrap 而是 BootstrapBootstrap bootstrap = new Bootstrap();//设置相关参数bootstrap.group(group) //设置线程组.channel(NioSocketChannel.class) // 设置客户端通道的实现类(反射).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//在pipeline中加入 ProtoBufEncoderpipeline.addLast("encoder", new ProtobufEncoder()); //!!!入站编码二进制位java编码pipeline.addLast(new NettyClientHandler()); //加入自己的处理器}});System.out.println("客户端 ok..");//启动客户端去连接服务器端//关于 ChannelFuture 要分析，涉及到netty的异步模型ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();//给关闭通道进行监听channelFuture.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
}

4.入站(服务器数据向cli) 出站(cli->server)(整个过程在pipeline里面进行) (想象为一个管道入栈需要解码(二进制变为数据),出站需要需要编码[变成二进制])

5.netty处理沾包粘包问题(需要相互判断接收的字节数,不然都是按一个包来处理,而我们是需要分开处理小的包)

//但是分段的数据会被handler和decode重复处理

6.netty handler链 图14.handler链

//如果发送的数据,不是我要编码的就不处理,所以我们要注意传入数据和传出数据类型要一样
//客户端和服务端都写解码和编码

7.解码器 RaplayingDecoder处理界面不用判断是否有足够的数据

   //但是不是所有ByteBuf操作都支持,会抛异常,速度会变慢(信息变碎片->分片太多) 

8.其他解码器

1.LineBasedFrameDecoder 行尾使用 \n或者\r\n分隔符解析数据
2.DelimiterBasedFrameDecoder 自定义特殊字符进行分割
3.LengthFieldBasedFrameDecoder指定长度来识别包,解决沾包粘包问题
4.好用的对数据进行压缩ZlibDecoder
5.httpObjectDecoder

9.maven项目使用netty,整合Log4j

10.TCP粘包(关键在数据长度)(小的包通过算法合成大包,可能出现接收的大包不完整,部分小包位置的问题)
和拆包问题

1.netty解决,自己定义协议对象+解密编码器,接收者得到长度,才得到数据(在这里不用判断长度,因为用长度创建byte[]数组,长度错误就报错) //客户端连续发送数据,就有粘包问题

//协议对象

public class MessageProtocol {private int len; //关键private byte[] content;public int getLen() {return len;}public void setLen(int len) {this.len = len;}public byte[] getContent() {return content;}public void setContent(byte[] content) {this.content = content;}
}

//解码器和加密器,可以分别识别出先写入大小和内容,解决粘包拆包问题

public class MyMessageDecoder extends ReplayingDecoder<Void> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {System.out.println("MyMessageDecoder decode 被调用");//需要将得到二进制字节码-> MessageProtocol 数据包(对象)int length = in.readInt();//!!!!!!!!!!!!!!!!!!!核心步骤,得到的数据长度,然后再创建byte数组接收byte[] content = new byte[length];in.readBytes(content);//封装成 MessageProtocol 对象，放入 out， 传递下一个handler业务处理MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length); //messageProtocol.setContent(content);out.add(messageProtocol);}
}
public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {System.out.println("MyMessageEncoder encode 方法被调用");out.writeInt(msg.getLen());out.writeBytes(msg.getContent());}
}

public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {private int count;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送10条数据 "今天天气冷，吃火锅" 编号for(int i = 0; i< 5; i++) {String mes = "今天天气冷，吃火锅";byte[] content = mes.getBytes(Charset.forName("utf-8"));int length = mes.getBytes(Charset.forName("utf-8")).length;//创建协议包对象MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);ctx.writeAndFlush(messageProtocol);}}//    @Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {int len = msg.getLen();byte[] content = msg.getContent();System.out.println("客户端接收到消息如下");System.out.println("长度=" + len);System.out.println("内容=" + new String(content, Charset.forName("utf-8")));System.out.println("客户端接收消息数量=" + (++this.count));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常消息=" + cause.getMessage());ctx.close();}
}//server

//处理业务的handler
public class MyServerHandler extends SimpleChannelInboundHandler<MessageProtocol>{private int count;@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {//cause.printStackTrace();ctx.close();}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {//接收到数据，并处理int len = msg.getLen();byte[] content = msg.getContent();System.out.println();System.out.println();System.out.println();System.out.println("服务器接收到信息如下");System.out.println("长度=" + len);System.out.println("内容=" + new String(content, Charset.forName("utf-8")));System.out.println("服务器接收到消息包数量=" + (++this.count));//回复消息String responseContent = UUID.randomUUID().toString();int responseLen = responseContent.getBytes("utf-8").length;byte[]  responseContent2 = responseContent.getBytes("utf-8");//构建一个协议包MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(responseLen);messageProtocol.setContent(responseContent2);//向所有channel发送数据ctx.writeAndFlush(messageProtocol);}
}