Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先)

這篇具有很好參考價值的文章主要介紹了Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先)。希望對大家有所幫助。如果存在錯誤或未考慮完全的地方，請大家不吝賜教，您也可以點擊"舉報違法"按鈕提交疑問。

1. 分布式鎖

1.1 什么是分布式鎖

在我們進行單機應(yīng)用開發(fā)涉及并發(fā)同步的時候，我們往往采用synchronized或者ReentrantLock的方式來解決多線程間的代碼同步問題。但是當(dāng)我們的應(yīng)用是在分布式集群工作的情況下，那么就需要一種更加高級的鎖機制，來處理種跨機器的進程之間的數(shù)據(jù)同步問題，這就是分布式鎖。

分布式鎖，是控制分布式系統(tǒng)之間同步訪問共享資源的一種方式。在分布式系統(tǒng)中，常常需要協(xié)調(diào)他們的動作。如果不同的系統(tǒng)或是同一個系統(tǒng)的不同主機之間共享了一個或一組資源，那么訪問這些資源的時候，往往需要互斥來防止彼此干擾來保證一致性，在這種情況下，便需要使用到分布式鎖。

分布式鎖可以理解為：控制分布式系統(tǒng)有序的去對共享資源進行操作，通過互斥來保證數(shù)據(jù)的一致性。

分布式鎖的實現(xiàn)方式有多種，例如：數(shù)據(jù)庫實現(xiàn)方式、ZooKeeper實現(xiàn)方式、Redis實現(xiàn)方式等。

可能有人會問,例如synchronized、ReentrantLock不就能解決問題了嗎？為什么還要使用分布式鎖？

1.2 為什么要使用分布式鎖

為了能夠說明分布式鎖的重要性，下面通過一個電商項目中減庫存的案例來演示如果沒有使用分布式鎖會出現(xiàn)什么問題。代碼如下：

第一步：導(dǎo)入坐標(biāo)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.5.RELEASE</version>
        <relativePath/>
    </parent>
    <groupId>com.itheima</groupId>
    <artifactId>lock-test</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--集成redis-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
            <version>1.4.1.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.3</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>

</project>

第二步：配置application.yml文件

server:
  port: 8080
spring:
  redis:
    host: 68.79.63.42
    port: 26379
    password: (you password)

第三步：編寫Controller

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class StockController {
    @Autowired
    private StringRedisTemplate redisTemplate;

    @GetMapping("/stock")
    public String stock(){
        int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            stock --;
            redisTemplate.opsForValue().set("stock",stock+"");
            System.out.println("庫存扣減成功，剩余庫存：" + stock);
        }else {
            System.out.println("庫存不足?。?！");
        }
        return "OK";
    }
}

第四步：編寫啟動類

@SpringBootApplication
public class MyApplication {
    public static void main(String[] args) {
        SpringApplication.run(MyApplication.class,args);
    }
}

第五步：設(shè)置redis

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

測試方式：使用jmeter進行壓力測試，如下：

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

注：Apache JMeter是Apache組織開發(fā)的基于Java的壓力測試工具。用于對軟件做壓力測試，它最初被設(shè)計用于Web應(yīng)用測試，但后來擴展到其他測試領(lǐng)域。

查看控制臺輸出，發(fā)現(xiàn)已經(jīng)出現(xiàn)了線程并發(fā)問題，如下：

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

由于當(dāng)前程序是部署在一個Tomcat中的，即程序運行在一個jvm中，此時可以對減庫存的代碼進行同步處理，如下：

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

再次進行測試(注意恢復(fù)redis中的數(shù)據(jù))，此時已經(jīng)沒有線程并發(fā)問題，控制臺輸出如下：

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

這說明如果程序運行在一個jvm中，使用synchronized即可解決線程并發(fā)問題。

下面將程序進行集群部署（如下圖所示），并通過Nginx進行負載，再進行測試。

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

操作過程：

