您当前的位置:首页 > 电脑百科 > 站长技术 > 服务器

从Redis、HTTP协议,看Nett协议设计,我发现了个惊天大秘密

时间:2023-01-03 17:26:42  来源:今日头条  作者:Java解白

1. 协议的作用

TCP/IP 中消息传输基于流的方式,没有边界

协议的目的就是划定消息的边界,制定通信双方要共同遵守的通信规则

2. redis 协议

如果我们要向 Redis 服务器发送一条 set name Nyima 的指令,需要遵守如下协议

// 该指令一共有3部分,每条指令之后都要添加回车与换行符
*3rn
// 第一个指令的长度是3
$3rn
// 第一个指令是set指令
setrn
// 下面的指令以此类推
$4rn
namern
$5rn
Nyimarn
复制代码

客户端代码如下

public class RedisClient {
    static final Logger log = LoggerFactory.getLogger(StudyServer.class);
    public static void main(String[] args) {
        NioEventLoopGroup group =  new NioEventLoopGroup();
        try {
            ChannelFuture channelFuture = new Bootstrap()
                    .group(group)
                    .channel(NIOSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            // 打印日志
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    // 回车与换行符
                                    final byte[] LINE = {'r','n'};
                                    // 获得ByteBuf
                                    ByteBuf buffer = ctx.alloc().buffer();
                                    // 连接建立后,向Redis中发送一条指令,注意添加回车与换行
                                    // set name Nyima
                                    buffer.writeBytes("*3".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$3".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("set".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$4".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("name".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$5".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("Nyima".getBytes());
                                    buffer.writeBytes(LINE);
                                    ctx.writeAndFlush(buffer);
                                }

                            });
                        }
                    })
                    .connect(new .NETSocketAddress("localhost", 6379));
            channelFuture.sync();
            // 关闭channel
            channelFuture.channel().close().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭group
            group.shutdownGracefully();
        }
    }
}
复制代码

控制台打印结果

1600 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler  - [id: 0x28c994f1, L:/127.0.0.1:60792 - R:localhost/127.0.0.1:6379] WRITE: 34B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 2a 33 0d 0a 24 33 0d 0a 73 65 74 0d 0a 24 34 0d |*3..$3..set..$4.|
|00000010| 0a 6e 61 6d 65 0d 0a 24 35 0d 0a 4e 79 69 6d 61 |.name..$5..Nyima|
|00000020| 0d 0a                                           |..              |
+--------+-------------------------------------------------+----------------+
复制代码

Redis 中查询执行结果

 

3. HTTP 协议

HTTP 协议在请求行请求头中都有很多的内容,自己实现较为困难,可以使用 HttpServerCodec 作为服务器端的解码器与编码器,来处理 HTTP 请求

// HttpServerCodec 中既有请求的解码器 HttpRequestDecoder 又有响应的编码器 HttpResponseEncoder
// Codec(CodeCombine) 一般代表该类既作为 编码器 又作为 解码器
public final class HttpServerCodec extends CombinedChannelDuplexHandler<HttpRequestDecoder, HttpResponseEncoder>
        implements HttpServerUpgradeHandler.SourceCodec
复制代码

服务器代码

public class HttpServer {
    static final Logger log = LoggerFactory.getLogger(StudyServer.class);

    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        new ServerBootstrap()
                .group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        // 作为服务器,使用 HttpServerCodec 作为编码器与解码器
                        ch.pipeline().addLast(new HttpServerCodec());
                        // 服务器只处理HTTPRequest
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) {
                                // 获得请求uri
                                log.debug(msg.uri());

                                // 获得完整响应,设置版本号与状态码
                                DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
                                // 设置响应内容
                                byte[] bytes = "<h1>Hello, World!</h1>".getBytes(StandardCharsets.UTF_8);
                                // 设置响应体长度,避免浏览器一直接收响应内容
                                response.headers().setInt(CONTENT_LENGTH, bytes.length);
                                // 设置响应体
                                response.content().writeBytes(bytes);

                                // 写回响应
                                ctx.writeAndFlush(response);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}
复制代码

