Netty框架之协议应用一（redis客户端简单实现以及使用websocket实现弹幕功能）

Posted 2022-11-26 木兮君

前言

今天小编带大家继续学习netty框架的应用，接下来的几篇文章都是讲解其应用，帮助大家理解一些高级框架底层的一些网络传输。话不多说进入正题。

redis客户端实现

redis网络传输采用的是文本协议，同时兼顾了二进制协议的优点，体积小，既保证了传输的效率又保证了可读性。可谓一箭双雕。相信大家在开发过程中使用过redis的客户端jedis，jedis也是非常好的客户端但是他使用的是BIO模式。目前可能采用Lettuce比较多，当然Redisson有更加强大的功能。小编推荐使用Lettuce，大家可以自行搜索为什么选择他。

redis报文

上面介绍了redis的一些基本情况，下面带大家看一下redis的文本协议报文是如何的：

相信看完报文的格式不需要小编解释，大家也非常清楚了吧。当然了，看到这儿大家是否有疑问啊，平时基本上不会这样去写命令，一般都会使用set name bob，即可。也清晰易懂，这是redis的内联命令，一般为：命令 + 空格 + 参数。这边小编使用telnet来稍微演示一下。

客户端实现

既然知道了redis的报文格式，解析的话无非就是前面是个符号，然后最后以\\r\\n结尾，这样就解析了一行报文。那小编来实现一个简易客户端，首先来咱们来理一下实现流程：

因为netty已经提供了对redis协议的支持，所以实现起来就比较简单了，encode相对decode简单多了，因为encode阶段已经知道要封装的参数是什么了，而decode稍微麻烦点，这里小编做一些解释：

1. redisDecoder提供了相应的解码器，前三个主要针对了简单的返回结果，如状态码，错误信息
2. 当有返回单个结果和多个结果的时候，必须要有结果头信息，单个结果则为BulkStringHeaderRedisMessage,然后则是他的内容BulkStringRedisContent。两个组合则为FullBulkStringRedisMessage。当返回为多个结果，结果头为ArrayHeaderRedisMessage,然后需要讲多个结果合在一起，则需要ArrayRedisMessage当然里面包含了多个单个结果集。这样就出现了上面的结构图，希望小编讲解清楚了。
3. 多个结果解码的时候首先解码ArrayHeaderRedisMessage。然后解析上面两个之后聚合到RedisBulkStringAggregator之后再交给RedisArrayAggregator(这个在代码演示中就可以看到)，也就是下面的箭头应该是反方向的传递。

代码示例

public class RedisClient 
    private Channel channel ;
    public void openConnection(String host, int port) throws InterruptedException 
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup(1)).
                channel(NiosocketChannel.class).
                handler(new ChannelInitializer<Channel>() 
                    @Override
                    protected void initChannel(Channel ch) throws Exception 
                        ch.pipeline().addLast("decoder",new RedisDecoder());
                        ch.pipeline().addLast("bulk-aggregator",new RedisBulkStringAggregator());
                        ch.pipeline().addLast("array-aggregator",new RedisArrayAggregator());
                        ch.pipeline().addLast("encode",new RedisEncoder());
                        ch.pipeline().addLast("handler",new MyRedisHandler());
                    
                );
        channel = bootstrap.connect(host, port).sync().channel();
        System.out.println("连接成功");
    

    public static void main(String[] args) throws IOException, InterruptedException 
        RedisClient client = new RedisClient();
        client.openConnection("127.0.0.1",6379);
        BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
        while (true) 
            String s = in.readLine();
            System.out.print(">");
            client.channel.writeAndFlush(s);
        
    
    private class MyRedisHandler extends ChannelDuplexHandler 

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception 
            if(!(msg instanceof String))
                ctx.write(msg);
                return;
            
            String cmd = (String) msg;
            String[] split = cmd.split("\\\\s+");
            List<RedisMessage> redisMessages = new ArrayList<>(split.length);
            for (String commend : split) 
                redisMessages.add(new FullBulkStringRedisMessage(Unpooled.wrappedBuffer(commend.getBytes())));
            
            RedisMessage arrayRedisMessage = new ArrayRedisMessage(redisMessages);
            super.write(ctx, arrayRedisMessage, promise);
        

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
            printAggregatedRedisResponse((RedisMessage) msg);
        
        private void printAggregatedRedisResponse(RedisMessage msg) 
            if (msg instanceof SimpleStringRedisMessage) 
                System.out.println(((SimpleStringRedisMessage) msg).content());
             else if (msg instanceof ErrorRedisMessage) 
                System.out.println(((ErrorRedisMessage) msg).content());
             else if (msg instanceof IntegerRedisMessage) 
                System.out.println(((IntegerRedisMessage) msg).value());
             else if (msg instanceof FullBulkStringRedisMessage) 
                System.out.println(getString((FullBulkStringRedisMessage) msg));
             else if (msg instanceof ArrayRedisMessage) 
                for (RedisMessage child : ((ArrayRedisMessage) msg).children()) 
                    printAggregatedRedisResponse(child);
                
             else 
                throw new CodecException("unknown message type: " + msg);
            
        
        private  String getString(FullBulkStringRedisMessage msg) 
            if (msg.isNull()) 
                return "(null)";
            
            return msg.content().toString(CharsetUtil.UTF_8);
        
    

测试结果：

