/** Sample Configuration **/

@Configuration

public class RedisTxContextConfiguration {

  @Bean

  public StringRedisTemplate redisTemplate() {

    StringRedisTemplate template = new StringRedisTemplate(redisConnectionFactory());

    // explicitly enable transaction support

    template.setEnableTransactionSupport(true);

    return template;

  }

  @Bean

  public PlatformTransactionManager transactionManager() throws SQLException {

    return new DataSourceTransactionManager(dataSource());

  }

  @Bean

  public RedisConnectionFactory redisConnectionFactory( // jedis, lettuce, srp,... );

  @Bean

  public DataSource dataSource() throws SQLException { // ... }

}

/** Usage Constrainsts **/

// executed on thread bound connection

template.opsForValue().set("foo", "bar");

// read operation executed on a free (not tx-aware)

connection template.keys("*");

// returns null as values set within transaction are not visible

template.opsForValue().get("foo");

    public Long leftPush(V value) {

        return this.ops.leftPush(this.getKey(), value);

    }

public Long leftPush(K key, V value) {

        final byte[] rawKey = this.rawKey(key);

        final byte[] rawValue = this.rawValue(value);

        return (Long)this.execute(new RedisCallback() {

            public Long doInRedis(RedisConnection connection) {

                return connection.lPush(rawKey, new byte[][]{rawValue});

            }

        }, true);

    }

    <T> T execute(RedisCallback<T> callback, boolean b) {

        return this.template.execute(callback, b);

    }

    public <T> T execute(RedisCallback<T> action, boolean exposeConnection) {

        return this.execute(action, exposeConnection, false);

    }

 public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline) {

        Assert.isTrue(this.initialized, "template not initialized; call afterPropertiesSet() before using it");

        Assert.notNull(action, "Callback object must not be null");

        RedisConnectionFactory factory = this.getConnectionFactory();

        RedisConnection conn = null;

        Object var11;

        try {

            if(this.enableTransactionSupport) {

                conn = RedisConnectionUtils.bindConnection(factory, this.enableTransactionSupport);

            } else {

                conn = RedisConnectionUtils.getConnection(factory);

            }

            boolean existingConnection = TransactionSynchronizationManager.hasResource(factory);

            RedisConnection connToUse = this.preProcessConnection(conn, existingConnection);

            boolean pipelineStatus = connToUse.isPipelined();

            if(pipeline && !pipelineStatus) {

                connToUse.openPipeline();

            }

            RedisConnection connToExpose = exposeConnection?connToUse:this.createRedisConnectionProxy(connToUse);

            Object result = action.doInRedis(connToExpose);

            if(pipeline && !pipelineStatus) {

                connToUse.closePipeline();

            }

            var11 = this.postProcessResult(result, connToUse, existingConnection);

        } finally {

            if(!this.enableTransactionSupport) {

                RedisConnectionUtils.releaseConnection(conn, factory);

            }

        }

        return var11;

    }

   @Test

    public void cas() throws InterruptedException, ExecutionException {

        String key = "test-cas-1";

        ValueOperations<String, String> strOps = redisTemplate.opsForValue();

        strOps.set(key, "hello");

        ExecutorService pool  = Executors.newCachedThreadPool();

        List<Callable<Object>> tasks = new ArrayList<>();

        for(int i=0;i<5;i++){

            final int idx = i;

            tasks.add(new Callable() {

                @Override

                public Object call() throws Exception {

                    return redisTemplate.execute(new SessionCallback() {

                        @Override

                        public Object execute(RedisOperations operations) throws DataAccessException {

                            operations.watch(key);

                            String origin = (String) operations.opsForValue().get(key);

                            operations.multi();

                            operations.opsForValue().set(key, origin + idx);

                            Object rs = operations.exec();

                            System.out.println("set:"+origin+idx+" rs:"+rs);

                            return rs;

                        }

                    });

                }

            });

        }

        List<Future<Object>> futures = pool.invokeAll(tasks);

        for(Future<Object> f:futures){

            System.out.println(f.get());

        }

        pool.shutdown();

        pool.awaitTermination(1000, TimeUnit.MILLISECONDS);

    }

template.setEnableTransactionSupport(true);

