一、背景
线上部署了两台服务器，通过nginx轮询的方式进行负载均衡。但是这样存在一个问题同一个用户的session共享问题。你或许会说，使用ipHash模式就可以解决session共享的问题，是的确实可以解决这个问题，但是同样也会带来另外一个问题，就是一台服务器很繁忙，另外一台服务器闲置的情况。所以为了避免服务器闲置的现象，我们采用了ip轮询和共享session存入redis的解决方案。不过今天要讲的主题不是这个，我们在单机测试的时候完全运行正常，但是部署了到正式的环境的时候出现了客户支付金额对应不上的情况。后来经过排查发现是并发问题，虽然单机做了并发处理，但是如果有多台服务器的时候就会出现服务器之间的并发问题。所以需要引入分布式锁，解决多服务器间的。
二、分布式锁的实现
1. jar包的引入
<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>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>net.javacrumbs.shedlock</groupId>
<artifactId>shedlock-provider-redis-spring</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.0</version>
</dependency>
<dependency>
<groupId>net.javacrumbs.shedlock</groupId>
<artifactId>shedlock-spring</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!-- swagger -->
<!-- https://mvnrepository.com/artifact/io.springfox/springfox-swagger-ui -->
<dependency>
<groupId>com.github.xiaoymin</groupId>
<artifactId>swagger-bootstrap-ui</artifactId>
<version>1.9.6</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<version>1.9.2</version>
</dependency>
2. redis的配置
-
#redis
redis.host=192.168.1.6
redis.password=
redis.port=6379
redis.taskScheduler.poolSize=100
redis.taskScheduler.defaultLockMaxDurationMinutes=10
redis.default.timeout=10
redisCache.expireTimeInMilliseconds=1200000
-
package com.example.redis_demo_limit.redis;
import io.lettuce.core.ClientOptions;
import io.lettuce.core.resource.ClientResources;
import io.lettuce.core.resource.DefaultClientResources;
import net.javacrumbs.shedlock.core.LockProvider;
import net.javacrumbs.shedlock.provider.redis.spring.RedisLockProvider;
import net.javacrumbs.shedlock.spring.ScheduledLockConfiguration;
import net.javacrumbs.shedlock.spring.ScheduledLockConfigurationBuilder;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisPassword;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.core.RedisTemplate;
import java.time.Duration;
@Configuration
public class RedisConfig {
@Value("${redis.host}")
private String redisHost;
@Value("${redis.port}")
private int redisPort;
@Value("${redis.password}")
private String password;
@Value("${redis.taskScheduler.poolSize}")
private int tasksPoolSize;
@Value("${redis.taskScheduler.defaultLockMaxDurationMinutes}")
private int lockMaxDuration;
@Bean(destroyMethod = "shutdown")
ClientResources clientResources() {
return DefaultClientResources.create();
}
@Bean
public RedisStandaloneConfiguration redisStandaloneConfiguration() {
RedisStandaloneConfiguration redisStandaloneConfiguration =
new RedisStandaloneConfiguration(redisHost, redisPort);
if (password != null && !password.trim().equals("")) {
RedisPassword redisPassword = RedisPassword.of(password);
redisStandaloneConfiguration.setPassword(redisPassword);
}
return redisStandaloneConfiguration;
}
@Bean
public ClientOptions clientOptions() {
return ClientOptions.builder()
.disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
.autoReconnect(true).build();
}
@Bean
LettucePoolingClientConfiguration lettucePoolConfig(ClientOptions options, ClientResources dcr) {
return LettucePoolingClientConfiguration.builder().poolConfig(new GenericObjectPoolConfig())
.clientOptions(options).clientResources(dcr).build();
}
@Bean
public RedisConnectionFactory connectionFactory(
RedisStandaloneConfiguration redisStandaloneConfiguration,
LettucePoolingClientConfiguration lettucePoolConfig) {
return new LettuceConnectionFactory(redisStandaloneConfiguration, lettucePoolConfig);
}
@Bean
@ConditionalOnMissingBean(name = "redisTemplate")
@Primary
public RedisTemplate<Object, Object> redisTemplate(
RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<Object, Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
return template;
}
@Bean
public LockProvider lockProvider(RedisConnectionFactory connectionFactory) {
return new RedisLockProvider(connectionFactory);
}
@Bean
public ScheduledLockConfiguration taskSchedulerLocker(LockProvider lockProvider) {
return ScheduledLockConfigurationBuilder.withLockProvider(lockProvider)
.withPoolSize(tasksPoolSize).withDefaultLockAtMostFor(Duration.ofMinutes(lockMaxDuration))
.build();
}
}
-
package com.example.redis_demo_limit.redis;
public interface DataCacheRepository<T> {
boolean add(String collection, String hkey, T object, Long timeout);
boolean delete(String collection, String hkey);
T find(String collection, String hkey, Class<T> tClass);
Boolean isAvailable();
/**
* redis 加锁
*
* @param key
* @param second
* @return
*/
Boolean lock(String key, String value, Long second);
Object getValue(String key);
/**
* redis 解锁
*
* @param key
* @return
*/
void unLock(String key);
void setIfAbsent(String key, long value, long ttl);
void increment(String key);
Long get(String key);
void set(String key, long value, long ttl);
void set(Object key, Object value, long ttl);
Object getByKey(String key);
void getLock(String key, String clientID) throws Exception;
void releaseLock(String key, String clientID);
boolean hasKey(String key);
}
实现类
package com.example.redis_demo_limit.redis;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
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.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.support.atomic.RedisAtomicLong;
import org.springframework.stereotype.Repository;
import java.time.Duration;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;
@Slf4j
@Repository
public class CacheRepository<T> implements com.example.redis_demo_limit.redis.DataCacheRepository<T> {
private static final ObjectMapper OBJECT_MAPPER;
private static final TimeZone DEFAULT_TIMEZONE = TimeZone.getTimeZone("UTC");
static {
OBJECT_MAPPER = new ObjectMapper();
OBJECT_MAPPER.setTimeZone(DEFAULT_TIMEZONE);
}
Logger logger = LoggerFactory.getLogger(CacheRepository.class);
@Autowired
RedisTemplate template; // and we're in business
@Value("${redis.default.timeout}00")
Long defaultTimeOut;
public boolean addPermentValue(String collection, String hkey, T object) {
try {
String jsonObject = OBJECT_MAPPER.writeValueAsString(object);
template.opsForHash().put(collection, hkey, jsonObject);
return true;
} catch (Exception e) {
logger.error("Unable to add object of key {} to cache collection '{}': {}", hkey, collection,
e.getMessage());
return false;
}
}
@Override
public boolean add(String collection, String hkey, T object, Long timeout) {
Long localTimeout;
if (timeout == null) {
localTimeout = defaultTimeOut;
} else {
localTimeout = timeout;
}
try {
String jsonObject = OBJECT_MAPPER.writeValueAsString(object);
template.opsForHash().put(collection, hkey, jsonObject);
template.expire(collection, localTimeout, TimeUnit.SECONDS);
return true;
} catch (Exception e) {
logger.error("Unable to add object of key {} to cache collection '{}': {}", hkey, collection,
e.getMessage());
return false;
}
}
@Override
public boolean delete(String collection, String hkey) {
try {
template.opsForHash().delete(collection, hkey);
return true;
} catch (Exception e) {
logger.error("Unable to delete entry {} from cache collection '{}': {}", hkey, collection,
e.getMessage());
return false;
}
}
@Override
public T find(String collection, String hkey, Class<T> tClass) {
try {
String jsonObj = String.valueOf(template.opsForHash().get(collection, hkey));
return OBJECT_MAPPER.readValue(jsonObj, tClass);
} catch (Exception e) {
if (e.getMessage() == null) {
logger.error("Entry '{}' does not exist in cache", hkey);
} else {
logger.error("Unable to find entry '{}' in cache collection '{}': {}", hkey, collection,
e.getMessage());
}
return null;
}
}
@Override
public Boolean isAvailable() {
try {
return template.getConnectionFactory().getConnection().ping() != null;
} catch (Exception e) {
logger.warn("Redis server is not available at the moment.");
}
return false;
}
@Override
public Boolean lock(String key, String value, Long second) {
Boolean absent = template.opsForValue().setIfAbsent(key, value, second, TimeUnit.SECONDS);
return absent;
}
@Override
public Object getValue(String key) {
return template.opsForValue().get(key);
}
@Override
public void unLock(String key) {
template.delete(key);
}
@Override
public void increment(String key) {
RedisAtomicLong counter = new RedisAtomicLong(key, template.getConnectionFactory());
counter.incrementAndGet();
}
@Override
public void setIfAbsent(String key, long value, long ttl) {
ValueOperations<String, Object> ops = template.opsForValue();
ops.setIfAbsent(key, value, Duration.ofSeconds(ttl));
}
@Override
public Long get(String key) {
RedisAtomicLong counter = new RedisAtomicLong(key, template.getConnectionFactory());
return counter.get();
}
@Override
public void set(String key, long value, long ttl) {
RedisAtomicLong counter = new RedisAtomicLong(key, template.getConnectionFactory());
counter.set(value);
counter.expire(ttl, TimeUnit.SECONDS);
}
@Override
public void set(Object key, Object value, long ttl) {
template.opsForValue().set(key, value, ttl, TimeUnit.SECONDS);
}
@Override
public Object getByKey(String key) {
return template.opsForValue().get(key);
}
@Override
public void getLock(String key, String clientID) throws Exception {
Boolean lock = false;
// 重试3次,每间隔1秒重试1次
for (int j = 0; j <= 3; j++) {
lock = lock(key, clientID, 10L);
if (lock) {
log.info("获得锁》》》" + key);
break;
}
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
log.error("线程休眠异常", e);
break;
}
}
// 重试3次依然没有获取到锁，那么返回服务器繁忙,请稍后重试
if (!lock) {
throw new Exception("服务繁忙");
}
}
@Override
public void releaseLock(String key, String clientID) {
if (clientID.equals(getByKey(key))) {
unLock(key);
}
}
@Override
public boolean hasKey(String key) {
return template.hasKey(key);
}
}
三、使用方法
import com.example.redis_demo_limit.annotation.LimitedAccess;
import com.example.redis_demo_limit.redis.DataCacheRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.UUID;
@Slf4j
@RestController
@RequestMapping("/redis")
public class RedisController {
private static final String KEY = "key";
@Resource
private DataCacheRepository dataCacheRepository;
@LimitedAccess(frequency = 1,second = 1)
@PostMapping("/add")
public String add(String str){
dataCacheRepository.set("str","add success",200L);
return "success";
}
//分布式锁使用示例
@PostMapping("/pay")
public String pay(String userName,Integer account){
String clientID = UUID.randomUUID().toString();
//设置锁的过期时间，避免死锁
Boolean lock = dataCacheRepository.lock(userName, clientID, 6000L);
if(!lock){
log.info("未获取到锁{}", userName);
return "程序繁忙，请稍后再试！";
}
try {
//等待5s，方便测试
Thread.sleep(5000);
if(dataCacheRepository.hasKey(KEY)){
Long aLong = dataCacheRepository.get(KEY);
dataCacheRepository.set(KEY,aLong+account,-1);
return account+aLong+"";
}else {
dataCacheRepository.set(KEY,account,-1);
return account+"";
}
} catch (InterruptedException e) {
log.error(e.getMessage(),e);
return "程序运行异常，请联系管理员！";
} finally {
if (clientID.equals(dataCacheRepository.getByKey(userName))) {
log.info("finally删除锁{}", userName);
dataCacheRepository.unLock(userName);
}
}
}
}
四、结果测试
-
先在8082点击pay
![]()
-
再在8081点击pay
![]()
从结果来看，分布式锁起到了作用。很简单吧，还不来试一下！
从结果来看，分布式锁起到了作用。很简单吧，还不来试一下！
