java手写redis客户端工具

Posted 2021-12-04 今夜月色很美

需要实现无论什么命令都可以执行到redis集群里面。
用jedis的话不同的命令需要用不同的方式执行，所以就采用了socket与redis集群通信。手工解决了redis key在不同服务器中进行切换的操作。

redis socket客户端SocketConnectRedisClient

import org.apache.commons.lang3.StringUtils;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;

/**
 * redis socket客户端
 */
public class SocketConnectRedisClient implements Closeable {

    private String host;
    private int port;
    private Socket socket = null;

    /**
     * 创建socket连接
     * @param host
     * @param port
     */
    public SocketConnectRedisClient(String host, int port) throws Exception {
        connect(host, port);
    }

    /**
     * 关闭旧的socket，重新创建socket
     * @param host
     * @param port
     */
    public void changeNode(String host, int port) throws Exception {
        this.close();
        connect(host, port);
    }

    /**
     * socket连接redis服务器
     * @param host
     * @param port
     * @throws IOException
     */
    private void connect(String host, int port) throws IOException {
        this.host = host;
        this.port = port;
        socket = new Socket();
        socket.setSendBufferSize(102400);
        socket.setReceiveBufferSize(102400);
        socket.connect(new InetSocketAddress(this.host, this.port),15 * 1000);
    }

    /**
     * 执行redis命令
     * @param command 待执行redis cmd
     * @return
     * @throws IOException
     */
    public String executeCommand(String command) throws Exception {
        return this.execute(command.split(" "));
    }

    /**
     * 执行redis命令
     * @param args
     * @return
     * @throws IOException
     */
    public String execute(String... args) throws Exception {
        if (socket == null || args == null || args.length <= 0) {
            return null;
        }
        StringBuilder request = new StringBuilder();
        request.append("*" + args.length).append("\\r\\n");//参数的数量

        for (int i = 0; i < args.length; i++) {
            request.append("$" + args[i].getBytes("utf8").length).append("\\r\\n");//参数的长度
            request.append(args[i]).append("\\r\\n");//参数的内容
        }

        socket.getOutputStream().write(request.toString().getBytes());
        socket.getOutputStream().flush();

        StringBuilder reply = new StringBuilder();
        int bufSize = 1024;
        while (true) {
            byte[] buf = new byte[bufSize];
            int len = socket.getInputStream().read(buf);
            if (len < 0) {
                break;
            }
            String str = new String(buf, 0, len);
            reply.append(str);
            if (str.endsWith("\\r\\n")) {
                break;
            }
        }
        String response = reply.toString();
        if (StringUtils.isNotBlank(response) && response.startsWith("-MOVED ")){
            return processResultMoved(response, args);
        }
        return response;
    }

    /**
     * 如果redis返回结果指向其他node，解析response返回的node地址信息，重新创建socket
     * @param response
     * @param args
     * @return
     * @throws IOException
     */
    private String processResultMoved(String response, String[] args) throws Exception {
        String[] responseArr = response.split(" ");
        String[] addressArr = responseArr[2].split(":");
        String ip = addressArr[0];
        String port = addressArr[1].replaceAll("\\r|\\n", "");
        this.changeNode(ip, Integer.valueOf(port));
        return this.execute(args);
    }

    @Override
    public void close() {
        try {
            if (socket != null) {
                socket.shutdownOutput();
                socket.close();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}

socket连接redis工具类SocketConnectRedisUtil

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * socket连接redis工具类
 */
public class SocketConnectRedisUtil {
    private static final Logger logger = LoggerFactory.getLogger(SocketConnectRedisUtil.class);

    private  static BlockingQueue<String> redisClusters = new LinkedBlockingDeque<>(100);
    /**
     * 执行redis命令
     * @param addresses redis node配置列表
     * @param cmd 待执行redis命令
     * @param recursive 如果配置列表中的redis服务地址不可用，最多尝试recursive次，防止等待时间过长
     * @return
     */
    public static String execCmd(List<String> addresses, String cmd, int recursive){
        SocketConnectRedisClient client = null;
        String result = null;
        String address = null;
        try {
            address = getCluster(addresses);
            if (StringUtils.isBlank(address)){
                logger.error("redis配置信息不正确", addresses);
                return "redis配置信息不正确";
            }
            String[] arr = address.split(":");
            client = new SocketConnectRedisClient(arr[0], Integer.valueOf(arr[1]));
            result = client.executeCommand(cmd);
        } catch (ConnectException | SocketTimeoutException e) {
            logger.error("redis服务器{}，执行redis命令异常", address, e);
            result = processException(addresses, cmd, recursive);
        } catch (Exception e) {
            logger.error("执行redis命令异常", e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return result;
    }

    /**
     * 连接异常重试
     * @param addresses
     * @param cmd
     * @param recursive
     * @return
     */
    private static String processException(List<String> addresses, String cmd, int recursive) {
        removeCluster();
        if (recursive > 0){
            //换下一个redis地址尝试执行redis cmd
            return execCmd(addresses, cmd, --recursive);
        } else {
            return "redis服务异常";
        }
    }

    private static synchronized String getCluster(List<String> addresses) {
        if (redisClusters.size() == 0){
            for (String address : addresses) {
                redisClusters.offer(address);
            }
            addresses.clear();
        }
        return redisClusters.peek();
    }

    private static synchronized void removeCluster(){
        redisClusters.poll();
    }
}

测试代码

public static void main(String[] args) {
        List<String> nodes = new ArrayList<>();
        nodes.add("ip1:port1");
        nodes.add("ip2:port2");
        //如果redis地址连接失败，默认重试一次
        String result = SocketConnectRedisUtil.execCmd(nodes, "get 1", 1);
        System.out.println(result);
    }