第一步：配置Nginx

	upstream upstream_name{
        server 127.0.0.1:8001;
        server 127.0.0.1:8002;
    }
	
	server {
        listen       8080;
        server_name  localhost;

        location / {
            proxy_pass http://upstream_name;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }
    }

第二步：修改application.yml中端口號改為8001和8002并分別啟動程序

第三步：使用jemter再次測試，可以看到又出現(xiàn)了并發(fā)問題

1.3 分布式鎖應(yīng)具有的特性

在分布式系統(tǒng)環(huán)境下，一個方法在同一時間只能被一個機器的一個線程執(zhí)行
高可用的獲取鎖與釋放鎖
高性能的獲取鎖與釋放鎖
具備可重入特性
具備鎖失效機制，防止死鎖
具備非阻塞鎖特性，即沒有獲取到鎖將直接返回獲取鎖失敗

2. 分布式鎖實現(xiàn)方案

2.1 數(shù)據(jù)庫實現(xiàn)分布式鎖

基于數(shù)據(jù)庫實現(xiàn)分布式鎖的核心思想是：在數(shù)據(jù)庫中創(chuàng)建一個表，表中包含方法名等字段，并在方法名字段上創(chuàng)建唯一索引。想要執(zhí)行某個方法，首先需要將這個方法名插入表中，成功插入則獲取鎖，執(zhí)行完成后刪除對應(yīng)的行數(shù)據(jù)釋放鎖。此種方式就是建立在數(shù)據(jù)庫唯一索引的特性基礎(chǔ)上的。

表結(jié)構(gòu)如下：

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

具體實現(xiàn)過程如下（在前面lock-test工程基礎(chǔ)上進行改造）：

第一步：在pom.xml中導(dǎo)入maven坐標(biāo)

<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.2.0</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
</dependency>

第二步：配置文件application.yml中配置mybatis-plus相關(guān)配置

server:
  port: 8002
spring:
  redis:
    host: 68.79.63.42
    port: 26379
    password: itheima123
  application:
    name: lockTest
  datasource:
      driver-class-name: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://127.0.0.1:3306/dlock
      username: root
      password: root
mybatis-plus:
  configuration:
    map-underscore-to-camel-case: false
    auto-mapping-behavior: full
    #log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
  mapper-locations: classpath*:mapper/**/*Mapper.xml

第三步：創(chuàng)建實體類

package com.itheima.entity;

import com.baomidou.mybatisplus.annotation.TableName;
import java.io.Serializable;

@TableName("mylock")
public class MyLock implements Serializable {
    private int id;
    private String methodName;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }
}

第四步：創(chuàng)建Mapper接口

package com.itheima.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.itheima.entity.MyLock;
import org.apache.ibatis.annotations.Mapper;

@Mapper
public interface MyLockMapper extends BaseMapper<MyLock> {
    public void deleteByMethodName(String methodName);
}

第五步：在resources/mapper目錄下創(chuàng)建Mapper映射文件MyLockMapper.xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.itheima.mapper.MyLockMapper">
    <delete id="deleteByMethodName" parameterType="string">
        delete from mylock where methodName = #{value}
    </delete>
</mapper>

第六步：改造StockController

@Autowired
private MyLockMapper myLockMapper;

@GetMapping("/stock")
public String stock(){
    MyLock entity = new MyLock();
    entity.setMethodName("stock");

    try {
        //插入數(shù)據(jù)，如果不拋異常則表示插入成功，即獲得鎖
        myLockMapper.insert(entity);
        
        int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            stock --;
            redisTemplate.opsForValue().set("stock",stock+"");
            System.out.println("庫存扣減成功，剩余庫存：" + stock);
        }else {
            System.out.println("庫存不足?。?！");
        }
        
        //釋放鎖
        myLockMapper.deleteByMethodName("stock");
    }catch (Exception ex){
        System.out.println("沒有獲取鎖，不能執(zhí)行減庫存操作！??！");
    }

    return "OK";
}

通過觀察控制臺輸出可以看到，使用此種方式已經(jīng)解決了線程并發(fā)問題。

注意，雖然使用數(shù)據(jù)庫方式可以實現(xiàn)分布式鎖，但是這種實現(xiàn)方式還存在如下一些問題：

1、因為是基于數(shù)據(jù)庫實現(xiàn)的，數(shù)據(jù)庫的可用性和性能將直接影響分布式鎖的可用性及性能，所以，數(shù)據(jù)庫需要雙機部署、數(shù)據(jù)同步、主備切換；

2、不具備可重入的特性，因為同一個線程在釋放鎖之前，行數(shù)據(jù)一直存在，無法再次成功插入數(shù)據(jù)，所以，需要在表中新增一列，用于記錄當(dāng)前獲取到鎖的機器和線程信息，在再次獲取鎖的時候，先查詢表中機器和線程信息是否和當(dāng)前機器和線程相同，若相同則直接獲取鎖；