服务器负责处理请求并响应浏览器。所以只需要处理 HTTP 请求即可

// 服务器只处理HTTPRequest
ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>()
复制代码

获得请求后,需要返回响应给浏览器。需要创建响应对象 DefaultFullHttpResponse,设置 HTTP 版本号及状态码,为避免浏览器获得响应后,因为获得 CONTENT_LENGTH 而一直空转,需要添加 CONTENT_LENGTH 字段,表明响应体中数据的具体长度

// 获得完整响应,设置版本号与状态码
DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
// 设置响应内容
byte[] bytes = "<h1>Hello, World!</h1>".getBytes(StandardCharsets.UTF_8);
// 设置响应体长度,避免浏览器一直接收响应内容
response.headers().setInt(CONTENT_LENGTH, bytes.length);
// 设置响应体
response.content().writeBytes(bytes);
复制代码

运行结果

浏览器

 

控制台

// 请求内容
1714 [nioEventLoopGroup-2-2] DEBUG io.netty.handler.logging.LoggingHandler  - [id: 0x72630ef7, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:55503] READ: 688B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 47 45 54 20 2f 66 61 76 69 63 6f 6e 2e 69 63 6f |GET /favicon.ico|
|00000010| 20 48 54 54 50 2f 31 2e 31 0d 0a 48 6f 73 74 3a | HTTP/1.1..Host:|
|00000020| 20 6c 6f 63 61 6c 68 6f 73 74 3a 38 30 38 30 0d | localhost:8080.|
|00000030| 0a 43 6f 6e 6e 65 63 74 69 6f 6e 3a 20 6b 65 65 |.Connection: kee|
|00000040| 70 2d 61 6c 69 76 65 0d 0a 50 72 61 67 6d 61 3a |p-alive..Pragma:|
....

// 响应内容
1716 [nioEventLoopGroup-2-2] DEBUG io.netty.handler.logging.LoggingHandler  - [id: 0x72630ef7, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:55503] WRITE: 61B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d |HTTP/1.1 200 OK.|
|00000010| 0a 43 6f 6e 74 65 6e 74 2d 4c 65 6e 67 74 68 3a |.Content-Length:|
|00000020| 20 32 32 0d 0a 0d 0a 3c 68 31 3e 48 65 6c 6c 6f | 22....<h1>Hello|
|00000030| 2c 20 57 6f 72 6c 64 21 3c 2f 68 31 3e          |, World!</h1>   |
+--------+-------------------------------------------------+----------------+

复制代码

4. 自定义协议

组成要素

  • 魔数:用来在第一时间判定接收的数据是否为无效数据包
  • 版本号:可以支持协议的升级
  • 序列化算法
  • :消息正文到底采用哪种序列化反序列化方式
    • 如:json、protobuf、hessian、jdk
  • 指令类型:是登录、注册、单聊、群聊… 跟业务相关
  • 请求序号:为了双工通信,提供异步能力
  • 正文长度
  • 消息正文

编码器与解码器

public class MessageCodec extends ByteToMessageCodec<Message> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        // 设置魔数 4个字节
        out.writeBytes(new byte[]{'N','Y','I','M'});
        // 设置版本号 1个字节
        out.writeByte(1);
        // 设置序列化方式 1个字节
        out.writeByte(1);
        // 设置指令类型 1个字节
        out.writeByte(msg.getMessageType());
        // 设置请求序号 4个字节
        out.writeInt(msg.getSequenceId());
        // 为了补齐为16个字节,填充1个字节的数据
        out.writeByte(0xff);

        // 获得序列化后的msg
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(msg);
        byte[] bytes = bos.toByteArray();

        // 获得并设置正文长度 长度用4个字节标识
        out.writeInt(bytes.length);
        // 设置消息正文
        out.writeBytes(bytes);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 获取魔数
        int magic = in.readInt();
        // 获取版本号
        byte version = in.readByte();
        // 获得序列化方式
        byte seqType = in.readByte();
        // 获得指令类型
        byte messageType = in.readByte();
        // 获得请求序号
        int sequenceId = in.readInt();
        // 移除补齐字节
        in.readByte();
        // 获得正文长度
        int length = in.readInt();
        // 获得正文
        byte[] bytes = new byte[length];
        in.readBytes(bytes, 0, length);
        ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
        Message message = (Message) ois.readObject();
		// 将信息放入List中,传递给下一个handler
        out.add(message);
        
        // 打印获得的信息正文
        System.out.println("===========魔数===========");
        System.out.println(magic);
        System.out.println("===========版本号===========");
        System.out.println(version);
        System.out.println("===========序列化方法===========");
        System.out.println(seqType);
        System.out.println("===========指令类型===========");
        System.out.println(messageType);
        System.out.println("===========请求序号===========");
        System.out.println(sequenceId);
        System.out.println("===========正文长度===========");
        System.out.println(length);
        System.out.println("===========正文===========");
        System.out.println(message);
    }
}

