網(wǎng)上都說jedis實(shí)例是非線程安全的，常常通過JedisPool連接池去管理實(shí)例，在多線程情況下讓每個(gè)線程有自己獨(dú)立的jedis實(shí)例，但都沒有具體說明為啥jedis實(shí)例時(shí)非線程安全的，下面詳細(xì)看一下非線程安全主要從哪個(gè)角度來看。

1. jedis類圖

2. 為什么jedis不是線程安全的

由上述類圖可知，Jedis類中有RedisInputStream和RedisOutputStream兩個(gè)屬性，而發(fā)送命令和獲取返回值都是使用這兩個(gè)成員變量，顯然，這很容易引發(fā)多線程問題。測(cè)試代碼如

public class BadConcurrentJedisTest {
 
    private static final ExecutorService pool = Executors.newFixedThreadPool(20);
 
    private static final Jedis jedis = new Jedis("192.168.58.99", 6379);
 
 
    public static void main(String[] args) {
        for(int i=0;i<20;i++){
            pool.execute(new RedisSet());
        }
    }
 
    static class RedisSet implements Runnable{
 
        @Override
        public void run() {
            while(true){
                jedis.set("hello", "world");
            }
        }
 
    }

報(bào)錯(cuò)：

Exception in thread "pool-1-thread-4" redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Socket Closed
? ? at redis.clients.jedis.Connection.connect(Connection.java:164)
? ? at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80)
? ? at redis.clients.jedis.Connection.sendCommand(Connection.java:100)
? ? at redis.clients.jedis.BinaryClient.set(BinaryClient.java:97)
? ? at redis.clients.jedis.Client.set(Client.java:32)
? ? at redis.clients.jedis.Jedis.set(Jedis.java:68)
? ? at threadsafe.BadConcurrentJedisTest$RedisSet.run(BadConcurrentJedisTest.java:26)
? ? at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
? ? at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
? ? at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.SocketException: Socket Closed
? ? at java.net.AbstractPlainSocketImpl.setOption(AbstractPlainSocketImpl.java:212)
? ? at java.net.Socket.setKeepAlive(Socket.java:1310)
? ? at redis.clients.jedis.Connection.connect(Connection.java:149)
? ? ... 9 more
redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Socket Closed
? ? at redis.clients.jedis.Connection.connect(Connection.java:164)
? ? at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80)
? ? at redis.clients.jedis.Connection.sendCommand(Connection.java:100)
? ? at redis.clients.jedis.BinaryClient.set(BinaryClient.java:97)
? ? at redis.clients.jedis.Client.set(Client.java:32)
? ? at redis.clients.jedis.Jedis.set(Jedis.java:68)
? ? at threadsafe.BadConcurrentJedisTest$RedisSet.run(BadConcurrentJedisTest.java:26)
? ? at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
? ? at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
? ? at java.lang.Thread.run(Thread.java:745)
.......

主要錯(cuò)誤：

ava.net.SocketException: Socket closed
java.net.SocketException: Socket is not connected　　

2.1 共享socket引起的異常

為什么會(huì)出現(xiàn)這2個(gè)錯(cuò)誤呢？ 我們可以很容易的通過堆棧信息定位到redis.clients.jedis.Connection的connect方法

public void connect() {
    if (!isConnected()) {
      try {
        socket = new Socket();
        // ->@wjw_add
        socket.setReuseAddress(true);
        socket.setKeepAlive(true); // Will monitor the TCP connection is
        // valid
        socket.setTcpNoDelay(true); // Socket buffer Whetherclosed, to
        // ensure timely delivery of data
        socket.setSoLinger(true, 0); // Control calls close () method,
        // the underlying socket is closed
        // immediately
        // <-@wjw_add
 
        socket.connect(new InetSocketAddress(host, port), connectionTimeout);
        socket.setSoTimeout(soTimeout);
 
        if (ssl) {
          if (null == sslSocketFactory) {
            sslSocketFactory = (SSLSocketFactory)SSLSocketFactory.getDefault();
          }
          socket = (SSLSocket) sslSocketFactory.createSocket(socket, host, port, true);
          if (null != sslParameters) {
            ((SSLSocket) socket).setSSLParameters(sslParameters);
          }
          if ((null != hostnameVerifier) &&
              (!hostnameVerifier.verify(host, ((SSLSocket) socket).getSession()))) {
            String message = String.format(
                "The connection to '%s' failed ssl/tls hostname verification.", host);
            throw new JedisConnectionException(message);
          }
        }
 
        outputStream = new RedisOutputStream(socket.getOutputStream());
        inputStream = new RedisInputStream(socket.getInputStream());
      } catch (IOException ex) {
        broken = true;
        throw new JedisConnectionException(ex);
      }
    }
  }

?jedis在執(zhí)行每一個(gè)命令之前都會(huì)先執(zhí)行connect方法，socket是一個(gè)共享變量，在多線程的情況下可能存在：線程1執(zhí)行到了

outputStream = new RedisOutputStream(socket.getOutputStream());
inputStream = new RedisInputStream(socket.getInputStream());　　

線程2執(zhí)行到了：

socket = new Socket();
//線程2
socket.connect(new InetSocketAddress(host, port), connectionTimeout);　　

因?yàn)榫€程2重新初始化了socket但是還沒有執(zhí)行connect，所以線程1執(zhí)行socket.getOutputStream()或者socket.getInputStream()就會(huì)拋出java.net.SocketException: Socket is not connected。java.net.SocketException: Socket closed是因?yàn)閟ocket異常導(dǎo)致共享變量socket關(guān)閉了引起的。

2.2 共享數(shù)據(jù)流引起的異常