3、沒有鎖失效機制，因為有可能出現(xiàn)成功插入數(shù)據(jù)后，服務(wù)器宕機了，對應(yīng)的數(shù)據(jù)沒有被刪除，當(dāng)服務(wù)恢復(fù)后一直獲取不到鎖，所以，需要在表中新增一列，用于記錄失效時間，并且需要有定時任務(wù)清除這些失效的數(shù)據(jù)；

4、不具備阻塞鎖特性，獲取不到鎖直接返回失敗，所以需要優(yōu)化獲取邏輯，循環(huán)多次去獲取。

5、在實施的過程中會遇到各種不同的問題，為了解決這些問題，實現(xiàn)方式將會越來越復(fù)雜；依賴數(shù)據(jù)庫需要一定的資源開銷，性能問題需要考慮。

2.2 ZooKeeper實現(xiàn)分布式鎖

Zookeeper的數(shù)據(jù)存儲結(jié)構(gòu)就像一棵樹，這棵樹由節(jié)點組成，這種節(jié)點叫做Znode。

Zookeeper中節(jié)點分為4種類型：

1.持久節(jié)點（PERSISTENT）

默認的節(jié)點類型。創(chuàng)建節(jié)點的客戶端與zookeeper斷開連接后，該節(jié)點依舊存在

2.持久順序節(jié)點（PERSISTENT_SEQUENTIAL）

所謂順序節(jié)點，就是在創(chuàng)建節(jié)點時，Zookeeper根據(jù)創(chuàng)建的時間順序給該節(jié)點名稱進行編號

3.臨時節(jié)點（EPHEMERAL）

和持久節(jié)點相反，當(dāng)創(chuàng)建節(jié)點的客戶端與zookeeper斷開連接后，臨時節(jié)點會被刪除

4.臨時順序節(jié)點（EPHEMERAL_SEQUENTIAL）

顧名思義，臨時順序節(jié)點結(jié)合和臨時節(jié)點和順序節(jié)點的特點：在創(chuàng)建節(jié)點時，Zookeeper根據(jù)創(chuàng)建的時間順序給該節(jié)點名稱進行編號；當(dāng)創(chuàng)建節(jié)點的客戶端與zookeeper斷開連接后，臨時節(jié)點會被刪除

Zookeeper實現(xiàn)分布式鎖的原理是基于Zookeeper的臨時順序節(jié)點，如下圖：

Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先),JAVA,分布式,java-zookeeper,java,redis

客戶端A和客戶端B爭搶分布式鎖，其實就是在/my_lock節(jié)點下創(chuàng)建臨時順序節(jié)點，這個順序節(jié)點由zk內(nèi)部自行維護一個節(jié)點序號，序號最小則表示對應(yīng)客戶端獲得鎖。

Apache Curator是一個比較完善的ZooKeeper客戶端框架，通過封裝的一套高級API 簡化了ZooKeeper的操作，其中就包括分布式鎖的實現(xiàn)。

具體操作過程如下（在前面lock-test工程基礎(chǔ)上進行改造）：

第一步：在pom.xml中導(dǎo)入maven坐標(biāo)

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.10</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.12.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.12.0</version>
</dependency>

第二步：編寫配置類

package com.itheima.config;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ZkConfig {
    @Bean
    public CuratorFramework curatorFramework(){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("localhost:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        return client;
    }
}

第三步：改造StockController

@Autowired
private CuratorFramework curatorFramework;

@GetMapping("/stock")
public String stock() {
    InterProcessMutex mutex = new InterProcessMutex(curatorFramework,"/mylock");

    try {
        //嘗試獲得鎖
        boolean locked = mutex.acquire(0, TimeUnit.SECONDS);
        if(locked){
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("庫存扣減成功，剩余庫存：" + stock);
            }else {
                System.out.println("庫存不足?。?！");
            }
            //釋放鎖
            mutex.release();
        }else{
            System.out.println("沒有獲取鎖，不能執(zhí)行減庫存操作?。?！");
        }
    }catch (Exception ex){
        System.out.println("出現(xiàn)異常?。?！");
    }

    return "OK";
}

通過觀察控制臺輸出可以看到，使用此種方式已經(jīng)解決了線程并發(fā)問題。

2.3 Redis實現(xiàn)分布式鎖

