Netty开发redis客户端，Netty发送redis命令，netty解析redis消息

来源：cnblogs　　作者：逃离沙漠　　时间：2018/11/20 22:53:42　　对本文有异议

关键字：Netty开发redis客户端，Netty发送redis命令，netty解析redis消息, netty redis ,redis RESP协议。redis客户端，netty redis协议

我们可以使用redis-cli这个客户端来操作redis，也可以使用window的命令行telnet连接redis。本文，我们的目标是使用netty来实现redis客户端，实现目标为：

1. 启动netty程序
2. 在命令行输入 set mykey hello，由netty发送给redis服务器
3. 在命令行输入 get mykey hello，得到结果：hello
4. 在命令行输入 quit，程序退出

前言

Redis在TCP端口6379（默认，可修改端口）上监听到来的连接，客户端连接到来时，Redis服务器为此创建一个TCP连接。在客户端与服务器端之间传输的每个Redis命令或者数据都以\r\n结尾。当redis服务启动之后，我们可以使用TCP与之链接，连接之后便可以发消息，也会受到redis服务器的消息。而这个消息是有格式的，这个格式是事先商量好的，我们称之为协议，redis的协议叫做RESP，比方说我们有一条redis命令set hello 123,这条命令我们知道它是一条设置命令，通过RESP协议“翻译”一下，他就是这样的:

*3\r\n$3\r\nSET\r\n$5\r\nhello\r\n$3\r\n123\r\n

然后，这条协议通过网络传输(二进制形式)，传到redis服务器，被redis服务器解析，最后完成设置。关于RESP洗衣详细可以看这里.

思路

上面我们介绍了redis是基于TCP传输，并使用了其自己的协议——RESP。RESP其实是数据交换可解析的协议，你可以理解为数据交换的格式，按照此格式组装好要传输的命令，并以二进制的形式由client端发往redis服务端。服务端接收这个消息之后，解析消息，执行命令，并将结果以协议好的格式组装好，传输给client端。client端接收到响应，解释成人类可以看懂的结果展示。

因此，我们可以整理一下思路：

1. 我们需要连接redis服务端，因此需要编写一个netty client端（此处联想一下netty client端的样板代码）。
2. 我们需要向redis服务端发送redis命令，很简单，获取channel，然后write。即channel.write(...)
3. 我们所编写的直白的命令，如set xx,get xx之类的需要编码之后才能传输给redis服务器。
   因此，我们需要 **编码器**。很荣幸netty自带了，可以直接使用。
   这里是 【输出】 所以要有  outbound handler.
4. redis会响应结果给我们，因此我们需要在 chanelRead方法中处理数据。
   这里是 【输入】 所以要有  inbound  handler.

编写代码

上的思路整理好了之后，我们可以写代码了。得益于netty的良好设计，我们只需要把netty client的“样板代码”拷贝过来生成一个client端代码即可。剩下的就是 handler ，decoder ，encoder 。我们需要编写的类有：

RedisClient 见名知义，我们的主类，包含client bootstrap信息。接收用户控制台输入redis命令。
RedisClientInitializer 初始化器，在此添加 handler，decoder，encoder
RedisClientHandler 核心逻辑，需要处理 inbound ，outbound 两种类型事件。

RedisClient 代码如下：

public class RedisClient {
    String host;    //   目标主机
    int port;       //   目标主机端口
    public RedisClient(String host,int port){
        this.host = host;
        this.port = port;
    }
    public void start() throws Exception{
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new RedisClientInitializer());
            Channel channel = bootstrap.connect(host, port).sync().channel();
            System.out.println(" connected to host : " + host + ", port : " + port);
            System.out.println(" type redis's command to communicate with redis-server or type 'quit' to shutdown ");
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
            ChannelFuture lastWriteFuture = null;
            for (;;) {
                String s = in.readLine();
                if(s.equalsIgnoreCase("quit")) {
                    break;
                }
                System.out.print(">");
                lastWriteFuture = channel.writeAndFlush(s);
                lastWriteFuture.addListener(new GenericFutureListener<ChannelFuture>() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            System.err.print("write failed: ");
                            future.cause().printStackTrace(System.err);
                        }
                    }
                });
            }
            if (lastWriteFuture != null) {
                lastWriteFuture.sync();
            }
            System.out.println(" bye ");
        }finally {
            group.shutdownGracefully();
        }
    }
    public static void main(String[] args) throws Exception{
        RedisClient client = new RedisClient("redis-cache2.228",5001);
        client.start();
    }
}

上面代码很长，但是，我们要熟悉netty的套路，它的样板代码就是如此。我们只需要看handler(new RedisClientInitializer()); 这一行，下面的就是一个 for(;;)循环，用来接收我们在控制台输入的redis命令。

RedisClientInitializer代码如下：

public class RedisClientInitializer extends ChannelInitializer<Channel>{
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new RedisDecoder());
        pipeline.addLast(new RedisBulkStringAggregator());
        pipeline.addLast(new RedisArrayAggregator());
        pipeline.addLast(new RedisEncoder());
        pipeline.addLast(new RedisClientHandler());
    }
}

这个类，很简单，上面的几个addLast方法，除了最后一个外，其他都是netty自带的redis协议实现相关的编解码。最后一个是我们自定义的业务逻辑处理器。源码如下：

public class RedisClientHandler extends ChannelDuplexHandler {
    // 发送 redis 命令
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        String[] commands = ((String) msg).split("\\s+");
        List<RedisMessage> children = new ArrayList<>(commands.length);
        for (String cmdString : commands) {
            children.add(new FullBulkStringRedisMessage(ByteBufUtil.writeUtf8(ctx.alloc(), cmdString)));
        }
        RedisMessage request = new ArrayRedisMessage(children);
        ctx.write(request, promise);
    }
    // 接收 redis 响应数据
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        RedisMessage redisMessage = (RedisMessage) msg;
        // 打印响应消息
        printAggregatedRedisResponse(redisMessage);
        // 是否资源
        ReferenceCountUtil.release(redisMessage);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.print("exceptionCaught: ");
        cause.printStackTrace(System.err);
        ctx.close();
    }
    private static 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 static String getString(FullBulkStringRedisMessage msg) {
        if (msg.isNull()) {
            return "(null)";
        }
        return msg.content().toString(CharsetUtil.UTF_8);
    }
}

注意，上面我们讨论过，我们需要两个handler，分别是inbound handler 和outbound handler 。这里我们使用的是ChannelDuplexHandler。这个ChannelDuplexHandler 支持处理 inbound 和 outbound，其定义如下：

public class ChannelDuplexHandler extends ChannelInboundHandlerAdapter implements ChannelOutboundHandler {
  ....
}

运行演示

按照开篇的思路分析，上面我们已经编写好了netty redis client所需要的代码。下面我们需要运行看看。main函数如下：

    public static void main(String[] args) throws Exception{
        RedisClient client = new RedisClient("your-redis-server-ip",6379);
        client.start();
    }

我在本地运行了一下，演示了一些命令：

get ,set 以及错误的命令
expire命令设置超时时间，及 ttl 命令查看超时时间
del 命令删除可以
quit退出程序。

结果如下：

 connected to host : 192.168.2.120, port : 6379
 type redis's command to communicate with redis-server or type 'quit' to shutdown 
get hello
>(null)
set hello
>ERR wrong number of arguments for 'set' command
set hello 123
>OK
expire hello 10
>1
ttl hello
>6
ttl hello
>4
get hello
>(null)
set hello world
>OK
get hello
>world
del hello
>1
quit
 bye 
Process finished with exit code 0