利用redis实现分布式锁,快速解决高并发时的线程安全问题

时间：2021-10-19 05:51:30

实际工作中，经常会遇到多线程并发时的类似抢购的功能，本篇描述一个简单的redis分布式锁实现的多线程抢票功能。

直接上代码。首先按照慣例，给出一个错误的示范：

我们可以看看，当20个线程一起来抢10张票的时候，会发生什么事。

?
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
package com.tiger.utils;
public class TestMutilThread {
 
    // 总票量
    public static int count = 10;
    public static void main(String[] args) {
        statrtMulti();
    }
 
    public static void statrtMulti() {
        for (int i = 1; i <= 20; i++) {
            TicketRunnable tickrunner = new TicketRunnable();
            Thread thread = new Thread(tickrunner, "Thread No: " + i);
            thread.start();
        }
    }
 
    public static class TicketRunnable implements Runnable {
 
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " start "
                    + count);
            // TODO Auto-generated method stub
            // logger.info(Thread.currentThread().getName()
            // + " really start" + count);
            if (count <= 0) {
                System.out.println(Thread.currentThread().getName()
                        + " ticket sold out ! No tickets remained!" + count);
                return;
            } else {
                count = count - 1;
                System.out.println(Thread.currentThread().getName()
                        + " bought a ticket,now remaining :" + (count));
            }
        }
    }
}

测试结果，从结果可以看到，票数在不同的线程中已经出现混乱。

?
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
Thread No: 2 start 10
Thread No: 6 start 10
Thread No: 4 start 10
Thread No: 5 start 10
Thread No: 3 start 10
Thread No: 9 start 6
Thread No: 1 start 10
Thread No: 1 bought a ticket,now remaining :3
Thread No: 9 bought a ticket,now remaining :4
Thread No: 3 bought a ticket,now remaining :5
Thread No: 12 start 3
Thread No: 5 bought a ticket,now remaining :6
Thread No: 4 bought a ticket,now remaining :7
Thread No: 8 start 7
Thread No: 7 start 8
Thread No: 12 bought a ticket,now remaining :1
Thread No: 14 start 0
Thread No: 6 bought a ticket,now remaining :8
Thread No: 16 start 0
Thread No: 2 bought a ticket,now remaining :9
Thread No: 16 ticket sold out ! No tickets remained!0
Thread No: 14 ticket sold out ! No tickets remained!0
Thread No: 18 start 0
Thread No: 18 ticket sold out ! No tickets remained!0
Thread No: 7 bought a ticket,now remaining :0
Thread No: 15 start 0
Thread No: 8 bought a ticket,now remaining :1
Thread No: 13 start 2
Thread No: 19 start 0
Thread No: 11 start 3
Thread No: 11 ticket sold out ! No tickets remained!0
Thread No: 10 start 3
Thread No: 10 ticket sold out ! No tickets remained!0
Thread No: 19 ticket sold out ! No tickets remained!0
Thread No: 13 ticket sold out ! No tickets remained!0
Thread No: 20 start 0
Thread No: 20 ticket sold out ! No tickets remained!0
Thread No: 15 ticket sold out ! No tickets remained!0
Thread No: 17 start 0
Thread No: 17 ticket sold out ! No tickets remained!0

为了解决多线程时出现的混乱问题，这里給出真正的测试类!!!

真正的测试类，这里启动20个线程，来抢10张票。

RedisTemplate 是用来实现redis操作的，由spring进行集成。这里是使用到了RedisTemplate，所以我以构造器的形式在外部将RedisTemplate传入到测试类中。

MultiTestLock 是用来实现加锁的工具类。

总票数使用volatile关键字，实现多线程时变量在系统内存中的可见性，这点可以去了解下volatile关键字的作用。

TicketRunnable用于模拟抢票功能。

其中由于lock与unlock之间存在if判断，为保证线程安全，这里使用synchronized来保证。

测试类：

?
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
package com.tiger.utils;
import java.io.Serializable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
public class MultiConsumer {
    Logger logger=LoggerFactory.getLogger(MultiTestLock.class);
    private RedisTemplate<Serializable, Serializable> redisTemplate; 
    public MultiTestLock lock;
    //总票量
    public volatile static int count = 10;
 
    public void statrtMulti() {
        lock = new MultiTestLock(redisTemplate);
        for (int i = 1; i <= 20; i++) {
            TicketRunnable tickrunner = new TicketRunnable();
            Thread thread = new Thread(tickrunner, "Thread No: " + i);
            thread.start();
            }
    }
 
    public class TicketRunnable implements Runnable {
 
        @Override
        public void run() {
            logger.info(Thread.currentThread().getName() + " start "
                    + count);
            // TODO Auto-generated method stub
            if (count > 0) {
//              logger.info(Thread.currentThread().getName()
//                      + " really start" + count);
                lock.lock();
                synchronized (this) {
                    if(count<=0){
                        logger.info(Thread.currentThread().getName()
                                + " ticket sold out ! No tickets remained!" + count);
                        lock.unlock();
                        return;
                    }else{
                        count=count-1;
                        logger.info(Thread.currentThread().getName()
                                + " bought a ticket,now remaining :" + (count));
                    }
                }
                lock.unlock();
            }else{
                logger.info(Thread.currentThread().getName()
                        + " ticket sold out !" + count);
            }
        }
    }
 
