Jedis高并发下操作redis报错：java.net.SocketException: Connection reset by peer: socket write error

项目中读写redis用的是spring-data-redis，因为方便嘛，但是有一天我想用Jedis来操作redis，于是我就从RedisTemplate中拿到了Jedis，也确实拿到了，于是就有了这个问题。开始自测时候没有发现问题，因为都是单线程环境下，没有覆盖到高并发情况，但是上生产后发现杯具了。

代码如下：

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import javax.annotation.PostConstruct;


@Slf4j
@Component
public class JedisServiceOperate {

    private final RedisTemplate redisTemplate;

    private static final String NX = "NX";

    private static final String PX = "PX";

    private static final String XX = "XX";

    private Jedis jedis;

    JedisServiceOperate(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    
    public void set(byte[] key, byte[] value) {
        log.info("set key={} result={}", key, jedis.set(key, value));
    }

    
    public void set(String key, String value) {
        jedis.set(key, value);
    }

    
    public void setNxPx(String key, byte[] value, long expire) {
        String result = jedis.set(key.getBytes(), value, NX.getBytes(), PX.getBytes(), expire);
        log.info("setNx key={} result={}", key, result);
    }

    
    public String setEx(String key, byte[] value, long expire) {
        String result = jedis.set(key.getBytes(), value, XX.getBytes(), PX.getBytes(), expire);
        log.info("setEx key={} result={}", key, result);
        return result;
    }

    
    public byte[] get(String key) {
        return jedis.get(key.getBytes());
    }

    @PostConstruct
    void init() {
        JedisConnection conn = (JedisConnection) redisTemplate.getConnectionFactory().getConnection();
        this.jedis = conn.getNativeConnection();
    }
}

        for (int i=0; i< 50; i++) {
            int finalI = i;
            new Thread(() -> {
                byte[] lastTimeNotify = jedisServiceOperate.get("abc" + finalI);
                if (lastTimeNotify == null) {
                    jedisServiceOperate.setNxPx("abc" + finalI, String.valueOf(System.currentTimeMillis()).getBytes(),
                            259200000L);
                } else {
                    // do other thing
                }
            }).start();
        }

意思是如果当前Key不存在就写入redis，存在就拿出来做自己的业务判断逻辑。

上线后出现问题：

1. java.net.SocketException: Connection reset by peer: socket write error.
2. setNxPx进去redis的value莫名其妙变成字符串：“OK”，我设置的值明明是String.valueOf(System.currentTimeMillis()).getBytes()

开始还以为是程序逻辑bug导致的，找了半天没发现什么猫腻，后面经过一番折腾，才发现是jedis在高并发环境下会有线程安全问题，会出现很多莫名其妙的错误，比如set进去的value变成字符串“OK”

Jedis不是线程安全的（只能怪自己技术不精啊-_-）

依然可以使用Jedis，但是每个线程都要隔离，也就是都有自己的Jedis实例，不能共享同一个实例，但是如果每次都开一个Jedis那网络开销必定会成为瓶颈，好，那怎么办？ 我们可以从JedisPool池中去拿，那么就不用每次开销了，对吧，好，说干就干。

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.lang.reflect.Field;


@Slf4j
@Component
public class JedisServiceOperate implements InitializingBean {

    private static final String NX = "NX";

    private static final String PX = "PX";

    private static final String XX = "XX";

    private JedisPool jedisPool;

    private JedisConnectionFactory jedisConnectionFactory;

    JedisServiceOperate(JedisConnectionFactory connectionFactory) {
        this.jedisConnectionFactory = connectionFactory;
    }

    
    public void setNxPx(String key, byte[] value, long expire) {
        Jedis jedis = null;
        try {
            jedis = getJedis();
            String result = jedis.set(key.getBytes(), value, NX.getBytes(), PX.getBytes(), expire);
            log.info("setNx key={} result={}", key, result);
        } catch (Exception e) {
            throw new RuntimeException("JedisServiceOperate setNxPx error", e);
        } finally {
            close(jedis);
        }
    }

    
    public String setEx(String key, byte[] value, long expire) {
        Jedis jedis = null;
        try {
            jedis = getJedis();
            String result = jedis.set(key.getBytes(), value, XX.getBytes(), PX.getBytes(), expire);
            log.info("setEx key={} result={}", key, result);
            return result;
        } catch (Exception e) {
            throw new RuntimeException("JedisServiceOperate setEx error", e);
        } finally {
            close(jedis);
        }
    }

    
    public byte[] get(String key) {
        Jedis jedis = null;
        try {
            jedis = getJedis();
            return jedis.get(key.getBytes());
        } catch (Exception e) {
            throw new RuntimeException("JedisServiceOperate get error", e);
        } finally {
            close(jedis);
        }
    }

    private void close(Jedis jedis) {
        if (jedis != null) {
            jedis.close();
        }
    }

    @Override
    public void afterPropertiesSet() {
        Field poolField = ReflectionUtils.findField(JedisConnectionFactory.class, "pool");
        ReflectionUtils.makeAccessible(poolField);
        jedisPool = (JedisPool) ReflectionUtils.getField(poolField, jedisConnectionFactory);
    }

    private Jedis getJedis() {
        if(jedisPool != null) {
            return jedisPool.getResource();
        } else {
            afterPropertiesSet();
            return jedisPool.getResource();
        }
    }
}

改进后主要是从JedisConnectionFactory对象中拿到pool这个属性，因为spring-data-redis和spring boot自动装配原理，程序启动已经创建好这个对象，直接拿就行。

好了，这样再去操作Redis就不会有什么问题了，一定要记得用完jedis要关闭close，要不然跑一段时间就报获取不到连接了，因为连接泄露了。