redis實現(xiàn)分布式鎖比較簡單，就是調(diào)用redis的set命令設(shè)置值，能夠設(shè)置成功則表示加鎖成功，即獲得鎖，通過調(diào)用del命令來刪除設(shè)置的鍵值，即釋放鎖。

2.3.1 版本一

加鎖命令：set lock_key lock_value NX

解鎖命令：del lock_key

Java程序：

@GetMapping("/stock")
public String stock() {
    try {
        //嘗試加鎖
        Boolean locked = redisTemplate.opsForValue().setIfAbsent("mylock", "mylock");
        if(locked){//加鎖成功
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("庫存扣減成功，剩余庫存：" + stock);
            }else {
                System.out.println("庫存不足?。?！");
            }
            //釋放鎖
            redisTemplate.delete("mylock");
        }else{
            System.out.println("沒有獲取鎖，不能執(zhí)行減庫存操作?。?！");
        }
    }catch (Exception ex){
        System.out.println("出現(xiàn)異常?。?！");
    }

    return "OK";
}

2.3.2 版本二

上面版本一的實現(xiàn)中存在一個問題，就是當(dāng)某個線程獲得鎖后程序掛掉，此時還沒來得及釋放鎖，這樣后面所有的線程都無法獲得鎖了。為了解決這個問題可以在加鎖時設(shè)置一個過期時間防止死鎖。

加鎖命令：set lock_key lock_value NX PX 5000

解鎖命令：del lock_key

Java程序：

@GetMapping("/stock")
public String stock() {
    try {
        //嘗試加鎖
        Boolean locked = redisTemplate.opsForValue().setIfAbsent("mylock", "mylock",5000,TimeUnit.MILLISECONDS);
        if(locked){//加鎖成功
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("庫存扣減成功，剩余庫存：" + stock);
            }else {
                System.out.println("庫存不足?。?！");
            }
            //釋放鎖
            redisTemplate.delete("mylock");
        }else{
            System.out.println("沒有獲取鎖，不能執(zhí)行減庫存操作?。?！");
        }
    }catch (Exception ex){
        System.out.println("出現(xiàn)異常！??！");
    }

    return "OK";
}

2.3.3 版本三

針對前面版本二還有一點需要優(yōu)化，就是加鎖和解鎖必須是同一個客戶端，所以在加鎖時可以設(shè)置當(dāng)前線程id，在釋放鎖時判斷是否為當(dāng)前線程加的鎖，如果是再釋放鎖即可。

Java程序：

@GetMapping("/stock")
public String stock() {
    try {
        String threadId = Thread.currentThread().getId()+"";
        //嘗試加鎖
        Boolean locked = redisTemplate.opsForValue().setIfAbsent("mylock",threadId,5000,TimeUnit.MILLISECONDS);
        if(locked){//加鎖成功
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("庫存扣減成功，剩余庫存：" + stock);
            }else {
                System.out.println("庫存不足?。?！");
            }
            String myValue = redisTemplate.opsForValue().get("mylock");
            if(threadId.equals(myValue)){
                //釋放鎖
                redisTemplate.delete("mylock");
            }
        }else{
            System.out.println("沒有獲取鎖，不能執(zhí)行減庫存操作！?。?);
        }
    }catch (Exception ex){
        System.out.println("出現(xiàn)異常?。?！");
    }

    return "OK";
}

3. Redisson

3.1 Redisson介紹

Redisson是架設(shè)在Redis基礎(chǔ)上的一個Java駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory Data Grid）。充分的利用了Redis鍵值數(shù)據(jù)庫提供的一系列優(yōu)勢，基于Java實用工具包中常用接口，為使用者提供了一系列具有分布式特性的常用工具類。使得原本作為協(xié)調(diào)單機多線程并發(fā)程序的工具包獲得了協(xié)調(diào)分布式多機多線程并發(fā)系統(tǒng)的能力，大大降低了設(shè)計和研發(fā)大規(guī)模分布式系統(tǒng)的難度。同時結(jié)合各富特色的分布式服務(wù)，更進一步簡化了分布式環(huán)境中程序相互之間的協(xié)作。

Redisson已經(jīng)內(nèi)置提供了基于Redis的分布式鎖實現(xiàn)，此種方式是我們推薦的分布式鎖使用方式。

Redisson的maven坐標(biāo)如下：

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.10.1</version>
</dependency>

