自定义消息协议的实现逻辑

消息协议：这一次消息需要包含两个部分，即消息长度和消息内容本身。
自定义消息编码器︰消息编码器将客户端发送的消息转换成遵守消息协议的消息，即包含消息长度和消息内容的消息
自定义消息解码器∶消息解码器根据消息协议的消息长度，来获得指定长度的消息内容。

自定义编码器

自定义消息协议：

//自定义消息协议
public class MessageProtocal {
    //消息的长度
    private int length;
    //消息的内容
    private byte[] content;
    public int getLength() {
        return length;
    }
    public void setLength(int length) {
        this.length = length;
    }
    public byte[] getContent() {
        return content;
    }
    public void setContent(byte[] content) {
        this.content = content;
    }
}

客户端基本代码

public class NettyClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap bootstrap = new Bootstrap();
        //设置相关的参数
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //添加处理器，分包编码器
                        pipeline.addLast(new MessageEncoder());
                        //添加具体的业务处理器
                        pipeline.addLast(new NettyMessageClientHandler());
                    }
                });
        System.out.println("客户端启动了");
        ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9090).sync();
        channelFuture.channel().closeFuture().sync();
        group.shutdownGracefully();
    }
}

客户端业务代码

public class NettyMessageClientHandler extends SimpleChannelInboundHandler<MessageProtocal> {
    //连接通道创建后要向服务端发送消息
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for(int i=0;i<200;i++){
            String msg = "西安科技大学";
            //创建消息协议对象
            MessageProtocal messageProtocal = new MessageProtocal();
            messageProtocal.setLength(msg.getBytes(StandardCharsets.UTF_8).length);
            messageProtocal.setContent(msg.getBytes(StandardCharsets.UTF_8));
            //发送协议对象,注意此时ctx只能发送Bytebuf数据，因此需要用编码器把它编码成Bytebuf数据
            ctx.writeAndFlush(messageProtocal);
        }
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocal msg) throws Exception {
    }
}

自定义编码器

public class MessageEncoder extends MessageToByteEncoder<MessageProtocal> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocal msg, ByteBuf out) throws Exception {
        out.writeInt(msg.getLength());
        out.writeBytes(msg.getContent());
    }
}

服务端基本代码

public class NettyServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup boosGroup = new NioEventLoopGroup(1);
        EventLoopGroup workGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(boosGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //添加解码器
                        pipeline.addLast(new MessageDecoder());
                        pipeline.addLast(new NettyMessageServerHandler());
                    }
                });
        System.out.println("Netty的服务端启动了");
        ChannelFuture channelFuture = serverBootstrap.bind(9090).sync();
        channelFuture.channel().closeFuture().sync();
        boosGroup.shutdownGracefully();
        workGroup.shutdownGracefully();
    }
}

自定义解码器

//自定义解码器代码
public class MessageDecoder extends ByteToMessageDecoder {
    int length = 0;
    //ctx
    //in:客户端发送来的MessageProtocol编码后的ByteBuf数据
    //out:out里的数据会被放行到下一个handler把解码出来的MessageProtocol放到out里面
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("ByteBuf:"+in);
        //获得前面的4个字节的数据 == 描述实际内容的长度
        if(in.readableBytes()>=4){
            //ByteBuf里面可能有MessageProtocol数据
            if(length==0){
                length = in.readInt();
            }
            //length = 15
            if(in.readableBytes()<length){
                //说明数据还没到齐，等待下一次调用decode
                System.out.println("当前数据量不够，继续等待");
                return;
            }
            //可读数据量>=length ==> 意味着这一次的MessageProtocol的内容已经到齐了
            //创建了一个指定length长度的字节数组
            byte[] content = new byte[length];
            //把ByteBuf里面的指定长度的数据读到content数组中
            in.readBytes(content);
            //创建协议MessageProtocol对象赋值
            MessageProtocal messageProtocal = new MessageProtocal();
            messageProtocal.setLength(length);
            messageProtocal.setContent(content);
            out.add(messageProtocal);
            length=0;
        }
    }
}

