网上都说jedis实例是非线程安全的，常常通过JedisPool连接池去管理实例，在多线程情况下让每个线程有自己独立的jedis实例，但都没有具体说明为啥jedis实例时非线程安全的，下面详细看一下非线程安全主要从哪个角度来看。
1. jedis类图
2. 为什么jedis不是线程安全的？
由上述类图可知，Jedis类中有RedisInputStream和RedisOutputStream两个属性，而发送命令和获取返回值都是使用这两个成员变量，显然，这很容易引发多线程问题。测试代码如
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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"
);
}
}
}
|
报错：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
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
)
.......
|
主要错误：
|
1
2
|
ava.net.SocketException: Socket closed
java.net.SocketException: Socket is not connected
|
2.1 共享socket引起的异常
为什么会出现这2个错误呢？ 我们可以很容易的通过堆栈信息定位到redis.clients.jedis.Connection的connect方法
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
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在执行每一个命令之前都会先执行connect方法，socket是一个共享变量，在多线程的情况下可能存在：线程1执行到了
|
1
2
|
outputStream =
new
RedisOutputStream(socket.getOutputStream());
inputStream =
new
RedisInputStream(socket.getInputStream());
|
线程2执行到了：
|
1
2
3
|
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 connected。java.net.SocketException: Socket closed是因为socket异常导致共享变量socket关闭了引起的。
2.2 共享数据流引起的异常
上面是因为多个线程共享jedis引起的socket异常。除了socket连接引起的异常之外，还有共享数据流引起的异常。下面就看一下，因为共享jedis实例引起的共享数据流错误问题。
为了避免多线程连接的时候引起的错误，我们在初始化的时候就先执行一下connect操作：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
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"
);
}
}
}
}
|
报错：（每次报的错可能不完全一样）
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
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)
|
送命令相关，一个jedis实例使用一个线程与使用100个线程去发送命令没有本质上的区别，所以没必要设置为线程安全的。但是如果需要用多线程方式访问redis服务器怎么做呢？那就使用多个jedis实例，每个线程对应一个jedis实例，而不是一个jedis实例多个线程共享。一个jedis关联一个Client，相当于一个客户端，Client继承了Connection，Connection维护了Socket连接，对于Socket这种昂贵的连接，一般都会做池化，jedis提供了JedisPool
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
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();
}
}
}
}
|