   /**

     * Gets a Redis connection. Is aware of and will return any existing corresponding connections bound to the current

     * thread, for example when using a transaction manager. Will create a new Connection otherwise, if

     * {@code allowCreate} is <tt>true</tt>.

     *

     * @param factory connection factory for creating the connection

     * @param allowCreate whether a new (unbound) connection should be created when no connection can be found for the

     *          current thread

     * @param bind binds the connection to the thread, in case one was created

     * @param enableTransactionSupport

     * @return an active Redis connection

     */

    public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind,

            boolean enableTransactionSupport) {

        Assert.notNull(factory, "No RedisConnectionFactory specified");

        RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);

        if (connHolder != null) {

            if (enableTransactionSupport) {

                potentiallyRegisterTransactionSynchronisation(connHolder, factory);

            }

            return connHolder.getConnection();

        }

        if (!allowCreate) {

            throw new IllegalArgumentException("No connection found and allowCreate = false");

        }

        if (log.isDebugEnabled()) {

            log.debug("Opening RedisConnection");

        }

        RedisConnection conn = factory.getConnection();

        if (bind) {

            RedisConnection connectionToBind = conn;

            if (enableTransactionSupport && isActualNonReadonlyTransactionActive()) {

                connectionToBind = createConnectionProxy(conn, factory);

            }

            connHolder = new RedisConnectionHolder(connectionToBind);

            TransactionSynchronizationManager.bindResource(factory, connHolder);

            if (enableTransactionSupport) {

                potentiallyRegisterTransactionSynchronisation(connHolder, factory);

            }

            return connHolder.getConnection();

        }

        return conn;

    }

 public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline) {

        Assert.isTrue(this.initialized, "template not initialized; call afterPropertiesSet() before using it");

        Assert.notNull(action, "Callback object must not be null");

        RedisConnectionFactory factory = this.getConnectionFactory();

        RedisConnection conn = null;

        Object var11;

        try {

            if(this.enableTransactionSupport) {

        conn = RedisConnectionUtils.bindConnection(factory, this.enableTransactionSupport);
//这个是否支持的开关可以在@Configuration中配置：

            } else {

                conn = RedisConnectionUtils.getConnection(factory);

            }

            boolean existingConnection = TransactionSynchronizationManager.hasResource(factory);

            RedisConnection connToUse = this.preProcessConnection(conn, existingConnection);

            boolean pipelineStatus = connToUse.isPipelined();

            if(pipeline && !pipelineStatus) {

                connToUse.openPipeline();

            }

            RedisConnection connToExpose = exposeConnection?connToUse:this.createRedisConnectionProxy(connToUse);

            Object result = action.doInRedis(connToExpose);

            if(pipeline && !pipelineStatus) {

                connToUse.closePipeline();

            }

            var11 = this.postProcessResult(result, connToUse, existingConnection);

        } finally {

            if(!this.enableTransactionSupport) {

                RedisConnectionUtils.releaseConnection(conn, factory);

            }

        }

        return var11;

    }

    @Bean

    public RedisTemplate redisTemplate(){

        RedisTemplate<StringRedisSerializer, Serializable> rt = new RedisTemplate<>();

        rt.setConnectionFactory(jedisConnectionFactory());

        StringRedisSerializer stringSerializer = new StringRedisSerializer();

        JdkSerializationRedisSerializer jdkSerializationRedisSerializer = new JdkSerializationRedisSerializer();

        RedisHashKeySerializer redisHashKeySerializer = new RedisHashKeySerializer();

        rt.setKeySerializer(stringSerializer);

        rt.setValueSerializer(jdkSerializationRedisSerializer);

        rt.setHashKeySerializer(redisHashKeySerializer);

        rt.setHashValueSerializer(jdkSerializationRedisSerializer);

        rt.afterPropertiesSet();

    rt.setEnableTransactionSupport(true);

    return rt;

    }

    public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind, boolean enableTransactionSupport) {

        Assert.notNull(factory, "No RedisConnectionFactory specified");

        RedisConnectionUtils.RedisConnectionHolder connHolder = (RedisConnectionUtils.RedisConnectionHolder)TransactionSynchronizationManager.getResource(factory);

        if(connHolder != null) {

            if(enableTransactionSupport) {

                potentiallyRegisterTransactionSynchronisation(connHolder, factory);

            }

            return connHolder.getConnection();

        } else if(!allowCreate) {

            throw new IllegalArgumentException("No connection found and allowCreate = false");

        } else {

            if(log.isDebugEnabled()) {

                log.debug("Opening RedisConnection");

            }

            RedisConnection conn = factory.getConnection();

            if(bind) {

                RedisConnection connectionToBind = conn;

                if(enableTransactionSupport && isActualNonReadonlyTransactionActive()) {

                    connectionToBind = createConnectionProxy(conn, factory);

                }

                connHolder = new RedisConnectionUtils.RedisConnectionHolder(connectionToBind);

                TransactionSynchronizationManager.bindResource(factory, connHolder);

                if(enableTransactionSupport) {

                    potentiallyRegisterTransactionSynchronisation(connHolder, factory);

                }

                return connHolder.getConnection();

            } else {

                return conn;

            }

        }

    }

