分布式锁是控制分布式系统之间同步访问共享资源的一种方式。在分布式系统中,常常需要协调他们的动作。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源,那么访问这些资源的时候,往往需要互斥来防止彼此干扰来保证一致性,在这种情况下,便需要使用到分布式锁。
数据库的悲观锁和乐观锁也能保证不同主机共享数据的一致性。但是却存在以下问题:
本文使用Redis分布式锁的开源项目redisson做分布式锁的简单实现。
redisson官网:分布式锁和同步器
代码可以在SpringBoot组件化构建https://www.pomit.cn/java/spring/springcloud.html中的LockRedis、LockSupport和LockTest组件中查看,并下载。LockSupport和LockTest是配合LockRedis的测试项目。
**如果大家正在寻找一个java的学习环境,或者在开发中遇到困难,可以<a
href="https://jq.qq.com/?_wv=1027&k=52sgH1J"
target="_blank">
加入我们的java学习圈,点击即可加入
</a>
,共同学习,节约学习时间,减少很多在学习中遇到的难题。**
首先是必然要安装redis的。其次,我们要了解redisson锁的种类,为了配合锁的使用,使用了LockSupport和LockTest两个项目分别做商品服务和客户端测试,多个项目使用Consul做服务注册与发现。这里会简单介绍下LockSupport、LockRedis和LockTest三个项目是如何配合使用分布式锁的。
redisson为我们提供了很多种锁,如:
可重入锁(Reentrant Lock) :基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。
公平锁(Fair Lock) :基于Redis的Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。同时还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。它保证了当多个Redisson客户端线程同时请求加锁时,优先分配给先发出请求的线程。
联锁(MultiLock) :基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁,每个RLock对象实例可以来自于不同的Redisson实例。
红锁(RedLock) :基于Redis的Redisson红锁RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁,每个RLock对象实例可以来自于不同的Redisson实例。
读写锁(ReadWriteLock) :基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。
信号量(Semaphore) :基于Redis的Redisson的分布式信号量(Semaphore)Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。同时还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。
可过期性信号量(PermitExpirableSemaphore) :基于Redis的Redisson可过期性信号量(PermitExpirableSemaphore)是在RSemaphore对象的基础上,为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识,释放时只能通过提交这个ID才能释放。它提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。
闭锁(CountDownLatch) :基于Redisson的Redisson分布式闭锁(CountDownLatch)Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。
锁的种类虽多,但是使用方法都是类似的,本文就选择公平锁来测试使用。
LockSupport代码可以在SpringBoot组件化构建https://www.pomit.cn/java/spring/springcloud.html中的LockSupport组件中下载,这里就只说LockSupport的功能。
LockSupport只是个简单的SpringBoot项目,使用Spring Data Jpa做数据库操作,开放以下接口做服务:
LockRedis代码可以在SpringBoot组件化构建https://www.pomit.cn/java/spring/springcloud.html中的LockRedis组件中下载。
第三章节将详细介绍LockRedis项目。
LockTest代码可以在SpringBoot组件化构建https://www.pomit.cn/java/spring/springcloud.html中的LockTest组件中下载,这里就只说LockSupport的功能。
LockTest只是个简单的SpringBoot项目,使用Feign请求LockRedis来测试分布式锁的使用。
下面详细介绍LockRedis是如何使用redisson做分布式锁的。
需要引入数据库相关jar、jpa、spring-boot-starter-data-redis、redisson-spring-boot-starter;
因为使用了consul做服务注册发现,需要引入spring-cloud-starter-consul-discovery和spring-cloud-starter-openfeign。
依赖如下:
<?xml version="1.0"?>
<project
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>cn.pomit</groupId>
<artifactId>springcloudwork</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<artifactId>LockRedis</artifactId>
<name>LockRedis</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
<maven-jar-plugin.version>2.6</maven-jar-plugin.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-consul-discovery</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
<!-- redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.11.0</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.34</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-dbcp2</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
</dependencies>
</project>
父模块pom文件可以在https://www.pomit.cn/spring/SpringCloudWork/pom.xml获取。
这里使用yaml文件写配置,配置文件application.yml:
application.yml:
server:
port: 8018
useLock: true
spring:
application:
name: lockService
cloud:
consul:
host: 127.0.0.1
port: 8500
discovery:
prefer-ip-address: true
healthCheckPath: /consul/health
redis:
database: 0
port: 6379
host: 192.168.99.100
lettuce:
pool:
max-idle: 8
min-idle: 5
max-active: 10
max-wait: 1000
timeout: 1000
datasource:
type: org.apache.commons.dbcp2.BasicDataSource
dbcp2:
max-wait-millis: 60000
min-idle: 20
initial-size: 2
connection-properties: characterEncoding=utf8
validation-query: SELECT 1
test-while-idle: true
test-on-borrow: true
test-on-return: false
driverClassName: com.mysql.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/boot?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC
username: cff
password: 123456
logging:
level:
root: INFO
这里,应用名称是lockService,在8018端口监听。
这里使用@FeignClient来请求lockSupport项目,获取商品信息或者消费商品。
GoodsInfoService :
package cn.pomit.springbootwork.lock.redis.inter;
import java.util.List;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import cn.pomit.springbootwork.lock.redis.domain.TGoodInfo;
@FeignClient("lockSupport")
public interface GoodsInfoService {
@RequestMapping(method = RequestMethod.GET, value = "/goods/take", consumes = "application/json")
public TGoodInfo getGoodsInfo(@RequestParam("goodId") Integer goodId);
@RequestMapping(method = RequestMethod.GET, value = "/goods/page", consumes = "application/json")
public List<TGoodInfo> getGoodsList(@RequestParam("page") Integer page, @RequestParam("size") Integer size);
@RequestMapping(method = RequestMethod.GET, value = "/goods/consume", consumes = "application/json")
public Integer consume(@RequestParam("goodId") Integer goodId, @RequestParam("num") Integer num, @RequestParam("serverId") Integer serverId);
}
消费的业务逻辑只是简单的处理商品,并调用GoodsInfoService 进行商品信息服务请求。
UserConsumeService:
package cn.pomit.springbootwork.lock.redis.service;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import cn.pomit.springbootwork.lock.redis.dao.TUserGoodDao;
import cn.pomit.springbootwork.lock.redis.domain.TGoodInfo;
import cn.pomit.springbootwork.lock.redis.domain.TUserGood;
import cn.pomit.springbootwork.lock.redis.dto.ResultModel;
import cn.pomit.springbootwork.lock.redis.inter.ConsumeService;
import cn.pomit.springbootwork.lock.redis.inter.GoodsInfoService;
@Service("UserConsumeService")
public class UserConsumeService implements ConsumeService {
protected Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
GoodsInfoService goodsInfoService;
@Autowired
TUserGoodDao tUserGoodDao;
@Value("${server.port}")
private Integer serverId;
public ResultModel goodList(Integer page, Integer size) {
List<TGoodInfo> pageList = goodsInfoService.getGoodsList(page, size);
return ResultModel.ok(pageList);
}
public ResultModel goodDetail(Integer goodId) {
return ResultModel.ok(goodsInfoService.getGoodsInfo(goodId));
}
public ResultModel consume(Integer goodId, Integer num, String userName) {
log.info("开始消费:userName:{};goodId:{},num:{}", userName, goodId, num);
TGoodInfo tGoodInfo = goodsInfoService.getGoodsInfo(goodId);
log.info("消费前:goodId:{}剩余:{}", goodId, tGoodInfo.getGoodNum());
Integer ret = goodsInfoService.consume(goodId, num, serverId);
log.info("消费结果:ret:{};", ret);
if (ret > 0) {
log.info("保存用户消费信息");
TUserGood tUserGood = new TUserGood();
tUserGood.setConsumeNum(num);
tUserGood.setGoodId(goodId);
tUserGood.setUserName(userName);
tUserGoodDao.save(tUserGood);
return ResultModel.ok();
}
return ResultModel.error("消费失败!");
}
}
同时,这个业务逻辑实现了ConsumeService接口:
ConsumeService:
package cn.pomit.springbootwork.lock.redis.inter;
import cn.pomit.springbootwork.lock.redis.dto.ResultModel;
public interface ConsumeService {
public ResultModel goodList(Integer page, Integer size);
public ResultModel goodDetail(Integer goodId);
public ResultModel consume(Integer goodId, Integer num, String userName);
}
实现这个接口,是为了使用代理模式。
DistributedLocker类实现基于redis的分布式锁。
RLock lock = redissonClient.getFairLock
;lock.tryLock
成功获取到锁之后,调用LockWorker的invoke方法,执行传入的LockWorker对象。lock.unlock()
释放锁。DistributedLocker :
package cn.pomit.springbootwork.lock.redis.lock;
import java.util.concurrent.TimeUnit;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import cn.pomit.springbootwork.lock.redis.dto.ResultModel;
import cn.pomit.springbootwork.lock.redis.inter.LockWorker;
public class DistributedLocker {
protected Logger log = LoggerFactory.getLogger(this.getClass());
public final static String LOCKER_PREFIX = "lock:";
public final static String LOCKER_GOODS_CONSUME = "GOODS_CONSUME";
public final static Long LOCKER_WAITE_TIME = 10l;
public final static Long LOCKER_LOCK_TIME = 100l;
@Value("${server.port}")
private Integer serverId;
@Value("${server.useLock}")
private boolean useLock = true;
/**
* The Redisson client.
*/
@Autowired
protected RedissonClient redissonClient;
public ResultModel tryLock(LockWorker lockWorker) {
try {
if (useLock) {
log.info("当前server 节点为:{}", serverId);
log.info("开始获取lock key:{}", LOCKER_PREFIX + LOCKER_GOODS_CONSUME);
RLock lock = redissonClient.getFairLock(LOCKER_PREFIX + LOCKER_GOODS_CONSUME);
log.info("创建lock key:{},尝试lock", LOCKER_PREFIX + LOCKER_GOODS_CONSUME);
// (公平锁)最多等待10秒,锁定后经过lockTime秒后自动解锁
boolean success = lock.tryLock(LOCKER_WAITE_TIME, LOCKER_LOCK_TIME, TimeUnit.SECONDS);
if (success) {
try {
log.info("成功到获取lock key:{}", LOCKER_PREFIX + LOCKER_GOODS_CONSUME);
return lockWorker.invoke();
} finally {
log.info("释放lock key:{}", LOCKER_PREFIX + LOCKER_GOODS_CONSUME);
lock.unlock();
}
}
log.info("获取lock key:{}失败", LOCKER_PREFIX + LOCKER_GOODS_CONSUME);
return ResultModel.error("获取分布式锁失败!");
} else {
log.info("当前server 节点为:{}", serverId);
log.info("当前server 没有使用分布式锁:{}");
return lockWorker.invoke();
}
} catch (Exception e) {
log.error("获取lock key异常", e);
return ResultModel.error("获取分布式锁过程异常!");
}
}
public Integer getServerId() {
return serverId;
}
public void setServerId(Integer serverId) {
this.serverId = serverId;
}
}
为了避免对代码侵入性太强,我这里使用代理模式,使用LockWorker来操作消费业务。这样,如果想去掉分布式锁的代码,只需注掉代理即可。
UserConsumeServiceProxy:
package cn.pomit.springbootwork.lock.redis.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import cn.pomit.springbootwork.lock.redis.dto.ResultModel;
import cn.pomit.springbootwork.lock.redis.inter.ConsumeService;
import cn.pomit.springbootwork.lock.redis.inter.LockWorker;
import cn.pomit.springbootwork.lock.redis.lock.DistributedLocker;
@Service("consumeService")
public class UserConsumeServiceProxy extends DistributedLocker implements ConsumeService {
@Autowired
UserConsumeService userConsumeService;
public ResultModel goodList(Integer page, Integer size) {
return userConsumeService.goodList(page, size);
}
public ResultModel goodDetail(Integer goodId) {
return userConsumeService.goodDetail(goodId);
}
public ResultModel consume(Integer goodId, Integer num, String userName) {
LockWorker lockWorker = new LockWorker() {
@Override
public ResultModel invoke() throws Exception {
log.info("开始远程消费:userName:{};goodId:{},num:{}", userName, goodId, num);
return userConsumeService.consume(goodId, num, userName);
}
};
return tryLock(lockWorker);
}
}
LockWorker 接口:
package cn.pomit.springbootwork.lock.redis.inter;
import cn.pomit.springbootwork.lock.redis.dto.ResultModel;
public interface LockWorker {
ResultModel invoke() throws Exception;
}
开放web接口,给模拟的客户端LockTest测试使用。
LockRedisRest:
package cn.pomit.springbootwork.lock.redis.web;
import java.security.Principal;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import cn.pomit.springbootwork.lock.redis.dto.ResultModel;
import cn.pomit.springbootwork.lock.redis.inter.ConsumeService;
@RestController
@RequestMapping("/lock")
public class LockRedisRest {
private Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
ConsumeService consumeService;
@RequestMapping(value = "/goodList")
public ResultModel goodList(@RequestParam(value = "page", required = false) Integer page,
@RequestParam(value = "size", required = false) Integer size) {
return consumeService.goodList(page, size);
}
@RequestMapping(value = "/goodDetail")
public ResultModel goodDetail(@RequestParam("goodId") Integer goodId) {
return consumeService.goodDetail(goodId);
}
@RequestMapping(value = "/consume")
public ResultModel consume(@RequestParam("goodId") Integer goodId, @RequestParam("num") Integer num,
Principal principal) {
String userName = "test";
if (principal != null && !StringUtils.isEmpty(principal.getName())) {
userName = principal.getName();
}
log.info("收到消费请求:userName:{};goodId:{}", userName, goodId);
return consumeService.consume(goodId, num, userName);
}
}
LockRedisApplication:
package cn.pomit.springbootwork.lock.redis;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;
@EnableFeignClients
@EnableDiscoveryClient
@SpringBootApplication
public class LockRedisApplication {
public static void main(String[] args) {
SpringApplication.run(LockRedisApplication.class, args);
}
@Bean
@LoadBalanced
RestTemplate restTemplate() {
return new RestTemplate();
}
}
package cn.pomit.springbootwork.lock.redis.dao;
import org.springframework.data.jpa.repository.JpaRepository;
import cn.pomit.springbootwork.lock.redis.domain.TUserGood;
public interface TUserGoodDao extends JpaRepository<TUserGood, Integer> {
}
这个是使用consul做注册中心才需要的。
HealthWeb:
package cn.pomit.springbootwork.lock.redis.web;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/consul")
public class HealthWeb {
@RequestMapping(value = "health", method = { RequestMethod.GET })
public String health() {
return "check health";
}
}
TUserGood:
package cn.pomit.springbootwork.lock.redis.domain;
import javax.persistence.Table;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Column;
@Entity
@Table(name = "t_user_good")
public class TUserGood {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private int id;
@Column(name = "good_id")
private int goodId;
@Column(name = "consume_num")
private int consumeNum;
@Column(name = "user_name")
private String userName;
public void setId(int id) {
this.id = id;
}
public int getId() {
return id;
}
public void setGoodId(int goodId) {
this.goodId = goodId;
}
public int getGoodId() {
return goodId;
}
public void setConsumeNum(int consumeNum) {
this.consumeNum = consumeNum;
}
public int getConsumeNum() {
return consumeNum;
}
public void setUserName(String userName) {
this.userName = userName;
}
public String getUserName() {
return userName;
}
}
TGoodInfo:
package cn.pomit.springbootwork.lock.redis.domain;
import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;
public class TGoodInfo {
private Integer id;
private String goodName;
private String goodDesc;
@JsonFormat(pattern = "yyyy-MM-dd", timezone = "GMT+8")
private Date createTime;
private int goodNum;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public void setGoodName(String goodName) {
this.goodName = goodName;
}
public String getGoodName() {
return goodName;
}
public void setGoodDesc(String goodDesc) {
this.goodDesc = goodDesc;
}
public String getGoodDesc() {
return goodDesc;
}
public void setCreateTime(Date createTime) {
this.createTime = createTime;
}
public Date getCreateTime() {
return createTime;
}
public int getGoodNum() {
return goodNum;
}
public void setGoodNum(int goodNum) {
this.goodNum = goodNum;
}
}