复制代码
  • 编码器与解码器方法源于父类 ByteToMessageCodec,通过该类可以自定义编码器与解码器, 泛型类型为被编码与被解码的类。此处使用了自定义类 Message,代表消息
public class MessageCodec extends ByteToMessageCodec<Message>
复制代码
  • 编码器负责将附加信息与正文信息写入到 ByteBuf 中,其中附加信息总字节数最好为 2n,不足需要补齐。正文内容如果为对象,需要通过序列化将其放入到 ByteBuf 中
  • 解码器负责将 ByteBuf 中的信息取出,并放入 List 中,该 List 用于将信息传递给下一个 handler

编写测试类

public class TestCodec {
    static final org.slf4j.Logger log = LoggerFactory.getLogger(StudyServer.class);
    public static void main(String[] args) throws Exception {
        EmbeddedChannel channel = new EmbeddedChannel();
        // 添加解码器,避免粘包半包问题
        channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 12, 4, 0, 0));
        channel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
        channel.pipeline().addLast(new MessageCodec());
        LoginRequestMessage user = new LoginRequestMessage("Nyima", "123");

        // 测试编码与解码
        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
        new MessageCodec().encode(null, user, byteBuf);
        channel.writeInbound(byteBuf);
    }
}
复制代码
  • 测试类中用到了 LengthFieldBasedFrameDecoder,避免粘包半包问题
  • 通过 MessageCodec 的 encode 方法将附加信息与正文写入到 ByteBuf 中,通过 channel 执行入站操作。入站时会调用 decode 方法进行解码

运行结果

 


 

@Sharable 注解

为了提高 handler 的复用率,可以将 handler 创建为 handler 对象,然后在不同的 channel 中使用该 handler 对象进行处理操作

LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG);
// 不同的channel中使用同一个handler对象,提高复用率
channel1.pipeline().addLast(loggingHandler);
channel2.pipeline().addLast(loggingHandler);
复制代码

但是并不是所有的 handler 都能通过这种方法来提高复用率的,例如
LengthFieldBasedFrameDecoder。如果多个 channel 中使用同一个 LengthFieldBasedFrameDecoder 对象,则可能发生如下问题

  • channel1 中收到了一个半包,LengthFieldBasedFrameDecoder 发现不是一条完整的数据,则没有继续向下传播
  • 此时 channel2 中也收到了一个半包,因为两个 channel 使用了同一个 LengthFieldBasedFrameDecoder,存入其中的数据刚好拼凑成了一个完整的数据包。LengthFieldBasedFrameDecoder 让该数据包继续向下传播,最终引发错误

为了提高 handler 的复用率,同时又避免出现一些并发问题,Netty 中原生的 handler 中用 @Sharable 注解来标明,该 handler 能否在多个 channel 中共享。

只有带有该注解,才能通过对象的方式被共享,否则无法被共享

自定义编解码器能否使用 @Sharable 注解

这需要根据自定义的 handler 的处理逻辑进行分析

我们的 MessageCodec 本身接收的是
LengthFieldBasedFrameDecoder 处理之后的数据,那么数据肯定是完整的,按分析来说是可以添加 @Sharable 注解的