private static void potentiallyRegisterTransactionSynchronisation(RedisConnectionUtils.RedisConnectionHolder connHolder, RedisConnectionFactory factory) {

        if(isActualNonReadonlyTransactionActive() && !connHolder.isTransactionSyncronisationActive()) {

            connHolder.setTransactionSyncronisationActive(true);

            RedisConnection conn = connHolder.getConnection();

            conn.multi();

            TransactionSynchronizationManager.registerSynchronization(new RedisConnectionUtils.RedisTransactionSynchronizer(connHolder, conn, factory));

        }

    }

    private boolean enableTransactionSupport = false;

    private boolean exposeConnection = false;

    private boolean initialized = false;

    private boolean enableDefaultSerializer = true;

    private RedisSerializer<?> defaultSerializer = new JdkSerializationRedisSerializer();

    private RedisSerializer keySerializer = null;

    private RedisSerializer valueSerializer = null;

    private RedisSerializer hashKeySerializer = null;

    private RedisSerializer hashValueSerializer = null;

    private RedisSerializer<String> stringSerializer = new StringRedisSerializer();

    private ScriptExecutor<K> scriptExecutor;

    // cache singleton objects (where possible)

    private ValueOperations<K, V> valueOps;

    private ListOperations<K, V> listOps;

    private SetOperations<K, V> setOps;

    private ZSetOperations<K, V> zSetOps;

RedisAtomicLong counter = new RedisAtomicLong("UNIQUE_COUNTER_NAME", redisTemplate.getConnectionFactory());

Long myCounter = counter.incrementAndGet();// return the incremented value

import org.apache.commons.lang3.StringUtils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.redis.core.BoundValueOperations;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.stereotype.Service;

import java.util.concurrent.ThreadLocalRandom;

import java.util.concurrent.TimeUnit;

@Service

public class DistributedLockRedis {

    public static final Logger LOGGER = LoggerFactory.getLogger(DistributedLockRedis.class);

    public static final String DISTRIBUTED_LOCK_REDIS_KEY = "distributed:lock:redis:";

    @Autowired

    private RedisTemplate<String, Long> redisTemplate;

    /**

     * 由于Redis是单线程模型，命令操作原子性，所以利用这个特性可以很容易的实现分布式锁。

     * 获得一个锁

     *

     * @param bizName     业务名

     * @param lockTimeout 线程占用锁的时间

     * @param unit        单位

     * @throws InterruptedException 锁可以被中断

     */

