自定义Redis连接池

一道大厂面试题：考察服务设计和一些接口要求。

接口二解题，要求一 限流； 要求二 接口幂等；要求三 网络编程超时设置；要求四 限流；

要求五 解决HttpClient线程安全问题,思路自定义HttpClient连接池。

错误写法：在并发场景下，来1000次请求，建立1000次连接，连接开销很致命。

我们用socket定义一个httpclient,来演示一下socket线程不安全现象：

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

/**
 * Description: Socket网络连接到redis,需要使用redis协议进行连接
 * Author: yz
 * Date: Created in 2021/1/6 11:14
 */
public class Connetion {
    private String host;
    private int post;
    private Socket socket;           // 线程不安全
    private InputStream inputStream;
    private OutputStream outputStream;

    public Connetion(String host, int post) {
        this.host = host;
        this.post = post;
    }

    /**
     * 判断连接是否已经建立,判断连接是不是初始化好了，或者连接没有断开
     */
    public boolean isConnection(){
        if(socket !=null && inputStream !=null && socket.isClosed()){
            return true;
        }
        try {
            socket = new Socket(host,post);
            inputStream = socket.getInputStream();
            outputStream = socket.getOutputStream();
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 发送命令
     */
    public String sendCommand(byte[] command){
        if(isConnection()){
            try {
                // 客户端先写数据
                outputStream.write(command);
                // 读取服务端响应
                byte[] res = new byte[1024];
                int length = 0;
                while ((length=inputStream.read(res))>0){
                    return new String(res,0,length);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

}

/**
 * Description: redis协议工具类
 * Author: yz
 * Date: Created in 2021/1/6 11:41
 */
public class RedisProtocolUtils {
    public static final String DOLIER = "$";
    public static final String ALLERTSTIC = "*";
    public static final String CRLE = "\r\n";

    public static byte[] buildRespByte(Command command,byte[]... bytes){
        StringBuilder sb = new StringBuilder();
        sb.append(ALLERTSTIC).append(bytes.length+1).append(CRLE);
        sb.append(DOLIER).append(command.name().length()).append(CRLE);
        sb.append(command.name()).append(CRLE);
        for (byte[] arg : bytes) {
            sb.append(DOLIER).append(arg.length).append(CRLE);
            sb.append(new String(arg)).append(CRLE);
        }
        return sb.toString().getBytes();
    }

    /**
     * redis set，get命令
     */
    public enum Command{
        SET,GET
    }

}

/**
 * Description:
 * Author: yz
 * Date: Created in 2021/1/6 15:01
 */
public class ClientRunalbe implements Runnable {

    private BatRedisClient client;
    private String value;

    public ClientRunalbe(BatRedisClient client, String value) {
        this.client = client;
        this.value = value;
    }

    @Override
    public void run() {
        client.set("ant",value);
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Description: redis连接客户端含redis协议和socket
 * Author: yz
 * Date: Created in 2021/1/6 11:53
 */
public class BatRedisClient {
    private Connetion connetion;
    public BatRedisClient(String host,int port){
        connetion = new Connetion(host,port);
    }

    public String set(String key,String value){
        String result = connetion.sendCommand(
                RedisProtocolUtils.buildRespByte(RedisProtocolUtils.Command.SET,key.getBytes(),value.getBytes()));
        System.out.println("Thread name:"+Thread.currentThread().getName()
        +"[result]:"+result.replace("\r\n","")+"[value]:"+value);
        return result;
    }

    public String get(String key){
        String result = connetion.sendCommand(
                RedisProtocolUtils.buildRespByte(RedisProtocolUtils.Command.GET,key.getBytes(),key.getBytes()));
        return result;
    }

    public static void main(String[] args) {
        BatRedisClient client = new BatRedisClient("localhost",6379);
        ExecutorService pool = Executors.newCachedThreadPool();
        for (int i = 0; i < 20; i++) {
            // 存在现象: 当前线程读取到redis返回给其他线程的响应数据
            pool.execute(new ClientRunalbe(client,"123"+i));
        }
    }

}

执行mian方法测试，20个线程并发执行，结果如下：1，并未执行20次，2，redis返回的结果有粘在一起的，其实是一个请求获取到另一个请求的结果了

1，并未执行20次：多个线程共用一个socket，当一个线程写一半的时候，cpu调度的时候时间片用完了，第二个线程获取到cup时间片，也会写入数据，这些没有写完的数据也会被redis读取到，但是不能被解析，redis不能返回响应，线程读取不到响应，进入io阻塞。

2，redis返回的结果有粘在一起的：发送的时候数据量比较小，线程发送的数据包是完整的，给到redis之后是可以解析成两个请求的，也是正常返回数据，假如线程1获取到cup使用权，在read读取数据的时候，可能会一次性把缓冲区所有的数据都读过来，解析之后发现是+OK+OK这种形式了

使用自定义线程池方式解决socket多线程不安全问题，把client对象放到pool池里面

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;  // 双向队列
import java.util.concurrent.LinkedBlockingQueue;  // 单向队列

/**
 * Description: redis连接池
 * Author: yz
 * Date: Created in 2021/1/6 15:19
 */
public class RedisClientPool {

    private List<BatRedisClient> allObject;
    private LinkedBlockingQueue<BatRedisClient> linkedBlockingQueue;

    public RedisClientPool(String host,int port,int connectionCount){
        allObject = new ArrayList<>();
        this.linkedBlockingQueue = new LinkedBlockingQueue<>(10);
        for (int i = 0; i < connectionCount; i++) {
            BatRedisClient client = new BatRedisClient(host,port);
            linkedBlockingQueue.add(client);
            allObject.add(client);
        }
    }

    /**
     * 获取client连接
     */
    public BatRedisClient getClient(){
        try {
            return linkedBlockingQueue.take(); // or poll
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 将client归还到连接池
     */
    public void returnClient(BatRedisClient client){
        if(client == null){
            return;
        }
        if(!allObject.contains(client)){
            throw new IllegalStateException(
                    "Returned object not currently part of this pool");
        }
        try {
            linkedBlockingQueue.put(client);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

/**
 * Description: 使用自定义redis连接池
 * Author: yz
 * Date: Created in 2021/1/6 15:28
 */
public class ClientPoolRunable implements Runnable {
    private RedisClientPool pool;
    private String value;

    public ClientPoolRunable(RedisClientPool pool, String value) {
        this.pool = pool;
        this.value = value;
    }

    @Override
    public void run() {
        BatRedisClient client = pool.getClient();
        client.set("ant",value);
        pool.returnClient(client);
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Description: 测试redis连接池
 * Author: yz
 * Date: Created in 2021/1/6 15:30
 */
public class Main {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newCachedThreadPool();
        RedisClientPool redisClientPool = new RedisClientPool("localhost",6379,10);
        for (int i = 0; i < 20; i++) {
            pool.execute(new ClientPoolRunable(redisClientPool,"123"+i));
        }
    }
}

执行main方法测试，结果如下，正好执行20次，也没有再出现+OK+OK现象了

扩展知识：队列