https://blog.csdn.net/weixin_40237626/article/details/117736457

前言

网上都说jedis实例是非线程安全的,常常通过JedisPool连接池去管理实例,在多线程情况下让每个线程有自己独立的jedis实例,但都没有具体说明为啥jedis实例时非线程安全的,下面详细看一下非线程安全主要从哪个角度来看。

 jedis类图

为什么jedis不是线程安全的?

1.共享socket引起的异常

由上述类图可知,Jedis类中有RedisInputStreamRedisOutputStream两个属性,而发送命令和获取返回值都是使用这两个成员变量,显然,这很容易引发多线程问题。测试代码如

  1. public class BadConcurrentJedisTest {
  2.     private static final ExecutorService pool = Executors.newFixedThreadPool(20);
  3.     private static final Jedis jedis = new Jedis("192.168.58.99", 6379);
  4.     public static void main(String[] args) {
  5.         for(int i=0;i<20;i++){
  6.             pool.execute(new RedisSet());
  7.         }
  8.     }
  9.     static class RedisSet implements Runnable {
  10.         @Override
  11.         public void run() {
  12.             while(true){
  13.                 jedis.set("hello", "world");
  14.             }
  15.         }
  16.     }

报错:

  1. Exception in thread "pool-1-thread-4" redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Socket Closed
  2.     at redis.clients.jedis.Connection.connect(Connection.java:164)
  3.     at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80)
  4.     at redis.clients.jedis.Connection.sendCommand(Connection.java:100)
  5.     at redis.clients.jedis.BinaryClient.set(BinaryClient.java:97)
  6.     at redis.clients.jedis.Client.set(Client.java:32)
  7.     at redis.clients.jedis.Jedis.set(Jedis.java:68)
  8.     at threadsafe.BadConcurrentJedisTest$RedisSet.run(BadConcurrentJedisTest.java:26)
  9.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  10.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  11.     at java.lang.Thread.run(Thread.java:745)
  12. Caused by: java.net.SocketException: Socket Closed
  13.     at java.net.AbstractPlainSocketImpl.setOption(AbstractPlainSocketImpl.java:212)
  14.     at java.net.Socket.setKeepAlive(Socket.java:1310)
  15.     at redis.clients.jedis.Connection.connect(Connection.java:149)
  16.     ... 9 more
  17. redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Socket Closed
  18.     at redis.clients.jedis.Connection.connect(Connection.java:164)
  19.     at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80)
  20.     at redis.clients.jedis.Connection.sendCommand(Connection.java:100)
  21.     at redis.clients.jedis.BinaryClient.set(BinaryClient.java:97)
  22.     at redis.clients.jedis.Client.set(Client.java:32)
  23.     at redis.clients.jedis.Jedis.set(Jedis.java:68)
  24.     at threadsafe.BadConcurrentJedisTest$RedisSet.run(BadConcurrentJedisTest.java:26)
  25.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  26.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  27.     at java.lang.Thread.run(Thread.java:745)
  28. .......

主要错误:

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

为什么会出现这2个错误呢? 我们可以很容易的通过堆栈信息定位到redis.clients.jedis.Connectionconnect方法

  1. public void connect() {
  2.     if (!isConnected()) {
  3.       try {
  4.         socket = new Socket();
  5.         // ->@wjw_add
  6.         socket.setReuseAddress(true);
  7.         socket.setKeepAlive(true); // Will monitor the TCP connection is
  8.         // valid
  9.         socket.setTcpNoDelay(true); // Socket buffer Whetherclosed, to
  10.         // ensure timely delivery of data
  11.         socket.setSoLinger(true, 0); // Control calls close () method,
  12.         // the underlying socket is closed
  13.         // immediately
  14.         // <-@wjw_add
  15.         socket.connect(new InetSocketAddress(host, port), connectionTimeout);
  16.         socket.setSoTimeout(soTimeout);
  17.         if (ssl) {
  18.           if (null == sslSocketFactory) {
  19.             sslSocketFactory = (SSLSocketFactory)SSLSocketFactory.getDefault();
  20.           }
  21.           socket = (SSLSocket) sslSocketFactory.createSocket(socket, host, port, true);
  22.           if (null != sslParameters) {
  23.             ((SSLSocket) socket).setSSLParameters(sslParameters);
  24.           }
  25.           if ((null != hostnameVerifier) &&
  26.               (!hostnameVerifier.verify(host, ((SSLSocket) socket).getSession()))) {
  27.             String message = String.format(
  28.                 "The connection to '%s' failed ssl/tls hostname verification.", host);
  29.             throw new JedisConnectionException(message);
  30.           }
  31.         }
  32.         outputStream = new RedisOutputStream(socket.getOutputStream());
  33.         inputStream = new RedisInputStream(socket.getInputStream());
  34.       } catch (IOException ex) {
  35.         broken = true;
  36.         throw new JedisConnectionException(ex);
  37.       }
  38.     }
  39.   }

jedis在执行每一个命令之前都会先执行connect方法,socket是一个共享变量,在多线程的情况下可能存在:

线程1执行到了:

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

线程2执行到了:

  1. socket = new Socket();
  2. socket.connect(new InetSocketAddress(host, port), connectionTimeout);

因为线程2重新初始化了socket但是还没有执行connect,所以线程1执行socket.getOutputStream()或者socket.getInputStream()就会抛出java.net.SocketException: Socket is not connectedjava.net.SocketException: Socket closed是因为socket异常导致共享变量socket关闭了引起的。

