通过Netty/Socket/C语言三种方式向Redis服务器发送命令-轻识

本文通过使用Netty,Java的Socket和C语言Socket这三种方式,基于RESP协议,向Redis服务器发送一个set命令.

向Redis服务器发送命令,即与Redis服务器通信,必须基于RESP协议. 就好像在B站看2021苹果秋季发布会的视频底层数据传输必须基于TCP协议一样.

RESP协议是一个简单的协议.它的协议格式如下

*<number of arguments> CR LF$<number of bytes of argument 1> CR LF <argument data> CR LF...$<number of bytes of argument N> CR LF <argument data> CR LF

【基于Java的Socket】

import java.io.IOException;import java.net.Socket;public class RedisOfSocket {
    private static final String CRLF = "\r\n";    public static void main(String[] args) throws IOException {        String key = "k6";        String value = "v6";        sendCmd(key, value);    }
    private static void sendCmd(String key, String value) throws IOException {
        // 1.建立连接        Socket client = new Socket("127.0.0.1", 6379);
        // 2.组装命令        StringBuilder command = new StringBuilder();
        // *<参数个数> CRLF        String number = "*3" + CRLF;        command.append(number);
        // $<参数字节数>CRLF<参数>CRLF        String cmd = "$3" + CRLF + "SET" + CRLF;        command.append(cmd);
        // $<参数字节数>CRLF<参数>CRLF        cmd = "$" + key.getBytes().length + CRLF + key + CRLF;        command.append(cmd);
        // $<参数字节数>CRLF<参数>CRLF        cmd = "$" + value.getBytes().length + CRLF + value + CRLF;        command.append(cmd);

        // 3.向服务器发送命令        client.getOutputStream().write(command.toString().getBytes());


        // 4.接收服务器响应        byte[] response = new byte[1024];        int c = client.getInputStream().read(response);        System.out.println(new String(response, 0, c));    }}

在程序运行之前,执行get k6未获取到数据,程序执行之后,再执行get k6就获取到了v6 .

同时我们通过Wireshark工具抓取了网络包,如下

【通过Netty方式】
以上是基于Java的Socket方式向Redis服务器发送了SET命令,接下来通过Netty的方式同样向Redis服务器发送SET命令.


import io.netty.bootstrap.Bootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.codec.string.StringEncoder;import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
@Slf4jpublic class RedisOfNetty {
    public static void main(String[] args) {
        Bootstrap bootstrap = new Bootstrap();        EventLoopGroup group = new NioEventLoopGroup();
        bootstrap.group(group)                .channel(NioSocketChannel.class)                .handler(new ChannelInitializer<NioSocketChannel>() {                    @Override                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline channelPipeline = ch.pipeline();                        channelPipeline.addLast(new StringEncoder());                        channelPipeline.addLast(new StringDecoder());                        // 将自定义的处理器添加到Pipeline里                        channelPipeline.addLast(new ClientInHandler());
                    }                });
        bootstrap.connect(new InetSocketAddress("127.0.0.1", 6379));
    }}


import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import lombok.extern.slf4j.Slf4j;
//自定义处理器@Slf4jpublic class ClientInHandler extends SimpleChannelInboundHandler<String> {
    private static final String CRLF = "\r\n";
    // 我们的程序连接到Redis服务器之后,会回调这个方法,在这个方法里,向Redis服务器发送SET命令     @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String key = "k9";        String value = "v9";
        // 2.组装命令        StringBuilder command = new StringBuilder();
        // *<参数个数> CRLF        String number = "*3" + CRLF;        command.append(number);
        // $<参数字节数>CRLF<参数>CRLF        String cmd = "$3" + CRLF + "SET" + CRLF;        command.append(cmd);
        // $<参数字节数>CRLF<参数>CRLF        cmd = "$" + key.getBytes().length + CRLF + key + CRLF;        command.append(cmd);
        // $<参数字节数>CRLF<参数>CRLF        cmd = "$" + value.getBytes().length + CRLF + value + CRLF;        command.append(cmd);

        // 3.向服务器发送命令        ctx.writeAndFlush(command);

    }

    @Override    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println("接收到服务端的响应: " + msg);
    }}

底层的RESP协议都是一样的,只是通过Netty的方式,将命令发送出去而已.

【通过C语言方式】

#include <stdio.h>#include <stdlib.h>#include <string.h>#include <sys/stat.h>#include <sys/socket.h>#include <sys/types.h>#include <sys/un.h>#include <errno.h>#include <stddef.h>#include <unistd.h>#include <string.h>#include <arpa/inet.h>#include <sys/types.h>#include <sys/socket.h>#include <netinet/in.h>#include <netinet/ip.h>

int main(int argc, char **argv){    struct sockaddr_in sockaddr;
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    bzero(&sockaddr, sizeof(sockaddr));    sockaddr.sin_family = AF_INET;    inet_pton(AF_INET, "127.0.0.1", &sockaddr.sin_addr);    sockaddr.sin_port=htons(6379);
    // 建立连接    connect(fd, (struct sockaddr *)&sockaddr, sizeof(sockaddr));
    // 发送命令    write(fd, "*3\r\n", 4);    write(fd, "$3\r\nSET\r\n", 9);    write(fd, "$2\r\nk7\r\n", 8);    write(fd, "$2\r\nv7\r\n", 8);
    // 读取返回值    char buf[1024];    memset(buf, '\0', sizeof(buf));    read(fd, buf, sizeof(buf));    printf("%s\n", buf);
    close(fd);
    return 0;}

通过C语言的方式,更能清晰的看出来RESP协议的面貌,如何向Redis服务器发送数据的.

如上图所示,我们同样抓取了网络包,这一次我们的C语言程序向Redis服务器发送了2个数据,加起来29个字节. 在第一个我们的Java的Socket实验中,客户端只发送了一次就把29个字节发送出去了,因为当时只调用了一次write, 29个字节也足够小,不存在拆包的情况. 而这次C语言中,我们调用了4次write, 实际发送了2次网络写. 出现了粘包情况. 但是基于RESP协议, Redis服务器自然能知道到哪里是命令的结束.

在Redis的源码中,服务端接收到客户端的命令之后,会根据\r\n进行切割处理我们客户端发送的命令.

在我们的日常工作中,无时无刻不在接触协议. 除了TCP/IP协议, 这里说的RESP协议, 还有Dubbo的协议, RocketMQ的协议, 包括之前我们介绍如何给JVM发送命令dump出来线程栈.

一些皆协议!