Customize redis connection pool

A big factory interview question ： Examine the service design and some interface requirements .

Interface II problem solving , Demand one Current limiting ; Requirement 2 Interface idempotent ; Requirement 3 Network programming timeout setting ; Requirement 4 Current limiting ;

Demand five solve HttpClient Thread safety problem , Idea customization HttpClient Connection pool .

Wrong writing ： In a concurrent scenario , Come on 1000 Requests , establish 1000 Secondary connection , Connection overhead is fatal .

We use it socket Define a httpclient, Let's demonstrate socket Thread insecurity ：

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

/**
 * Description: Socket The network is connected to redis, Need to use redis Protocol to connect 
 * Author: yz
 * Date: Created in 2021/1/6 11:14
 */
public class Connetion {
    private String host;
    private int post;
    private Socket socket;           //  Thread unsafe 
    private InputStream inputStream;
    private OutputStream outputStream;

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

    /**
     *  Determine whether the connection has been established , Determine whether the connection is initialized , Or the connection is not disconnected 
     */
    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;
    }

    /**
     *  dispatch orders 
     */
    public String sendCommand(byte[] command){
        if(isConnection()){
            try {
                //  The client writes data first 
                outputStream.write(command);
                //  Read server response 
                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 Protocol tool class 
 * 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 command 
     */
    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 The connection client contains redis The protocol and 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++) {
            //  Existing phenomenon :  The current thread reads redis Response data returned to other threads 
            pool.execute(new ClientRunalbe(client,"123"+i));
        }
    }

}

perform mian Method test ,20 Threads executing concurrently , give the result as follows ：1, Not implemented 20 Time ,2,redis The returned results are sticky , In fact, one request gets the result of another request

1, Not implemented 20 Time ： Multiple threads share one socket, When a thread is half written ,cpu The time slice is used up when scheduling , The second thread gets cup Time slice , Data will also be written , These unfinished data will also be redis Read , But it cannot be parsed ,redis Can't return response , The thread cannot read the response , Get into io Blocking .

2,redis The returned results are sticky ： The amount of data sent is relatively small , The packets sent by the thread are complete , Give to the redis Then it can be parsed into two requests , It is also normal to return data , If a thread 1 Get cup Right to use , stay read When reading data , You may read all the data in the buffer at one time , After parsing, it is found that +OK+OK This is the form

Use custom thread pool to solve socket Multithreading is unsafe , hold client Objects in the pool In the pool

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;  //  The bidirectional queue 
import java.util.concurrent.LinkedBlockingQueue;  //  One way queue 

/**
 * Description: redis Connection pool 
 * 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);
        }
    }

    /**
     *  obtain client Connect 
     */
    public BatRedisClient getClient(){
        try {
            return linkedBlockingQueue.take(); // or poll
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     *  take client Return to connection pool 
     */
    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:  Use customization redis Connection pool 
 * 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:  test redis Connection pool 
 * 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));
        }
    }
}

perform main Method test , give the result as follows , Just in time for 20 Time , No more +OK+OK There is a phenomenon

Expanding knowledge ： queue