但是实际情况我们并不能添加该注解,会抛出异常信息 ChannelHandler
cn.nyimac.study.day8.protocol.MessageCodec is not allowed to be shared

  • 因为 MessageCodec 继承自 ByteToMessageCodec,ByteToMessageCodec 类的注解如下
  •  

这就意味着 ByteToMessageCodec 不能被多个 channel 所共享的

  • 原因:因为该类的目标是:将 ByteBuf 转化为 Message,意味着传进该 handler 的数据还未被处理过。所以传过来的 ByteBuf 可能并不是完整的数据,如果共享则会出现问题

如果想要共享,需要怎么办呢?

继承 MessageToMessageDecoder 即可。 该类的目标是:将已经被处理的完整数据再次被处理。传过来的 Message 如果是被处理过的完整数据,那么被共享也就不会出现问题了,也就可以使用 @Sharable 注解了。实现方式与 ByteToMessageCodec 类似

@ChannelHandler.Sharable
public class MessageSharableCodec extends MessageToMessageCodec<ByteBuf, Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> out) throws Exception {
        ...
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
		...
    }
}


Tags:Netty   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,如有任何标注错误或版权侵犯请与我们联系(Email:2595517585@qq.com),我们将及时更正、删除,谢谢。
▌相关推荐
1. 协议的作用TCP/IP 中消息传输基于流的方式,没有边界协议的目的就是划定消息的边界,制定通信双方要共同遵守的通信规则2. Redis 协议如果我们要向 Redis 服务器发送一条 set...【详细内容】
2023-01-03  Tags: Netty  点击:(0)  评论:(0)  加入收藏
一、IO模型 阻塞式IO模型 非阻塞式IO模型 IO复用 信号驱动式IO 异步IO Linux IO流程 2. 各个IO模型的比较 NIO的优势 事件驱动模型 避免多线程 单线程处理多任务 非阻塞,IO...【详细内容】
2022-10-02  Tags: Netty  点击:(50)  评论:(0)  加入收藏
关于NettyNetty 是一个利用 Java 的高级网络的能力,隐藏其背后的复杂性而提供一个易于使用的 API 的客户端/服务器框架。MAVEN依赖 <dependencies> <!-- https://mvnrepos...【详细内容】
2022-09-27  Tags: Netty  点击:(49)  评论:(0)  加入收藏
一、目录介绍 功能梳理 具体实现二、需求梳理通过前面两章内容的学习,我们基本学会了如何使用 Netty 建立一个长连接,接下来我们就在这个基础上,实现一个单机版的 im 系统。主...【详细内容】
2022-09-23  Tags: Netty  点击:(97)  评论:(0)  加入收藏
websokct 使用 netty 实现websocket 是一个全双工的长连接通讯话不多说直接上代码。。服务端的代码实现public void init() throws InterruptedException { EventLoo...【详细内容】
2022-09-12  Tags: Netty  点击:(84)  评论:(0)  加入收藏
时间轮是一个高性能、低消耗的数据结构,它适合用非准实时,延迟的短平快任务,例如心跳检测。在Netty、Kafka、Zookeeper中都有使用。时间轮可通过时间与任务存储分离的形式,轻松...【详细内容】
2022-08-23  Tags: Netty  点击:(80)  评论:(0)  加入收藏
简介DNS的全称domain name system,既然是一个系统就有客户端和服务器之分。一般情况来说我们并不需要感知这个DNS客户端的存在,因为我们在浏览器访问某个域名的时候,浏览器作为...【详细内容】
2022-07-20  Tags: Netty  点击:(124)  评论:(0)  加入收藏
本文笔者来为大家介绍下Netty的核心引擎Reactor的运转架构,希望通过本文的介绍能够让大家对Reactor是如何驱动着整个Netty框架的运转有一个全面的认识。也为我们后续进一步...【详细内容】
2022-07-05  Tags: Netty  点击:(86)  评论:(0)  加入收藏
在过去的五个月里,我一直在我当前的项目中使用 Spring Webflux,我还编写了很多 Nodejs 应用程序,并且使用 Promise 样式编码(async/await)的方式也几乎相同,而 Webflux 使用 Mon...【详细内容】
2022-06-27  Tags: Netty  点击:(299)  评论:(0)  加入收藏
概述本文介绍什么是 TCP 粘包和拆包现象,并通过 Netty 编写详细的案例来重现 TCP 粘包问题,最后再通过一个 Netty 的 demo 来解决这个问题。具体内容如下 什么是 TCP 粘包和拆...【详细内容】
2022-04-24  Tags: Netty  点击:(115)  评论:(0)  加入收藏
▌简易百科推荐
1. 协议的作用TCP/IP 中消息传输基于流的方式,没有边界协议的目的就是划定消息的边界,制定通信双方要共同遵守的通信规则2. Redis 协议如果我们要向 Redis 服务器发送一条 set...【详细内容】
2023-01-03  Java解白  今日头条  Tags:Netty   点击:(0)  评论:(0)  加入收藏
用phpstudy_pro部署的基于Apace的简易Web网站,部署完存在目录列表的blank.gif,folder.gif,unknown.gif等icons图标不能显示问题。无法正常显示图标 用浏览器查看网页源代码...【详细内容】
2023-01-03  后浪爱分享  今日头条  Tags:Apache   点击:(2)  评论:(0)  加入收藏
Zookeeper 和 KRaft这里有一篇 Kafka 功能改进的 proposal 原文。要了解移除 ZK 的原因,可以仔细看看该文章。以下是对该文章的翻译。动机目前,Kafka 使用 Zookeeper 保存与分...【详细内容】
2022-12-27  Java码农之路  今日头条  Tags:Apache   点击:(13)  评论:(0)  加入收藏
魔兽世界服务端开服架设服务器搭建教程作为魔兽世界资深玩家,你是否有想过自己开服当服主? 相信能看到此篇文章的你心里是有想法的。首先你需要知道开魔兽需要准备哪些东西,我...【详细内容】
2022-12-19  艾西  网易号  Tags:服务器   点击:(19)  评论:(0)  加入收藏
之前有个项目里面,dcs系统中配置了两台opc ua服务器。可是作为client的wincc只能填写其中一台opc ua server的地址,如果第一台opc ua server有问题,wincc不能实现热备访问第二...【详细内容】
2022-12-12  牛鼻子老仙  今日头条  Tags:服务器冗余   点击:(14)  评论:(0)  加入收藏
今天配置下分析下nginx关于URI的部分,另外需要看下DNS的长度.简单而已,DNS长度不超过255个octets.我们是用.来分割域名的,那么每个点之间的长度不超过63个octets.Nginx本身...【详细内容】
2022-12-10  Mac乐  今日头条  Tags:URL   点击:(27)  评论:(0)  加入收藏
我是艾西,还是有很多小白同学问我linux系统服务器怎么远程连接。那么今天我们重点来教教大家如何用电脑远程服务器配上图文教程,让不懂的新手小白一看就会,分分钟上手教程远程...【详细内容】
2022-12-08  艾西  搜狐号  Tags:服务器   点击:(20)  评论:(0)  加入收藏
在本教程中,我们将学习在Windows Server上配置静态ip地址。我们的Windows Server以太网适配器可能配置为从DHCP服务器获取ip地址。在某些网络中,网络上可能没有运行DHCP服务器...【详细内容】
2022-12-08  电子发烧友  今日头条  Tags:Windows Server   点击:(43)  评论:(0)  加入收藏
修改Windows系统实例默认远程端口以Windows Server 2012为例介绍如何修改Windows系统实例默认远程端口。远程连接进入服务器后修改注册表子项PortNumber的值。按快捷键 Win(W...【详细内容】
2022-12-07  艾西  搜狐号  Tags:端口   点击:(19)  评论:(0)  加入收藏
服务器是计算机的一种,它比普通计算机运行更快、负载更高、价格更贵。服务器在网络中为其它客户机(如PC机、智能手机、ATM等终端甚至是火车系统等大型设备)提供计算或者应用服...【详细内容】
2022-12-04  晨阳SEO  搜狐号  Tags:服务器   点击:(20)  评论:(0)  加入收藏
站内最新
站内热门
站内头条