服务端业务处理代码

public class NettyMessageServerHandler extends SimpleChannelInboundHandler<MessageProtocal> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocal msg) throws Exception {
        System.out.println("---服务器收到的数据---");
        System.out.println("消息的长度:"+msg.getLength());
        System.out.println("消息的内容:"+new String(msg.getContent(), StandardCharsets.UTF_8));
    }
}

运行结果：

心跳机制

在分布式系统中，心跳机制常常在注册中心组件中提及，比如Zookeeper、Eureka、Nacos等，通过维护客户端的心跳，来判断客户端是否正常在线。如果客户端达到超时次数等预设的条件时，服务端将释放客户端的连接资源。
试想一下，当我们一个用来写数据的通道，它虽然没有下线，但这个通道长时间都不写数据了，是不是我们可以利用心跳机制，关闭此类通道及其对应的客户端

实现客户端发送心跳包

客户端基本代码

public class NettyClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //添加编解码器
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new NettyClientHandler());
                    }
                });
        System.out.println("客户端启动了");
        ChannelFuture channelFuture = bootstrap.connect("127.0.0.1",9090).sync();
        //模拟向服务端发送心跳数据
        String packet = "heartbeat packet";
        Random random = new Random();
        Channel channel = channelFuture.channel();
        while (channel.isActive()){
            //随机的事件来实现时间间隔等待
            int num = random.nextInt(10);
            Thread.sleep(num*1000);
            channel.writeAndFlush(packet);
        }
        group.shutdownGracefully();
    }
}

客户端拦截器

public class NettyClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        System.out.println("客户端收到的数据"+s);
    }
}

IdleStateHandler类描述三种空闲状态
读空闲:在指定时间间隔内没有从Channel中读到数据，将会创建状态为READER_IDLE的IdleStateEvent对象。
写空闲︰在指定时间间隔内没有数据写入到Channel中，将会创建状态为WRITER_IDLE的ldleStateEvent对象。
读写空闲:在指定时间间隔内Channel中没有发生读写操作，将会创建状态为ALL_IDLE的ldleStateEvent对象。
服务端基本代码

public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new StringDecoder());
                        //超时状态处理器会在服务端发现有超过3秒没有没有发生读操作的话会触发超时事件
                        //创建出IdleStateEvent对象，将该对象交给下一个Handler
                        pipeline.addLast(new IdleStateHandler(3,0,0, TimeUnit.SECONDS));
                        //HeartbeatServerHandler必领重写userEventTriggered方法，用来做具体的超时的业务处理
                        pipeline.addLast(new HeartbeatServerHandler());
                    }
                });
        System.out.println("Netty服务端启动了");
        ChannelFuture channelFuture = serverBootstrap.bind(9090).sync();
        channelFuture.channel().closeFuture().sync();
        bossGroup.shutdownGracefully();
        workGroup.shutdownGracefully();
    }
}

服务端业务代码

public class HeartbeatServerHandler extends SimpleChannelInboundHandler<String> {
    int readIdleTimes = 0;
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        System.out.println("服务端收到的心跳"+s);
        channelHandlerContext.writeAndFlush("服务端已经收到了心跳");
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        IdleStateEvent event = (IdleStateEvent)evt;
        switch (event.state()){
            case READER_IDLE:
                readIdleTimes++;
                break;
            case WRITER_IDLE:
                System.out.println("写超时");
                break;
            case ALL_IDLE:
                System.out.println("读写超时");
                break;
        }
        if(readIdleTimes>3){
            System.out.println("读超时超过三次,关闭连接");
            ctx.writeAndFlush("超时关闭");
            ctx.channel().close();
        }
    }
}

Netty自定义消息协议的实现逻辑处理粘包拆包、心跳机制

自定义消息协议的实现逻辑

自定义编码器

心跳机制

实现客户端发送心跳包

热门文章

最新文章

相关电子书