上面是因?yàn)槎鄠€(gè)線程共享jedis引起的socket異常。除了socket連接引起的異常之外，還有共享數(shù)據(jù)流引起的異常。下面就看一下，因?yàn)楣蚕韏edis實(shí)例引起的共享數(shù)據(jù)流錯(cuò)誤問題。

為了避免多線程連接的時(shí)候引起的錯(cuò)誤，我們?cè)诔跏蓟臅r(shí)候就先執(zhí)行一下connect操作：

public class BadConcurrentJedisTest1 {
 
    private static final ExecutorService pool = Executors.newCachedThreadPool();
 
    private static final Jedis jedis = new Jedis("192.168.58.99", 6379);
 
    static{
        jedis.connect();
    }
 
    public static void main(String[] args) {
        for(int i=0;i<20;i++){
            pool.execute(new RedisTest());
        }
 
    }
 
    static class RedisTest implements Runnable{
 
        @Override
        public void run() {
            while(true){
                jedis.set("hello", "world");
            }
        }
 
    }
 
}　

報(bào)錯(cuò)：（每次報(bào)的錯(cuò)可能不完全一樣）

Exception in thread "pool-1-thread-7" Exception in thread "pool-1-thread-1" redis.clients.jedis.exceptions.JedisDataException: ERR Protocol error: invalid multibulk length
? ? at redis.clients.jedis.Protocol.processError(Protocol.java:123)
? ? at redis.clients.jedis.Protocol.process(Protocol.java:157)
? ? at redis.clients.jedis.Protocol.read(Protocol.java:211)
? ? at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:297)
? ? at redis.clients.jedis.Connection.getStatusCodeReply(Connection.java:196)
? ? at redis.clients.jedis.Jedis.set(Jedis.java:69)
? ? at threadsafe.BadConcurrentJedisTest1$RedisTest.run(BadConcurrentJedisTest1.java:30)
? ? at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
? ? at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
? ? at java.lang.Thread.run(Thread.java:745)
Exception in thread "pool-1-thread-4" redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Broken pipe (Write failed)
? ? at redis.clients.jedis.Connection.flush(Connection.java:291)
? ? at redis.clients.jedis.Connection.getStatusCodeReply(Connection.java:194)
? ? at redis.clients.jedis.Jedis.set(Jedis.java:69)
? ? at threadsafe.BadConcurrentJedisTest1$RedisTest.run(BadConcurrentJedisTest1.java:30)
? ? at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
? ? at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
? ? at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.SocketException: Broken pipe (Write failed)

Protocol error: invalid multibulk lengt是因?yàn)槎嗑€程通過RedisInputStream和RedisOutputStream讀寫緩沖區(qū)的時(shí)候引起的問題造成的數(shù)據(jù)問題不滿足RESP協(xié)議引起的。舉個(gè)簡(jiǎn)單的例子，例如多個(gè)線程執(zhí)行命令,線程1執(zhí)行 set hello world命令。本來應(yīng)該發(fā)送：

*3\r\n$3\r\nSET\r\n$5\r\nhello\r\n$5\r\nworld\r\n

但是線程執(zhí)行寫到

*3\r\n$3\r\nSET\r\n$5\r\nhello\r\n

然后被掛起了，線程2執(zhí)行了寫操作寫入了' '，然后線程1繼續(xù)執(zhí)行，最后發(fā)送到redis服務(wù)器端的數(shù)據(jù)可能就是：

*3\r\n$3\r\nSET\r\n$5\r\nhello\r\n' '$5\r\nworld\r\n

至于java.net.SocketException: Connection reset或ReadTimeout錯(cuò)誤，是因?yàn)閞edis服務(wù)器接受到錯(cuò)誤的命令，執(zhí)行了socket.close這樣的操作，關(guān)閉了連接。服務(wù)器會(huì)返回復(fù)位標(biāo)志"RST"，但是客戶端還在繼續(xù)執(zhí)行讀寫數(shù)據(jù)操作。

3、jedis多線程操作

jedis本身不是多線程安全的，這并不是jedis的bug，而是jedis的設(shè)計(jì)與redis本身就是單線程相關(guān)，jedis實(shí)例抽象的是發(fā)送命令相關(guān)，一個(gè)jedis實(shí)例使用一個(gè)線程與使用100個(gè)線程去發(fā)送命令沒有本質(zhì)上的區(qū)別，所以沒必要設(shè)置為線程安全的。但是如果需要用多線程方式訪問redis服務(wù)器怎么做呢？那就使用多個(gè)jedis實(shí)例，每個(gè)線程對(duì)應(yīng)一個(gè)jedis實(shí)例，而不是一個(gè)jedis實(shí)例多個(gè)線程共享。一個(gè)jedis關(guān)聯(lián)一個(gè)Client，相當(dāng)于一個(gè)客戶端，Client繼承了Connection，Connection維護(hù)了Socket連接，對(duì)于Socket這種昂貴的連接，一般都會(huì)做池化，jedis提供了JedisPool。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
 
public class JedisPoolTest {
 
    private static final ExecutorService pool = Executors.newCachedThreadPool();
 
    private static final CountDownLatch latch = new CountDownLatch(20);
 
    private static final JedisPool jPool = new JedisPool("192.168.58.99", 6379);
 
    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        for(int i=0;i<20;i++){
            pool.execute(new RedisTest());
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() - start);
        pool.shutdownNow();
 
    }
 
    static class RedisTest implements Runnable{
 
        @Override
        public void run() {
            Jedis jedis = jPool.getResource();
            int i = 1000;
            try{
                while(i-->0){
                        jedis.set("hello", "world");
                }
            }finally{
                jedis.close();
                latch.countDown();
            }
        }
 
    }
 
}