編寫本身的handler簡單操做redis

　　RESP是Redis Serialization Protocol的簡稱,也就是專門爲redis設計的一套序列化協議。這個協議比較簡單，簡單的說就是發送請求的時候按Redis 約定的數據格式進行發送，解析數據的時候按redis規定的響應數據格式進行相應。

　　不管是jedis仍是lettuce， 最終發給redis的數據與接收到redis的響應數據的格式必須知足redis的協議要求。

1. RedisClient 類

package cn.xm.netty.example.redis3;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;

import java.io.BufferedReader;
import java.io.InputStreamReader;

public class RedisClient {

    private static final String HOST = System.getProperty("host", "192.168.145.139");
    private static final int PORT = Integer.parseInt(System.getProperty("port", "6379"));

    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new RedisClientHandler());
                        }
                    });

            // Start the connection attempt.
            Channel ch = b.connect(HOST, PORT).sync().channel();

            // Read commands from the stdin.
            System.out.println("Enter Redis commands (quit to end)");
            ChannelFuture lastWriteFuture = null;
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
            for (; ; ) {
                final String input = in.readLine();
                final String line = input != null ? input.trim() : null;
                if (line == null || "quit".equalsIgnoreCase(line)) { // EOF or "quit"
                    ch.close().sync();
                    break;
                } else if (line.isEmpty()) { // skip `enter` or `enter` with spaces.
                    continue;
                }
                // Sends the received line to the server.
                lastWriteFuture = ch.writeAndFlush(line);
                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);
                        }
                    }
                });
            }

            // Wait until all messages are flushed before closing the channel.
            if (lastWriteFuture != null) {
                lastWriteFuture.sync();
            }
        } finally {
            group.shutdownGracefully();
        }
    }
}

　　這個類比較簡單就是一直for 循環等待控制檯輸入數據。而且添加的handler 只有一個handler。 也就是輸入輸出都是一個handler。

2.  RedisClientHandler

package cn.xm.netty.example.redis3;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.util.CharsetUtil;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * An example Redis client handler. This handler read input from STDIN and write output to STDOUT.
 */
public class RedisClientHandler extends ChannelDuplexHandler {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        // 轉換髮出去的數據格式
        msg = rehandleRequest(msg);
        ctx.writeAndFlush(Unpooled.copiedBuffer(msg.toString(), CharsetUtil.UTF_8));
    }

    /**
     * 從新處理消息，處理爲 RESP 承認的數據
     * set foo bar
     * 對應下面數據
     * *3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n
     */
    private String rehandleRequest(Object msg) {
        String result = msg.toString().trim();
        String[] params = result.split(" ");
        List<String> allParam = new ArrayList<>();
        Arrays.stream(params).forEach(s -> {
            allParam.add("$" + s.length() + "\r\n" + s + "\r\n"); // 參數前$length\r\n, 參數後增長 \r\n
        });
        allParam.add(0, "*" + allParam.size() + "\r\n");
        StringBuilder stringBuilder = new StringBuilder();
        allParam.forEach(p -> {
            stringBuilder.append(p);
        });
        return stringBuilder.toString();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf byteBuf = (ByteBuf) msg;
        byte[] bytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(bytes);
        String result = new String(bytes);
        // 轉換接受到的數據格式
        result = rehandleResponse(result).toString();
        System.out.println(result);
    }

    /**
     * 從新處理響應消息
     */
    private Object rehandleResponse(String result) {
        // 狀態恢復 - 「+OK\r\n」
        if (result.startsWith("+")) {
            return result.substring(1, result.length() - 2);
        }

        // 錯誤回覆（error reply）的第一個字節是 "-"。例如 `flushallE` 返回的 `-ERR unknown command 'flushallE'\r\n`
        if (result.startsWith("-")) {
            return result.substring(1, result.length() - 2);
        }

        // 整數回覆（integer reply）的第一個字節是 ":"。 例如 `llen mylist` 查看list 大小返回的 `:3\r\n`
        if (result.startsWith(":")) {
            return result.substring(1, result.length() - 2);
        }

        // 批量回復（bulk reply）的第一個字節是 "$", 例如:  `get foo` 返回的結果爲 `$3\r\nbar\r\n`
        if (result.startsWith("$")) {
            result = StringUtils.substringAfter(result, "\r\n");
            return StringUtils.substringBeforeLast(result, "\r\n");
        }

        // 多條批量回復（multi bulk reply）的第一個字節是 "*"， 例如: *2\r\n$3\r\nfoo\r\n$4\r\nname\r\n
        if (result.startsWith("*")) {
            result = StringUtils.substringAfter(result, "\r\n");
            String[] split = result.split("\\$\\d\r\n");
            List<String> collect = Arrays.stream(split).filter(tmpStr -> StringUtils.isNotBlank(tmpStr)).collect(Collectors.toList());
            List<String> resultList = new ArrayList<>(collect.size());
            collect.forEach(str1 -> {
                resultList.add(StringUtils.substringBeforeLast(str1, "\r\n"));
            });
            return resultList;
        }

        return "unknow result";
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.print("exceptionCaught: ");
        cause.printStackTrace(System.err);
        ctx.close();
    }

}

　　這個handler 繼承 ChannelDuplexHandler 複用處理器，也就是既能夠做爲入站處理讀取數據，也能夠做爲出站處理輸出數據。輸出和輸入的時候都是根據redis 的協議標準進行了一下消息的轉換。

3. 測試

1. 用strace 啓動redis， 監測調用指令

[root@localhost redistest]# strace -ff -o out ../redis-5.0.4/src/redis-server ../redis-5.0.4/redis.conf

2. 命令行測試結果以下

3. 從out文件查看redis 接收的命令：

[root@localhost redistest]# grep mytest ./*
./out.8384:read(8, "*3\r\n$3\r\nset\r\n$9\r\nmytestkey\r\n$11\r"..., 16384) = 46
./out.8384:read(8, "*3\r\n$3\r\nttl\r\n$9\r\nmytestkey\r\n$7\r\n"..., 16384) = 41
./out.8384:read(8, "*3\r\n$6\r\nexpire\r\n$9\r\nmytestkey\r\n$"..., 16384) = 43
./out.8384:read(8, "*2\r\n$3\r\nttl\r\n$9\r\nmytestkey\r\n", 16384) = 28
./out.8384:read(8, "*2\r\n$4\r\ntype\r\n$9\r\nmytestkey\r\n", 16384) = 29

 

注意： redis 接收到的數據必須以\r\n 結尾，不然redis 不會進行響應。

 

【當你用心寫完每一篇博客以後,你會發現它比你用代碼實現功能更有成就感!】