2 .共享数据流引起的异常

上面是因为多个线程共享jedis引起的socket异常。除了socket连接引起的异常之外,还有共享数据流引起的异常。下面就看一下,因为共享jedis实例引起的共享数据流错误问题。为了避免多线程连接的时候引起的错误,我们在初始化的时候就先执行一下connect操作:

  1. public class BadConcurrentJedisTest1 {
  2.     private static final ExecutorService pool = Executors.newCachedThreadPool();
  3.     private static final Jedis jedis = new Jedis("192.168.58.99", 6379);
  4.     static{
  5.         jedis.connect();
  6.     }
  7.     public static void main(String[] args) {
  8.         for(int i=0;i<20;i++){
  9.             pool.execute(new RedisTest());
  10.         }
  11.     }
  12.     static class RedisTest implements Runnable{
  13.         @Override
  14.         public void run() {
  15.             while(true){
  16.                 jedis.set("hello", "world");
  17.             }
  18.         }
  19.     }
  20. }

报错:(每次报的错可能不完全一样)

  1. 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
  2.     at redis.clients.jedis.Protocol.processError(Protocol.java:123)
  3.     at redis.clients.jedis.Protocol.process(Protocol.java:157)
  4.     at redis.clients.jedis.Protocol.read(Protocol.java:211)
  5.     at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:297)
  6.     at redis.clients.jedis.Connection.getStatusCodeReply(Connection.java:196)
  7.     at redis.clients.jedis.Jedis.set(Jedis.java:69)
  8.     at threadsafe.BadConcurrentJedisTest1$RedisTest.run(BadConcurrentJedisTest1.java:30)
  9.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  10.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  11.     at java.lang.Thread.run(Thread.java:745)
  12. Exception in thread "pool-1-thread-4" redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Broken pipe (Write failed)
  13.     at redis.clients.jedis.Connection.flush(Connection.java:291)
  14.     at redis.clients.jedis.Connection.getStatusCodeReply(Connection.java:194)
  15.     at redis.clients.jedis.Jedis.set(Jedis.java:69)
  16.     at threadsafe.BadConcurrentJedisTest1$RedisTest.run(BadConcurrentJedisTest1.java:30)
  17.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  18.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  19.     at java.lang.Thread.run(Thread.java:745)
  20. Caused by: java.net.SocketException: Broken pipe (Write failed)

 Protocol error: invalid multibulk lengt是因为多线程通过RedisInputStreamRedisOutputStream读写缓冲区的时候引起的问题造成的数据问题不满足RESP协议引起的。举个简单的例子,例如多个线程执行命令,线程1执行 set hello world命令。本来应该发送:

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

但是线程执行写到

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

然后被挂起了,线程2执行了写操作写入了' ',然后线程1继续执行,最后发送到redis服务器端的数据可能就是:

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

至于java.net.SocketException: Connection resetReadTimeout错误,是因为redis服务器接受到错误的命令,执行了socket.close这样的操作,关闭了连接。服务器会返回复位标志"RST",但是客户端还在继续执行读写数据操作。

3.jedis多线程操作

jedis本身不是多线程安全的,这并不是jedisbug,而是jedis的设计与redis本身就是单线程相关,jedis实例抽象的是发送命令相关,一个jedis实例使用一个线程与使用100个线程去发送命令没有本质上的区别,所以没必要设置为线程安全的。但是如果需要用多线程方式访问redis服务器怎么做呢?那就使用多个jedis实例,每个线程对应一个jedis实例,而不是一个jedis实例多个线程共享。一个jedis关联一个Client,相当于一个客户端,Client继承了ConnectionConnection维护了Socket连接,对于Socket这种昂贵的连接,一般都会做池化,jedis提供了JedisPool

  1. import java.util.concurrent.CountDownLatch;
  2. import java.util.concurrent.ExecutorService;
  3. import java.util.concurrent.Executors;
  4. import redis.clients.jedis.Jedis;
  5. import redis.clients.jedis.JedisPool;
  6. public class JedisPoolTest {
  7.     private static final ExecutorService pool = Executors.newCachedThreadPool();
  8.     private static final CountDownLatch latch = new CountDownLatch(20);
  9.     private static final JedisPool jPool = new JedisPool("192.168.58.99", 6379);
  10.     public static void main(String[] args) {
  11.         long start = System.currentTimeMillis();
  12.         for(int i=0;i<20;i++){
  13.             pool.execute(new RedisTest());
  14.         }
  15.         try {
  16.             latch.await();
  17.         } catch (InterruptedException e) {
  18.             e.printStackTrace();
  19.         }
  20.         System.out.println(System.currentTimeMillis() - start);
  21.         pool.shutdownNow();
  22.     }
  23.     static class RedisTest implements Runnable{
  24.         @Override
  25.         public void run() {
  26.             Jedis jedis = jPool.getResource();
  27.             int i = 1000;
  28.             try{
  29.                 while(i-->0){
  30.                         jedis.set("hello", "world");
  31.                 }
  32.             }finally{
  33.                 jedis.close();
  34.                 latch.countDown();
  35.             }
  36.         }
  37.     }
  38. }

Logo

华为开发者空间,是为全球开发者打造的专属开发空间,汇聚了华为优质开发资源及工具,致力于让每一位开发者拥有一台云主机,基于华为根生态开发、创新。

更多推荐