自定義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現象了

擴展知識：隊列