    public void lock(String bizName, int lockTimeout, TimeUnit unit) throws InterruptedException {

//        redisTemplate.getConnectionFactory().getConnection().setNX()

        String redisKey;

        if (StringUtils.isBlank(bizName)) {

            LOGGER.warn("is not recommended!");

            redisKey = DISTRIBUTED_LOCK_REDIS_KEY;

        } else {

            redisKey = DISTRIBUTED_LOCK_REDIS_KEY + bizName.trim();

        }

        BoundValueOperations<String, Long> valueOps = redisTemplate.boundValueOps(redisKey);

        while (true) {

            // https://redis.io/commands/setnx

            long currentTimeMillis = System.currentTimeMillis();

            long releaseLockTime = currentTimeMillis + unit.toMillis(lockTimeout) + 1;

            //这两个if else不能混写，因为多个相同类型的线程竞争锁时,在锁超时时，设置的超时时间是一样的

            if (valueOps.setIfAbsent(releaseLockTime)) {//第一次获取锁

                redisTemplate.expire(redisKey, lockTimeout, unit);

                return;

            } else if (currentTimeMillis > valueOps.get()) {//锁已经超时

                //如果其它线程占用锁，再重新设置的时间和原来时间的时间差，可以忽略

                Long lockCurrentValue = valueOps.getAndSet(releaseLockTime);

                //如果当前时间小于LockKey存放的时间，说明已经有其它线程加锁

                if (currentTimeMillis > lockCurrentValue) {

                    redisTemplate.expire(redisKey, lockTimeout, unit);

                    return;

                }

            } else {

                TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(100, 1000));

            }

        }

    }

}

LOCK_TIMEOUT = 3

lock = 0

lock_timeout = 0

lock_key = 'lock.foo'

# 获取锁

while lock != 1:

    now = int(time.time())

    lock_timeout = now + LOCK_TIMEOUT + 1

    lock = redis_client.setnx(lock_key, lock_timeout)

    if lock == 1 or (now > int(redis_client.get(lock_key))) and now > int(redis_client.getset(lock_key, lock_timeout)):

        break

    else:

        time.sleep(0.001)

# 已获得锁

do_job()

# 释放锁

now = int(time.time())

if now < lock_timeout:

    redis_client.delete(lock_key)

@Override

public List<LeaderboardEntry> fetchLeaderboard(String key, String... playerIds) {

   final List<LeaderboardEntry> entries = new ArrayList<>();

    redisTemplate.executePipelined(new RedisCallback<Object>() {    // enable Redis Pipeline

    @Override

        public Object doInRedis(RedisConnection connection) throws DataAccessException {

            for(String playerId : playerIds) {

                Long rank = connection.zRevRank(key.getBytes(), playerId.getBytes());

                Double score = connection.zScore(key.getBytes(), playerId.getBytes());

                LeaderboardEntry entry = new LeaderboardEntry(playerId,

                score!=null?score.intValue():-1, rank!=null?rank.intValue():-1);

                entries.add(entry);

            }

            return null;

        }

    });

    return entries;

}

I learned a hard lesson when enabling Redis transactions in the Spring RedisTemplate class redisTemplate.setEnableTransactionSupport(true);:

Redis started returning junk data after running for a few days, causing serious data corruption. A similar case was reported on StackOverflow.

By running a monitor command, my team discovered that after a Redis operation or RedisCallback, Spring doesn't close the Redis connection automatically, as it should do.

Reusing an unclosed connection may return junk data from an unexpected key in Redis. Interestingly, this issue doesn't show up when transaction support is set to false in RedisTemplate.

We discovered that we could make Spring close Redis connections automatically by configuring a PlatformTransactionManager (such as DataSourceTransactionManager) in the Spring context,

then using the @Transactional annotation to declare the scope of Redis transactions.

Based on this experience, we believe it's good practice to configure two separate RedisTemplates in the Spring context:

One with transaction set to false is used on most Redis operations; the other with transaction enabled is only applied to Redis transactions.

Of course PlatformTransactionManager and @Transactional must be declared to prevent junk values from being returned.

Moreover, we learned the downside of mixing a Redis transaction with a relational database transaction, in this case JDBC.

Mixed transactions do not behave as you would expect.

@Bean

public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {

    RedisTemplate redisTemplate = new RedisTemplate();

    redisTemplate.setConnectionFactory(redisConnectionFactory);

    GenericFastJsonRedisSerializer fastJsonRedisSerializer = new GenericFastJsonRedisSerializer();

    redisTemplate.setDefaultSerializer(fastJsonRedisSerializer);//设置默认的Serialize，包含 keySerializer & valueSerializer

    //redisTemplate.setKeySerializer(fastJsonRedisSerializer);//单独设置keySerializer

    //redisTemplate.setValueSerializer(fastJsonRedisSerializer);//单独设置valueSerializer

    return redisTemplate;

}