业务场景多了，就应该把场景代码工具化，减少重复代码；趁周六在这里总结一下java实现的redis分布式锁代码；
使用的技术点：
1、redis函数setnx；
2、redis监控函数watch；
3、String.intern在jvm的内存操作，和String.intern的替代方案：guava；
4、@FunctionalInterface函数接口和lambda表达式的应用。
5、RedisTemplate;尽量使用RedisTemplate操作redis；直接操作jedispool人多的时候会乱。
reidis锁操作类
package com.framework.cache;
import java.util.Random;
import com.framework.cache.execption.RedisExecption;
/**
* redis锁操作类
* @author lyq
*
*/
public class RedisLock {
public static final String LOCKED = "TRUE";
public static final long MILLI_NANO_CONVERSION = 1000000L;
public static final long DEFAULT_TIME_OUT = 1000L;
public static final Random RANDOM = new Random();
public static final int EXPIRE = 1200;
private String key;
private String lockId;
private boolean locked = false;
public RedisLock(String key) {
lockId = System.currentTimeMillis()+"";
this.key = String.format("lock:%s", key);
}
/**
* 加锁
* @param expireSeconds 超时时间，单位：秒
* @return 获取锁成功返回true；超时返回false；其它报异常
*/
public boolean lock(int expireSeconds) {
long nano = System.nanoTime();
long timeout= expireSeconds * 1000000000L;
try {
// 当前任务超时的时候，可能会被其它任务抢到锁，该任务会误删其它任务的锁；所以加入判断。
if (RedisHelper.getExpire(this.key) == -1) {
RedisHelper.expire(this.key, expireSeconds);
}
while (System.nanoTime() - nano < timeout) {
if (RedisHelper.setnx(this.key, lockId)) {
RedisHelper.expire(this.key, expireSeconds);
this.locked = true;
return this.locked;
}
Thread.sleep(3L, RANDOM.nextInt(500));
}
} catch (Exception e) {
throw new RedisExecption("Locking error", e);
}
return false;
}
public boolean lock() {
return lock(2);
}
/**
* 解锁
*/
@SuppressWarnings("unchecked")
public void unlock() {
if (this.locked) {
String _lockId = this.lockId;
RedisHelper.getRedisTemplate().execute(connection->{
byte[] keyByte = RedisHelper.serialize(key);
//开启watch之后，如果key的值被修改，则事务失败，exec方法返回null
connection.watch(keyByte);
String startTime = RedisHelper.unserialize(connection.get(keyByte), String.class);
if (_lockId.equals(startTime)) { // 当前任务超时的时候，可能会被其它任务抢到锁，该任务会误删其它任务的锁；所以加入判断。
connection.multi();
connection.del(keyByte);
connection.exec();
}
connection.unwatch();
return true;
},true);
}
}
public String getLockKey(){
return this.key;
}
}
redis对外帮助类
package com.framework.cache;
import java.lang.reflect.Type;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import com.framework.cache.execption.RedisExecption;
import com.framework.cache.execption.RedisLockTimeException;
import com.framework.cache.lock.LockCallback;
import com.framework.utils.NumUtil;
import com.google.common.collect.Interner;
import com.google.common.collect.Interners;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import redis.clients.jedis.JedisPoolConfig;
/**
* redis对外静态帮助类
* @author lyq
*
*/
public class RedisHelper {
public static RedisTemplate redisTemplate = null;
private static Interner<Object> pool = Interners.newWeakInterner();
private static final Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd hh:mm:ss").create();
private static final String IGNORE_EXEC_KEY = "_ignoreExec";
/**
* 加锁执行
* @param callBack 回调函数
* @param lockKey
* @param expireSeconds 执行超时时间，单位:秒
* @return 返回callback回调函数执行结果
*/
public static <T> T lockExec(LockCallback<T> callBack, String lockKey, Integer expireSeconds) {
/**
* redisLock其实就是规定时间内一直循环网络请求redis资源；
* 为了防止并发的时候，太多线程循环抢redis锁；导致资源不足，需要对字符串加锁；
* 字符加锁,需要使用intern；而1.7以后，String.intern的对象会一直存在jvm内存里面，大量使用会导致内存异常；替代方案就是使用guava的Interners.newWeakInterner()代替String.intern;
* synchronized不支持过期，如果需要加过期时间，则使用ReentrantLock 的 tryLock(long timeout, TimeUnit unit)方法。
*/
synchronized (pool.intern(lockKey)) {
RedisLock lock = new RedisLock(lockKey);
try {
if(lock.lock(expireSeconds)){
return callBack.exec();
}else{
throw new RedisLockTimeException(lockKey,expireSeconds);
}
} catch (Exception e) {
throw new RedisExecption(e);
} finally {
lock.unlock();
}
}
}
/**
* 获取超时时间
* @param key
* @return 单位秒
*/
public static Long getExpire(String key){
byte[] rawKey = RedisHelper.serialize(key);
return (Long) redisTemplate.execute(connection -> connection.ttl(rawKey), true);
}
/**
* 设置超时时间
* @param key
* @param seconds 单位秒
* @return
*/
public static Boolean expire(String key,Integer seconds){
byte[] rawKey = RedisHelper.serialize(key);
return (Boolean) redisTemplate.execute(connection -> connection.expire(rawKey,seconds), true);
}
/**
* 尝试加锁并执行操作 如果加锁失败直接返回 如果加锁成功，则指定锁时间
*
* @param action
* 待执行的动作
* @param key
* 加锁的key
* @param expireSeconds
* 锁超时时间，单位秒
* @param <T>
* 泛型参数
* @return
*/
public static <T> T mutexExec(LockCallback<T> action, String key, int expireSeconds) {
if (setnx(key + IGNORE_EXEC_KEY, "TRUE", expireSeconds)) {
try {
return action.exec();
} catch (Exception e) {
throw new RedisExecption("ignoreExec error", e);
} finally {
del(key + IGNORE_EXEC_KEY);
}
}
return null;
}
/**
* 设置值成功返回true，key已存在则设置值失败，返回false
* @param key
* @param value
* @param expireSeconds，设置值过期时间，即使设置值失败，也会一直覆盖过期是时间
* @return
* @throws RedisExecption
*/
@SuppressWarnings({ "unchecked" })
public static boolean setnx(final String key, final Object value, final Integer expireSeconds)
throws RedisExecption {
if (StringUtils.isEmpty(key) || value == null) {
throw new RedisExecption("key或value不能为空");
}
return (Boolean) redisTemplate.execute(connection-> {
try {
Boolean result = false;
connection.multi();
setnx(key, value, expireSeconds, connection);
List<Object> list = connection.exec();
if (CollectionUtils.isNotEmpty(list)) {
result = (Boolean) list.get(0);
}
return result;
} catch (Exception e) {
// connection.discard();
throw new RuntimeException(e);
}
}, true);
}
/**
* 设置值成功返回true，key已存在则设置值失败，返回false
* @param key
* @param value
* @return
* @throws RedisExecption
*/
@SuppressWarnings({ "unchecked" })
public static boolean setnx(final String key, final Object value) throws RedisExecption {
if (StringUtils.isEmpty(key) || value == null) {
throw new RedisExecption("key或value不能为空");
}
return (Boolean) redisTemplate.execute(connection-> {
try {
Boolean result = false;
connection.multi();
setnx(key, value, -1, connection);
List<Object> list = connection.exec();
if (CollectionUtils.isNotEmpty(list)) {
result = (Boolean) list.get(0);
}
return result;
} catch (Exception e) {
// connection.discard();
throw new RuntimeException(e);
}
}, true);
}
/**
* 事务执行，不返回
* @param key
* @param value
* @param expireSeconds
* @param connection
*/
public static void setnx(final String key, final Object value, final Integer expireSeconds,
final RedisConnection connection) {
byte[] byteKey = serialize(key);
connection.setNX(byteKey, serialize(value));
if (NumUtil.intValue(expireSeconds) != -1) {
connection.expire(byteKey, expireSeconds);
}
}
/**
* 删除键
* @param key
* @return
* @throws RedisExecption
*/
@SuppressWarnings("unchecked")
public static Long del(final String key) throws RedisExecption {
return (Long) redisTemplate.execute(connection-> del(key, connection), true);
}
public static Long del(final String key, final RedisConnection connection) {
try {
return connection.del(serialize(key));
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 序列化
* @param value
* @return
*/
public static byte[] serialize(Object value) {
String valStr = getJsonStr(value);
return redisTemplate.getStringSerializer().serialize(valStr);
}
/**
* 返序列化
* @param bytes
* @param clz
* @return
*/
public static <T> T unserialize(byte[] bytes, Class<T> clz) {
// String valStr = getJsonStr);
Object json = redisTemplate.getStringSerializer().deserialize(bytes);
if (json == null) {
return null;
}
if (clz == null || clz.getName().equals("java.lang.String")) {
return (T) json;
} else {
return gson.fromJson(json.toString(), clz);
}
}
/**
* private static java.lang.reflect.Type imgType = new TypeToken<ArrayList
* <T>>() {}.getType();
*
* @param bytes
* @param type
* @return
*/
public static <T> T unserialize(byte[] bytes, Type type) {
// String valStr = getJsonStr);
Object json = redisTemplate.getStringSerializer().deserialize(bytes);
if (json == null) {
return null;
}
if (type == null) {
return (T) json;
} else {
return gson.fromJson(json.toString(), type);
}
}
private static String getJsonStr(Object value) {
String valStr = null;
if (value == null) {
valStr = null;
}
if (value instanceof String) {
valStr = value.toString();
} else if (value instanceof Number) {
valStr = value + "";
} else {
valStr = gson.toJson(value);
}
return valStr;
}
@SuppressWarnings("rawtypes")
public static RedisTemplate getRedisTemplate() {
return redisTemplate;
}
@SuppressWarnings({ "rawtypes", "static-access" })
public void setRedisTemplate(RedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
}
}
redis异常处理类
package com.framework.cache.execption;
import java.io.PrintWriter;
import java.io.StringWriter;
/**
* redis异常类，redis执行异常报错
* @author lyq
*
*/
public class RedisExecption extends RuntimeException {
private static final long serialVersionUID = 155L;
//private Logger log = Logger.getLogger(this.getClass());
public RedisExecption() {
}
public RedisExecption(String message) {
super(message);
}
public RedisExecption(Throwable cause) {
super(cause);
}
public RedisExecption(String message, Throwable cause) {
super(message, cause);
}
/**
* 以字符串形式返回异常堆栈信息
* @param e
* @return 异常堆栈信息字符串
*/
public static String getStackTrace(Exception e) {
StringWriter writer = new StringWriter();
e.printStackTrace(new PrintWriter(writer,true));
return writer.toString();
}
}
分布式锁超时异常
package com.framework.cache.execption;
/**
* 分布式锁超时报错
* @author lyq
*
*/
public class RedisLockTimeException extends RedisExecption {
/**
* 超时报错
* @param expireSeconds 单位秒
*/
private static final long serialVersionUID = 1L;
public RedisLockTimeException(String lockKey,Integer expireSeconds) {
super(String.format("分布式锁 ‘%s’等待超时: %ss",lockKey,expireSeconds));
}
}
分布式锁测试类
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import com.framework.cache.RedisHelper;
import redis.clients.jedis.JedisPoolConfig;
public class RedisTest {
public static void main(String[] args) {
RedisTemplate t = new RedisTemplate();
// InputStream is =
// RedisHelper.class.getClassLoader().getResourceAsStream("redis.properties");
String hostname = "";
String password = "";
String port = "";
// try {
// prop.load(is);
// } catch (IOException e) {
// // TODO Auto-generated catch block
// e.printStackTrace();
// }
/*** 热备start ***/
// String sentinelHost = "localhost";
// int sentinelport = 26379;
// RedisSentinelConfiguration sentinelConfiguration = new
// RedisSentinelConfiguration();
// sentinelConfiguration.setMaster("mymaster");
// sentinelConfiguration.sentinel(sentinelHost, sentinelport);
// JedisConnectionFactory connectionFactory = new
// JedisConnectionFactory(sentinelConfiguration);
/*** 热备end ***/
/*** 单机start ***/
// hostname = "120.25.226.230";
hostname = "127.0.0.1";
password = "";
port = "6379";
JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
connectionFactory.setPassword(password);
connectionFactory.setDatabase(5);
connectionFactory.setHostName(hostname);
connectionFactory.setPort(Integer.parseInt(port));
/*** 单机end ***/
JedisPoolConfig config = new redis.clients.jedis.JedisPoolConfig();
config.setMaxTotal(5000);
config.setMaxIdle(200);
config.setMaxWaitMillis(5000);
config.setTestOnBorrow(true);
connectionFactory.afterPropertiesSet();
t.setConnectionFactory(connectionFactory);
connectionFactory.setPoolConfig(config);
t.afterPropertiesSet();
RedisHelper.redisTemplate = t;
/************************************
* 分布式锁测试 start
*****************************************/
ThreadPoolExecutor executor = null;
// 有限队列，缓存30个请求等待，线程池默认策略，缓存超出上限报错
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(30);
// 初始化30个线程，最多30个线程
executor = new ThreadPoolExecutor(30, 30, 10, TimeUnit.SECONDS, workQueue, new ThreadFactory() {
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("redis-lock-thread");
return thread;
}
});
CyclicBarrier barrier = new CyclicBarrier(30);
for (Integer i = 0; i < 30; i++) {
final int _i = i;
executor.execute(() -> {
try {
barrier.await();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//锁等待10秒超时
RedisHelper.lockExec(() -> {
System.out.println(_i + "=====");
try {
Thread.sleep(2000);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return 2;
}, "test",10);
});
}
/************************************
* 分布式锁测试 end
*****************************************/
}
}
相关推荐
Java基于Redis实现分布式锁代码实例 分布式锁的必要性 在多线程环境中，资源竞争是一个常见的问题。例如，在一个简单的用户操作中，一个线程修改用户状态，首先在内存中读取用户状态，然后在内存中进行修改，然后...
以下是一个简单的C# Redis分布式锁实现： ```csharp public bool TryAcquireLock(string lockKey, int lockTimeout) { var redis = ConnectionMultiplexer.Connect("localhost:6379"); var db = redis.Get...
现在很多项目单机版已经不满足了，分布式变得越受欢迎，同时也带来很多问题，分布式锁也变得没那么容易实现，分享一个redis分布式锁工具类，里面的加锁采用lua脚本（脚本比较简单，采用java代码实现，无须外部调用...
redis实现分布式锁(java/jedis),其中包含工具方法以及使用demo 本资源是利用java的jedis实现 redis实现分布式锁(java/jedis),其中包含工具方法以及使用demo 本资源是利用java的jedis实现
在实际项目中，为了提高代码的健壮性和可维护性，通常会封装一个 Redis 分布式锁的客户端库，例如 Java 中的 Jedis 或 lettuce 库。这些库提供了更高级别的接口，隐藏了底层的 Redis 操作细节，使开发者更容易地使用...
3. **简单的API**：Redis提供了丰富的命令集，如`SETNX`、`EXPIRE`等，方便实现锁的逻辑。 要在SpringBoot中集成Redis，你需要先在项目中添加对应的依赖。SpringBoot提供了对Redis的自动配置支持，只需在`pom.xml`...
本篇文章将详细探讨如何使用Redisson实现Redis分布式事务锁，以及在Spring Boot环境中如何进行集成。 首先，Redis作为一个内存数据库，其高速读写性能使其成为实现分布式锁的理想选择。分布式锁的主要作用是在多...
Java Redis分布式锁的正确实现方式详解 Java Redis分布式锁是指使用Redis实现的分布式锁机制，旨在解决分布式系统中的并发问题。分布式锁有三种实现方式：数据库乐观锁、基于Redis的分布式锁和基于ZooKeeper的...
本文将深入探讨如何使用Redis实现分布式锁，以及如何利用自旋式加锁和Lua脚本实现原子性解锁。 首先，我们来理解分布式锁的基本概念。分布式锁是在多节点之间共享资源时，用于协调各个节点的访问控制机制。在分布式...
Redis 分布式锁工具包是为了解决在分布式系统中实现锁的问题，它提供了一种纯Java的方式来调用，使得在传统的Spring工程中可以轻松集成和使用。在现代的高并发、分布式环境下，单机锁已经无法满足需求，因为它们不能...
在IT行业中，尤其是在高并发的电子商务系统中，"redis分布式锁实现抢单秒杀"是一个常见的挑战。这个场景模拟了多个用户同时参与秒杀活动，系统需要确保库存的准确性和抢单的公平性，避免超卖和数据不一致的问题。...
2. Redlock：由Redis作者Antirez提出的分布式锁算法，通过在多台Redis实例上实现锁来提高可用性和容错性。 综上所述，Redis是实现分布式锁的理想选择，其高效、原子性的特性使得在分布式系统中控制共享资源成为可能...
Java开发基于SpringBoot+WebSocket+Redis分布式即时通讯群聊系统。一个基于Spring Boot + WebSocket + Redis，可快速开发的分布式即时通讯群聊系统。适用于直播间聊天、游戏内聊天、客服聊天等临时性群聊场景。 ...
redis分布式锁的工具类，采用的是Lua代码的方式，保证了Java执行方法的原子性。
redis 分布式锁java 实现
java面试题Redis分布式锁+zk 分布式锁.md
Redis分布式锁工具包简化了这一过程，提供了封装好的API，使开发者能够快速、安全地在代码中实现锁的功能。 该工具包可能包含以下核心功能： 1. **锁的获取与释放**：工具包通常会提供一个`lock()`方法用于获取锁...
在学习Java多线程编程的时候，锁是一个很重要也很基础的概念，锁可以看成是多... 掌握redis分布式锁的实现原理 掌握redis分布式锁在微服务项目中的应用 掌握redis分布式锁常见的面试题 以下是课程部分讲义截图：
以下是一个简单的Java示例，展示了如何使用Jedis客户端库来实现Redis分布式锁。 ```java public class RedisLock { private JedisPool jedisPool; public RedisLock(JedisPool jedisPool) { this.jedisPool = ...