Redis集群实现分布式锁的正确方式

胖虎

发布于 2019-06-26 17:11:38

7.4K0

发布于 2019-06-26 17:11:38

文章被收录于专栏：晏霖晏霖

前言

上文我们介绍的 Redis实现分布式锁的正确方式是 redis 单机的方式，所以本篇要基于 redis 集群做分布式锁，我们使用 Redisson

来做。

有同学肯定有这样的疑问

为什么用 redis 集群方式啊，单机不是挺好的？

集群和单机不就是 redis 多节点配置的问题吗？为什么使用另一种方式 Redisson呢？

Redisson 是什么呢？

可不可以不用Redisson呢？

……

其实在研究集群实现分布式锁之前我也有这些问题，所以，我下面一一为各位解答，并用代码实现。

正文

介绍一下本次使用所有框架和中间件的版本

框架	版本
Spring Boot	2.0.3.RELEASE
Spring Cloud	Finchley.RELEASE
redis	redis-4.0.11
JDK	1.8.x

前置准备工作

本机安装一个 redis 集群，端口按默认的，然后启动，mac用户可看博主博客 Mac系统搭建Redis集群模式，公众号用户可复制链接：https://blog.csdn.net/weixin_38003389/article/details/86295944。
创建一个 eureka-service ，端口随意，保证你的 eureka客户端能注册上，然后启动。
父工程pom文件，滑动滚轮即可看到pom 的内容。
本文基于上篇文章的redis -tool 工程改写的，有兴趣可关注并查看博主更多文章。

父 pom

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.3.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

<dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
<dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.5.4</version>
        </dependency>

Redisson概述

Redisson是一个基于java编程框架netty进行扩展了的redis。

Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid）。充分的利用了Redis键值数据库提供的一系列优势，基于Java实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。

地址：https://github.com/redisson/redisson

Redisson 适用于：分布式应用，分布式缓存，分布式回话管理，分布式服务（任务，延迟任务，执行器），分布式redis客户端。

还有一个重要的点需要说明

使用 Redisson 使用除了上面父pom 中的依赖，还需要进行 Redisson 配置、连接、设置参数等等，这是必须的，好比使用 Jedis 你要配置一个 redisPool 的Bean一样。

目前操作 Redisson 有三种方式

第一种：纯java操作，本文就是使用这种，所有的配置都写在一个 Class 里。
第二种：spring集成操作，编写一个 xml，配置一个bean，启动还需读取这个文件，一堆很原始的操作。使用这种 xml 配置我看着都烦，强烈不推荐。
第三种：文件方式配置，是把所有配置的参数放到配置文件声明，然后在 Class 中读取。

核心代码示例

我们要使用 Redisson

Redisson管理类

import org.redisson.Redisson;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;



public class RedissonManager {
    private static Config config = new Config();
    private static RedissonClient redisson = null;
    private static final String RAtomicName = "genId_";
    public static void init(){
        try{
            config.useClusterServers()
                    .setScanInterval(200000)//设置集群状态扫描间隔
                    .setMasterConnectionPoolSize(10000)//设置对于master节点的连接池中连接数最大为10000
                    .setSlaveConnectionPoolSize(10000)//设置对于slave节点的连接池中连接数最大为500
                    .setIdleConnectionTimeout(10000)//如果当前连接池里的连接数量超过了最小空闲连接数，而同时有连接空闲时间超过了该数值，那么这些连接将会自动被关闭，并从连接池里去掉。时间单位是毫秒。
                    .setConnectTimeout(30000)//同任何节点建立连接时的等待超时。时间单位是毫秒。
                    .setTimeout(3000)//等待节点回复命令的时间。该时间从命令发送成功时开始计时。
                    .setRetryInterval(3000)//当与某个节点的连接断开时，等待与其重新建立连接的时间间隔。时间单位是毫秒。
                    .addNodeAddress("redis://127.0.0.1:7000","redis://127.0.0.1:7001","redis://127.0.0.1:7002","redis://127.0.0.1:7003","redis://127.0.0.1:7004","redis://127.0.0.1:7005");
            redisson = Redisson.create(config);

            RAtomicLong atomicLong = redisson.getAtomicLong(RAtomicName);
            atomicLong.set(0);//自增设置为从0开始
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    public static RedissonClient getRedisson(){
        if(redisson == null){
            RedissonManager.init(); //初始化
        }
        return redisson;
    }

代码解释

我们配置了很多参数，其实一共有十来种参数，我们只是设置几个比较重要的而已。

getRedisson 方法是使用者初始化 Redisson。

nextID 方法返回一共为 RAtomicName 变量操作了多少次，也就是我成功使用分布式锁的次数。

分布式锁操作类

import com.config.RedissonManager;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;


@Component
public class RedissonLock {
    private static final Logger LOGGER = LoggerFactory.getLogger(RedissonLock.class);

    private static RedissonClient redissonClient = RedissonManager.getRedisson();


    public void lock(String lockName) {
        String key = lockName;
        RLock myLock = redissonClient.getLock(key);
        //lock提供带timeout参数，timeout结束强制解锁，防止死锁
        myLock.lock(2, TimeUnit.SECONDS);
        // 1. 最常见的使用方法
        //lock.lock();
        // 2. 支持过期解锁功能,10秒以后自动解锁, 无需调用unlock方法手动解锁
        //lock.lock(10, TimeUnit.SECONDS);
        // 3. 尝试加锁，最多等待3秒，上锁以后10秒自动解锁
//        try {
//            boolean res = mylock.tryLock(3, 10, TimeUnit.SECONDS);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        System.err.println("======lock======" + Thread.currentThread().getName());
    }

    public void unLock(String lockName) {
        String key = lockName;
        RLock myLock = redissonClient.getLock(key);
        myLock.unlock();
        System.err.println("======unlock======" + Thread.currentThread().getName());
    }
}

代码解释

lock 方法是加锁操作，unLock 方法是解锁操作。

注释中的代码列举类 3中lock 的方法，大家学习更多操作请查看下面博客

https://blog.csdn.net/l1028386804/article/details/73523810

教你 RedissonClient 所有操作。

All code ends , so easy

测试

在我们的 eureka 客户端启动类编辑

@SpringBootApplication
@EnableDiscoveryClient
@ComponentScan(value = {"com.annotaion", "cn.springcloud", "com.config", "com.redislock"})

public class Ch34EurekaClientApplication implements ApplicationRunner {
    private static final ExecutorService executorService = Executors.newFixedThreadPool(5);
    private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
   

    @Autowired
    RedissonLock redissonLock;

    public static void main(String[] args) {
        SpringApplication.run(Ch34EurekaClientApplication.class, args);

    }

    @Override
    public void run(ApplicationArguments args) throws Exception {

//******* Redis集群测试方法*********

        for (int i = 0; i < 5; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(Thread.currentThread().getName() + "开始等待其他线程");
                        cyclicBarrier.await();
                        System.out.println(Thread.currentThread().getName() + "线程就位，即将同时执行");
                        String key = "test123";
                        redissonLock.lock(key);
                        Thread.sleep(1000); //获得锁之后可以进行相应的处理
                        System.out.println(Thread.currentThread().getName() + "获取成功，并开始执行业务逻辑");
                        redissonLock.unLock(key);
                        System.out.println(Thread.currentThread().getName() + "释放成功");

                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }

                }
            });
        }
        executorService.shutdown();
        Long result = RedissonManager.nextID();
        System.out.print("获取redis中的原子ID" + result);

    }
}

输出如下