使用Jedis面臨的非線程安全問題詳解
網(wǎng)上都說jedis實例是非線程安全的,常常通過JedisPool連接池去管理實例,在多線程情況下讓每個線程有自己獨立的jedis實例,但都沒有具體說明為啥jedis實例時非線程安全的,下面詳細看一下非線程安全主要從哪個角度來看。
1. jedis類圖

2. 為什么jedis不是線程安全的
由上述類圖可知,Jedis類中有RedisInputStream和RedisOutputStream兩個屬性,而發(fā)送命令和獲取返回值都是使用這兩個成員變量,顯然,這很容易引發(fā)多線程問題。測試代碼如
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");
}
}
}報錯:
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)
.......
主要錯誤:
ava.net.SocketException: Socket closed
java.net.SocketException: Socket is not connected
2.1 共享socket引起的異常
為什么會出現(xiàn)這2個錯誤呢? 我們可以很容易的通過堆棧信息定位到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í)行每一個命令之前都會先執(zhí)行connect方法,socket是一個共享變量,在多線程的情況下可能存在:線程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);
因為線程2重新初始化了socket但是還沒有執(zhí)行connect,所以線程1執(zhí)行socket.getOutputStream()或者socket.getInputStream()就會拋出java.net.SocketException: Socket is not connected。java.net.SocketException: Socket closed是因為socket異常導致共享變量socket關閉了引起的。
2.2 共享數(shù)據(jù)流引起的異常
上面是因為多個線程共享jedis引起的socket異常。除了socket連接引起的異常之外,還有共享數(shù)據(jù)流引起的異常。下面就看一下,因為共享jedis實例引起的共享數(shù)據(jù)流錯誤問題。
為了避免多線程連接的時候引起的錯誤,我們在初始化的時候就先執(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");
}
}
}
} 報錯:(每次報的錯可能不完全一樣)
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是因為多線程通過RedisInputStream和RedisOutputStream讀寫緩沖區(qū)的時候引起的問題造成的數(shù)據(jù)問題不滿足RESP協(xié)議引起的。舉個簡單的例子,例如多個線程執(zhí)行命令,線程1執(zhí)行 set hello world命令。本來應該發(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服務器端的數(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錯誤,是因為redis服務器接受到錯誤的命令,執(zhí)行了socket.close這樣的操作,關閉了連接。服務器會返回復位標志"RST",但是客戶端還在繼續(xù)執(zhí)行讀寫數(shù)據(jù)操作。
3、jedis多線程操作
jedis本身不是多線程安全的,這并不是jedis的bug,而是jedis的設計與redis本身就是單線程相關,jedis實例抽象的是發(fā)送命令相關,一個jedis實例使用一個線程與使用100個線程去發(fā)送命令沒有本質(zhì)上的區(qū)別,所以沒必要設置為線程安全的。但是如果需要用多線程方式訪問redis服務器怎么做呢?那就使用多個jedis實例,每個線程對應一個jedis實例,而不是一個jedis實例多個線程共享。一個jedis關聯(lián)一個Client,相當于一個客戶端,Client繼承了Connection,Connection維護了Socket連接,對于Socket這種昂貴的連接,一般都會做池化,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();
}
}
}
}到此這篇關于使用Jedis面臨的非線程安全問題詳解的文章就介紹到這了,更多相關Jedis非線程安全問題內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!