到此，redis的客户端就实现了，大家也可以看看Lettuce的实现代码。

websocket协议

接下来小编要分享的是弹幕是如何实现的，B站弹幕文化兴起后，小编就很感兴趣，有想过这个怎么实现的，一是靠定时任务，二是靠websocket协议。今天小编就解释一下websocket协议。

websocket协议：是html5开始提供浏览器以及服务之间的进行全双工二进制通信协议，是一种基于TCP连接上进行消息传递，同一时刻既可以接受也可以发送消息，想比http半双工协议性能有很大的提高。

半双工以及全双工的区别

上面有讲到全双工和半双工那小编用下图简单解释一下：

通过上面的图大家应该很好的理解了全双工通信与半双工通信的区别了吧，接下来带大家看下websocket报文

websocket协议报文

看到协议报文是否和小编一样懵，那稍微解释一下吧：

1. 上面的0123456789为比特位，0-7为一个字节，所以上面总共是四个字节，标示是0,1,2,3。
2. 直接看下面表格：

列名	含义
FIN	是否是最后一个包，0表示后面还有，1表示最后
RSV	RSV1,RSV2,RSV3扩展协议，目前没有用到为保留
opcode	有4位，可以表示16种消息类型。 0x0:中间数据包，0x1:表示一个text文本类型数据包，0x2:表示一个binary二进制类型数据包，0x3-7暴露，0x8:表示一个断开连接数据包，0x9:表示一个ping，0xA:表示一个pong，0xB-F:保留
MASK	消息是否进过掩码加密处理，0否1是
Payload length	0-125:表示payload真实长度，126后面两个字节16位来表示消息体的真实长度，127：后8个字节64位表示消息体的真实长度
Masking-key	掩码
Payload Data	数据

websocket的处理流程

看完了报文，咱们来看下弹幕系统处理流程

实现弹幕系统

好了，咱们用代码实现一下：
弹幕服务代码：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

import java.io.RandomAccessFile;
import java.net.URL;
import java.nio.ByteBuffer;

/**
 * 弹幕server
 *
 * @author hasee
 * @version v1.0
 * @since 2021/8/14 20:18
 */
public class BarrageServer 
    ChannelGroup channels;

    private ByteBuf indexPage;
    
        try 
            initStaticPage();
         catch (Exception e) 
            e.printStackTrace();
        
    
    public void openServer(int port) throws InterruptedException 
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));
        bootstrap.channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<Channel>() 
            @Override
            protected void initChannel(Channel ch) throws Exception 
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("http decode",new HttpRequestDecoder());
                pipeline.addLast("http aggregator",new HttpObjectAggregator(65536));
                pipeline.addLast("http encode",new HttpResponseEncoder());
                pipeline.addLast("http servlet",new MyHttpServlet());
                pipeline.addLast("ws protocol",new WebSocketServerProtocolHandler("/ws"));
                pipeline.addLast("ws handler",new MyWsServlet());
            
        );
        ChannelFuture sync = bootstrap.bind(port).sync();
        channels=new DefaultChannelGroup(sync.channel().eventLoop());
        System.out.println("服务开启成功");
    
    private void initStaticPage() throws Exception 
        URL location = BarrageServer.class.getProtectionDomain().getCodeSource().getLocation();
        String path = location.toURI() + "WebsocketBarrage.html";
        path = !path.contains("file:") ? path : path.substring(5);

        RandomAccessFile file = new RandomAccessFile(path, "r");//4
        ByteBuffer dst = ByteBuffer.allocate((int) file.length());
        file.getChannel().read(dst);
        dst.flip();
        indexPage = Unpooled.wrappedBuffer(dst);
        file.close();
    
    private class MyHttpServlet extends SimpleChannelInboundHandler<FullHttpRequest> 
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception 
            if (request.uri().equals("/")) 
                DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
                response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
                response.headers().set(HttpHeaderNames.CONTENT_LENGTH, indexPage.capacity());
                response.content().writeBytes(indexPage.duplicate());
                ctx.writeAndFlush(response);
             else if (request.uri().equals("/ws")) 
            	//引用数+1，让消息不释放
                ctx.fireChannelRead(request.retain());// 转到webSocket 协议进行处理
            
        
    

    private class MyWsServlet extends  SimpleChannelInboundHandler<TextWebSocketFrame> 
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception 
            // 接收客户端发送的消息
            System.out.println(msg.text());
            //封装一个新的消息，避免回收
            channels.writeAndFlush(new TextWebSocketFrame(msg.text()));
            if (msg.text().equals("add")) 
                channels.add(ctx.channel());
            
        
    

    public static void main(String[] args) throws  Exception 
        BarrageServer server=new BarrageServer();
        server.openServer(8080);
        while (true);
    

测试结果：
多个浏览器都可以接受到。

如果想要弹幕的html，回头可以和小编联系。这边多个浏览器打开，如果关闭自动会关闭管道。

总结

小编今天主要是利用netty框架然后根据redis报文协议以及websocket协议实现了两个小小的功能，后续小编会分享rpc中dubbo协议的实现。同时对于redis后续小编会开辟一个板块专门分享redis的内容，包括器数据格式他的集群模式和常见面试。这边先做个铺垫，犹如小编在讲解dubbo的时候说过要学习分享netty内容一下。好了今天分享到这儿结束了。