    public RedisTemplate<Serializable, Serializable> getRedisTemplate() {
        return redisTemplate;
    }
 
    public void setRedisTemplate(
            RedisTemplate<Serializable, Serializable> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
 
    public MultiConsumer(RedisTemplate<Serializable, Serializable> redisTemplate) {
        super();
        this.redisTemplate = redisTemplate;
    }
}

Lock工具类：

我们知道为保证线程安全，程序中执行的操作必须时原子的。redis后续的版本中可以使用set key同时设置expire超时时间。

想起上次去 电信翼支付 面试时，面试官问过一个问题：分布式锁如何防止死锁，问题关键在于我们在分布式中进行加锁操作时成功了，但是后续业务操作完毕执行解锁时出现失败。导致分布式锁无法释放。出现死锁，后续的加锁无法正常进行。所以这里设置expire超时时间的目的就是防止出现解锁失败的情况，这样，即使解锁失败了，分布式锁依然会在超时时间过了之后自动释放。

具体在代码中也有注释，也可以作为参考。

?
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
package com.tiger.utils;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import javax.sound.midi.MidiDevice.Info;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.script.RedisScript;
 
public class MultiTestLock implements Lock {   
    Logger logger=LoggerFactory.getLogger(MultiTestLock.class);
    private RedisTemplate<Serializable, Serializable> redisTemplate; 
    public MultiTestLock(RedisTemplate<Serializable, Serializable> redisTemplate) {
        super();
        this.redisTemplate = redisTemplate;
    }
 
    @Override
    public void lock() {
        //这里使用while循环强制线程进来之后先进行抢锁操作。只有抢到锁才能进行后续操作
        while(true){
            if(tryLock()){
                try {
                    //这里让线程睡500毫秒的目的是为了模拟业务耗时，确保业务结束时之前设置的值正好打到超时时间，
                    //实际生产中可能有偏差，这里需要经验
                    Thread.sleep(500l);
//                  logger.info(Thread.currentThread().getName()+" time to awake");
                    return;
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }else{
                try {
                    //这里设置一个随机毫秒的sleep目的时降低while循环的频率
                    Thread.sleep(new Random().nextInt(200)+100);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
    }
 
    @Override
    public boolean tryLock() {
        //这里也可以选用transactionSupport支持事务操作
        SessionCallback<Object> sessionCallback=new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations)
                    throws DataAccessException {
                operations.multi();
                operations.opsForValue().setIfAbsent("secret", "answer");
                //设置超时时间要根据业务实际的可能处理时间来，是一个经验值
                operations.expire("secret", 500l, TimeUnit.MILLISECONDS);
                Object object=operations.exec();
                return object;
            }
        };
        //执行两部操作，这里会拿到一个数组值 [true,true],分别对应上述两部操作的结果，如果中途出现第一次为false则表明第一步set值出错
        List<Boolean> result=(List) redisTemplate.execute(sessionCallback);
//      logger.info(Thread.currentThread().getName()+" try lock "+ result);
        if(true==result.get(0)||"true".equals(result.get(0)+"")){
            logger.info(Thread.currentThread().getName()+" try lock success");
            return true;
        }else{
            return false;
        }
    }
 
    @Override
    public boolean tryLock(long arg0, TimeUnit arg1)
            throws InterruptedException {
        // TODO Auto-generated method stub
        return false;
    }
 
    @Override
    public void unlock() {
        //unlock操作直接删除锁，如果执行完还没有达到超时时间则直接删除，让后续的线程进行继续操作。起到补刀的作用，确保锁已经超时或被删除
        SessionCallback<Object> sessionCallback=new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations)
                    throws DataAccessException {
                operations.multi();
                operations.delete("secret");
                Object object=operations.exec();
                return object;
            }
        };
        Object result=redisTemplate.execute(sessionCallback);
    }
 
    @Override
    public void lockInterruptibly() throws InterruptedException {
        // TODO Auto-generated method stub
    }
 
    @Override
    public Condition newCondition() {
        // TODO Auto-generated method stub
        return null;
    }
    
    public RedisTemplate<Serializable, Serializable> getRedisTemplate() {
        return redisTemplate;
    }
 
    public void setRedisTemplate(
            RedisTemplate<Serializable, Serializable> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
}

执行结果

利用redis实现分布式锁,快速解决高并发时的线程安全问题

可以看到，票数稳步减少，后续没有抢到锁的线程余票为0，无票可抢。

tips:

这其中也出现了一个问题，redis进行多部封装操作时，系统报错：ERR EXEC without MULTI

后经过查阅发现问题出在：

在spring中，多次执行MULTI命令不会报错，因为第一次执行时，会将其内部的一个isInMulti变量设为true，后续每次执行命令是都会检查这个变量，如果为true，则不执行命令。

而多次执行EXEC命令则会报开头说的"ERR EXEC without MULTI"错误。

利用redis实现分布式锁,快速解决高并发时的线程安全问题

以上为个人经验，希望能给大家一个参考，也希望大家多多支持服务器之家。如有错误或未考虑完全的地方，望不吝赐教。

原文链接：https://blog.csdn.net/qq_23974323/article/details/93165049

标签：

相关文章