netty编解码器

当 Netty 发送或者接受一个消息的时候，就将会发生一次数据转换。入站消息会被解码：从字节转换为另一种格式（比如 java 对象）；如果是出站消息，它会被编码成字节。
Netty 提供一系列实用的编解码器，他们都实现了 ChannelInboundHadnler 或者 ChannelOutboundHandler 接口。在这些类中，channelRead 方法已经被重写了。以入站为例，对于每个从入站 Channel 读取的消息，这个方法会被调用。随后，它将调用由解码器所提供的 decode() 方法进行解码，并将已经解码的字节转发给 ChannelPipeline 中的下一个 ChannelInboundHandler。

解码器- ByteToMessageDecoder

关系继承图
在这里插入图片描述

小细节

不论解码器 handler 还是编码器 handler 即接收的消息类型必须与待处理的消息类型一致，否则该 handler 不会被执行。
当我们在handler中调用ctx.writeAndFlush()方法后，就会将数据交给ChannelOutboundHandler进行出站处理(如果下一个handler是编码器并且数据类型符合编码器处理的数据类型则会传递给编码器)，只是我们没有去定义出站类而已，若有需求可以自己去实现ChannelOutboundHandler出站类。
解码器一般不需要设置泛型(待处理数据的类型)。

TCP粘包和拆包基本介绍

1.TCP 是面向连接的，面向流的，提供高可靠性服务。收发两端（客户端和服务器端）都要有一一成对的 socket，因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化方法（Nagle 算法），将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包。这样做虽然提高了效率，但是接收端就难于分辨出完整的数据包了，因为面向流的通信是无消息保护边界的
2.由于 TCP 无消息保护边界,需要在接收端处理消息边界问题，也就是我们所说的粘包、拆包问题

在这里插入图片描述

TCP粘包和拆包解决方案

在这里插入图片描述
具体事例：
1.要求客户端发送 5 个 Message 对象，客户端每次发送一个 Message 对象
2.服务器端每次接收一个 Message，分 5 次进行解码，每读取到一个 Message，会回复一个 Message 对象给客户端。
代码：
MessageProtocol.class

package com.haust.tcp;

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;
    }
}

MyServer.class

package com.haust.tcp;

import io.netty.bootstrap.ServerBootstrap;
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.nio.NioServerSocketChannel;

public class MyServer  {
    public static void main(String[] args) throws Exception{
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer()); //自定义一个初始化类


            ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();

        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }


}

MyServerInitializer.class

package com.haust.tcp;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast(new MyMessageDecoder());//解码器
        pipeline.addLast(new MyMessageEncoder());//编码器
        pipeline.addLast(new MyServerHandler());
    }
}

MyServerHandler.class

package com.haust.tcp;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.UUID;

//处理业务的handler
public class MyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
    private int count;

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected 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));

        //回复消息
        System.out.println("服务端开始回复消息------");
        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);

        //客户端需要输入信息，创建一个扫描器
//        MessageProtocol messageProtocol = new MessageProtocol();
//        int inLen=0;
//        byte[] incontent=null ;
//        Scanner scanner = new Scanner(System.in);
//        while (scanner.hasNextLine()) {
//            String msg1 = scanner.nextLine();
//            //将字符串加工成协议包类型
//            inLen = msg1.getBytes().length;
//            incontent = msg1.getBytes();
//            messageProtocol.setLen(inLen);
//            messageProtocol.setContent(incontent);
//            //通过ctx 发送到服务器端
            ctx.writeAndFlush(messageProtocol);
//        }





    }
}

MyClient.class

package com.haust.tcp;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.Scanner;

public class MyClient {
    public static void main(String[] args)  throws  Exception{

        EventLoopGroup group = new NioEventLoopGroup();

        try {

            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class)
                    .handler(new MyClientInitializer()); //自定义一个初始化类

            ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();

            channelFuture.channel().closeFuture().sync();







        }finally {
            group.shutdownGracefully();
        }
    }
}

MyClientInitializer.class

package com.haust.tcp;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {

        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new MyMessageEncoder()); //加入编码器
        pipeline.addLast(new MyMessageDecoder()); //加入解码器
        pipeline.addLast(new MyClientHandler());
    }
}

MyClientHandler.class

package com.haust.tcp;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;

public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {

    private int count;
    @Override
    public 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);

        }

    }

//    @Override
    protected 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));

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("异常消息=" + cause.getMessage());
        ctx.close();
    }
}

MyMessageDecoder.class

package com.haust.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

public class MyMessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println();
        System.out.println();
        System.out.println("MyMessageDecoder decode 被调用");
        //需要将得到二进制字节码-> MessageProtocol 数据包(对象)
        int length = in.readInt();

        byte[] content = new byte[length];
        in.readBytes(content);

        //封装成 MessageProtocol 对象，放入 out， 传递下一个handler业务处理
        MessageProtocol messageProtocol = new MessageProtocol();
        messageProtocol.setLen(length);
        messageProtocol.setContent(content);

        //放入out传给下一个hanlder进行处理
        out.add(messageProtocol);

    }
}

MyMessageEncoder.class

package com.haust.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
        System.out.println("MyMessageEncoder encode 方法被调用");
        out.writeInt(msg.getLen());
        out.writeBytes(msg.getContent());
    }
}

效果：
server端输出

MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=1
服务端开始回复消息------
MyMessageEncoder encode 方法被调用


MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=2
服务端开始回复消息------
MyMessageEncoder encode 方法被调用


MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=3
服务端开始回复消息------
MyMessageEncoder encode 方法被调用


MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=4
服务端开始回复消息------
MyMessageEncoder encode 方法被调用


MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=5
服务端开始回复消息------
MyMessageEncoder encode 方法被调用

client端输出

MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用


MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=f71b9a7d-96c0-4f26-a2ed-5684c30f0a5a
客户端接收消息数量=1


MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=53cee25c-4a81-4d06-8ec8-bf6a664729e8
客户端接收消息数量=2


MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=556d84ce-907b-4228-a722-9965b54e589e
客户端接收消息数量=3


MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=774c3992-a39d-4aa2-9a20-e7fdbd3d4abd
客户端接收消息数量=4


MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=69f8777f-dc61-4f2f-8f28-fdb0680c96d9
客户端接收消息数量=5

Java知识库最新文章

计算距离春节还有多长时间

系统开发系列之WebService（spring框架+ma

springBoot+Cache(自定义有效时间配置)

SpringBoot整合mybatis实现增删改查、分页查

spring教程

SpringBoot+Vue实现美食交流网站的设计与实

加:2022-02-07 13:35:18 更:2022-02-07 13:36:32

360图书馆购物三丰科技阅读网日历万年历 2024年11日历

-2024/11/24 12:08:35-

图片自动播放器
↓图片自动播放器↓

TxT小说阅读器
↓语音阅读,小说下载,古典文学↓

一键清除垃圾
↓轻轻一点,清除系统垃圾↓

图片批量下载器
↓批量下载图片,美女图库↓

网站联系: qq:121756557 email:121756557@qq.com IT数码