Redis事务实现

package com.example.redis.command;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @author ZxT
 * @date 2020-03-10
 * @desc redis分布式锁事务实现
 */
@Component
public class RedisLock {

    private final String lock_prefix = "lock:";
    private static ThreadLocal<Thread> THREA_LOCAL = new ThreadLocal<>();

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public String lock(String lock, long timeout) throws InterruptedException {
        return this.lock(lock, timeout, 2000);
    }

    public String lock(String lock, long timeout, long tryLockTimeout) throws InterruptedException {
        String identification = UUID.randomUUID().toString();
        String lockName = String.format("%s%s", lock_prefix, lock);
        tryLockTimeout = System.currentTimeMillis() + tryLockTimeout;
        while (System.currentTimeMillis() < tryLockTimeout) {
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockName, identification, timeout, TimeUnit.MILLISECONDS);
            if (flag) {
                System.out.println(Thread.currentThread().getName() + "获得锁");

                // 续签
                Thread renewal = new Thread(new RenewalLock(lock, identification, timeout));
                renewal.setDaemon(true);
                THREA_LOCAL.set(renewal);
                renewal.start();
                return identification;
            }
            Thread.sleep(10);
        }
        System.err.println(Thread.currentThread().getName() + "超时未获得锁");
        return null;
    }

    public void release(String lock, String identification) {
        redisTemplate.setEnableTransactionSupport(true);
        String lockName = String.format("%s%s", lock_prefix, lock);
        redisTemplate.watch(lockName);
        String identificationDB = redisTemplate.opsForValue().get(lockName);
        if(identificationDB != null && identification != null && identification.equals(identificationDB)) {
            System.out.println(Thread.currentThread().getName() + "释放锁");
            redisTemplate.multi();
            redisTemplate.delete(lockName);
            redisTemplate.exec();
            THREA_LOCAL.get().interrupt();
        }
        redisTemplate.unwatch();
    }

    /**
     * 续签lock
     */
    private class RenewalLock implements Runnable {

        private String lockName;
        private String identification;
        private Long timeout;

        public RenewalLock(String lock, String identification, Long timeout) {
            this.lockName = lock_prefix + lock;
            this.identification = identification;
            this.timeout = timeout;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    if(Thread.currentThread().isInterrupted()) {
                        return;
                    }
                    Thread.sleep(timeout - 500);
                    redisTemplate.setEnableTransactionSupport(true);
                    redisTemplate.watch(lockName);
                    String identificationDB = redisTemplate.opsForValue().get(lockName);
                    if(identificationDB != null && identification != null && identificationDB.equals(identification)) {
                        redisTemplate.multi();
                        redisTemplate.expire(lockName, timeout, TimeUnit.MILLISECONDS);
                        redisTemplate.exec();
                    }
                    redisTemplate.unwatch();
                } catch (InterruptedException e) {
                    System.err.println("守护线程续签中断");
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

Redis Lua实现

package com.example.redis.command;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.UUID;

/**
 * @author ZxT
 * @date 2020-03-10
 * @desc redis分布式锁lua实现
 */
@Component
public class RedisLockLua {

    private final String lock_prefix = "lock:";
    private static ThreadLocal<Thread> THREA_LOCAL = new ThreadLocal<>();

    public static DefaultRedisScript<Long> TRY_LOCK;
    public static DefaultRedisScript<Long> RELEASE_LOCK;
    public static DefaultRedisScript<Long> RENEWAL_LOCK;

    static {
//        StringBuilder tryLock = new StringBuilder();
//        tryLock.append("if redis.call('set', KEYS[1], ARGV[1], 'nx', 'px', ARGV[2]) == 1 ");
//        tryLock.append("then");
//        tryLock.append("    return 1 ");
//        tryLock.append("else");
//        tryLock.append("    return 0 ");
//        tryLock.append("end");
//        TRY_LOCK = new DefaultRedisScript<>(tryLock.toString());
        StringBuilder tryLock = new StringBuilder();
        tryLock.append("if redis.call('setnx', KEYS[1], ARGV[1]) == 1 ");
        tryLock.append("then ");
        tryLock.append("    return redis.call('pexpire', KEYS[1], ARGV[2]) ");
        tryLock.append("else");
        tryLock.append("    return 0 ");
        tryLock.append("end");
        TRY_LOCK = new DefaultRedisScript<>(tryLock.toString());
        TRY_LOCK.setResultType(Long.class);

        StringBuilder releaseLock = new StringBuilder();
        releaseLock.append("if redis.call('get', KEYS[1]) == ARGV[1] ");
        releaseLock.append("then ");
        releaseLock.append("    return redis.call('del', KEYS[1]) ");
        releaseLock.append("else");
        releaseLock.append("    return 0 ");
        releaseLock.append("end");
        RELEASE_LOCK = new DefaultRedisScript<>(releaseLock.toString());
        RELEASE_LOCK.setResultType(Long.class);

        StringBuilder renewalLock = new StringBuilder();
        renewalLock.append("if redis.call('get', KEYS[1]) == ARGV[1] ");
        renewalLock.append("then ");
        renewalLock.append("    return redis.call('expire', KEYS[1], ARGV[2]) ");
        renewalLock.append("else");
        renewalLock.append("    return 0 ");
        renewalLock.append("end");
        RENEWAL_LOCK = new DefaultRedisScript<>(renewalLock.toString());
        RENEWAL_LOCK.setResultType(Long.class);

        System.out.println(RELEASE_LOCK.getSha1());
        System.out.println(RENEWAL_LOCK.getSha1());
    }


    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public String lock(String lock, long timeout) throws InterruptedException {
        return this.lock(lock, timeout, 2000);
    }

    public String lock(String lock, long timeout, long tryLockTimeout) throws InterruptedException {
        String identification = UUID.randomUUID().toString();
        String lockName = String.format("%s%s", lock_prefix, lock);
        tryLockTimeout = System.currentTimeMillis() + tryLockTimeout;
        while (System.currentTimeMillis() < tryLockTimeout) {
            Long execute = redisTemplate.execute(TRY_LOCK, Collections.singletonList(lockName), identification, String.valueOf(timeout));
            if(execute == 1) {
                System.out.println(Thread.currentThread().getName() + "获得锁");
                // 续签
                Thread renewal = new Thread(new RedisLockLua.RenewalLock(lock, identification, timeout));
                renewal.setDaemon(true);
                THREA_LOCAL.set(renewal);
                renewal.start();
                return identification;
            }
            Thread.sleep(10);
        }
        System.err.println(Thread.currentThread().getName() + "超时未获得锁");
        return null;
    }

    public void release(String lock, String identification) {
        redisTemplate.setEnableTransactionSupport(true);
        String lockName = String.format("%s%s", lock_prefix, lock);
        Long execute = redisTemplate.execute(RELEASE_LOCK, Collections.singletonList(lockName), identification);
        if(execute == 1) {
            System.out.println(Thread.currentThread().getName() + "释放锁");
            THREA_LOCAL.get().interrupt();
        }
    }

    /**
     * 续签lock
     */
    private class RenewalLock implements Runnable {

        private String lockName;
        private String identification;
        private Long timeout;

        public RenewalLock(String lock, String identification, Long timeout) {
            this.lockName = lock_prefix + lock;
            this.identification = identification;
            this.timeout = timeout;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    if(Thread.currentThread().isInterrupted()) {
                        return;
                    }
                    Thread.sleep(timeout / 2);
                    Long execute = redisTemplate.execute(RENEWAL_LOCK, Collections.singletonList(lockName), identification, String.valueOf(timeout));
                    if(execute == 1) {
                        System.err.println("守护线程续签成功");
                    }
                } catch (InterruptedException e) {
                    System.err.println("守护线程续签中断");
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}