3.2 Redisson分布式鎖使用方式

第一步：在pom.xml中導(dǎo)入redisson的maven坐標(biāo)

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.10.1</version>
</dependency>

第二步：編寫配置類

package com.itheima.config;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {
    @Value("${spring.redis.host}")
    private String host;
    @Value("${spring.redis.port}")
    private String port;
    @Value("${spring.redis.password}")
    private String password;

    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://" + host + ":" + port);
        config.useSingleServer().setPassword(password);
        final RedissonClient client = Redisson.create(config);
        return client;
    }
}

第三步：改造Controller

@Autowired
private RedissonClient redissonClient;

@GetMapping("/stock")
public String stock() {
    //獲得分布式鎖對象，注意，此時還沒有加鎖成功
    RLock lock = redissonClient.getLock("mylock");
    try {
        //嘗試加鎖，如果加鎖成功則后續(xù)程序繼續(xù)執(zhí)行，如果加鎖不成功則阻塞等待
        lock.lock(5000,TimeUnit.MILLISECONDS);

        int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            stock --;
            redisTemplate.opsForValue().set("stock",stock+"");
            System.out.println("庫存扣減成功，剩余庫存：" + stock);
        }else {
            System.out.println("庫存不足！?。?);
        }
    }catch (Exception ex){
        System.out.println("出現(xiàn)異常?。?！");
    }finally {
        //解鎖
        lock.unlock();
    }

    return "OK";
}

3.3 Lua腳本

3.3.1 Lua簡介

Lua 是一種輕量小巧的腳本語言，用標(biāo)準(zhǔn)C語言編寫并以源代碼形式開放，其設(shè)計目的是為了嵌入應(yīng)用程序中，從而為應(yīng)用程序提供靈活的擴展和定制功能。

從Redis2.6.0版本開始提供了EVAL 和 EVALSHA 命令，這兩個命令可以執(zhí)行 Lua 腳本。

3.3.2 Redis中使用Lua的好處

Redis中使用Lua的好處：

減少網(wǎng)絡(luò)開銷。可以將多個請求通過腳本的形式一次發(fā)送，減少網(wǎng)絡(luò)時延
原子操作。redis會將整個腳本作為一個整體執(zhí)行，中間不會被其他命令插入。因此在編寫腳本的過程中無需擔(dān)心會出現(xiàn)競態(tài)條件，無需使用事務(wù)
復(fù)用。客戶端發(fā)送的腳本會永久存在redis中，這樣，其他客戶端可以復(fù)用這一腳本而不需要使用代碼完成相同的邏輯

3.3.3 如何在Redis中使用Lua

在redis中使用Lua腳本主要有三個命令

eval
evalsha
script load

eval用來直接執(zhí)行l(wèi)ua腳本，使用方式如下：

EVAL script numkeys key [key ...] arg [arg ...]

key代表要操作的redis key

arg可以傳自定義的參數(shù)

numkeys用來確定key有幾個

script就是你寫的lua腳本

lua腳本里面使用KEYS[1]和ARGV[1]來獲取傳遞的第一個key和第一個arg，后面以此類推

舉例：

eval "return redis.call('set',KEYS[1],ARGV[1])" 1 city beijing

eval "return redis.call('set','name','xiaoming')" 0

eval "return redis.call('del',KEYS[1])" 1 city

eval "return redis.call('get',KEYS[1])" 1 name

eval "if (redis.call('exists', KEYS[1]) == 0) then redis.call('set', KEYS[2], ARGV[1]); redis.call('expire', KEYS[2], ARGV[2]);return nil;end;" 2 citys city beijing 5000

在用eval命令的時候，可以注意到每次都要把執(zhí)行的腳本發(fā)送過去，這樣勢必會有一定的網(wǎng)絡(luò)開銷，所以redis對lua腳本做了緩存，通過script load 和 evalsha實現(xiàn)：

SCRIPT LOAD script

EVALSHA sha1 numkeys key [key ...] arg [arg ...]

舉例：文章來源地址http://www.zghlxwxcb.cn/news/detail-771938.html

script load "return redis.call('get',KEYS[1]);"

evalsha 0e11c9f252fd76115c38403ce6095872b8c70580 1 name

到了這里，關(guān)于Java中利用Redis,ZooKeeper,數(shù)據(jù)庫等實現(xiàn)分布式鎖(遙遙領(lǐng)先)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！