Redis实现分布式锁和等待序列的方法示例

 更新时间:2019年06月18日 08:26:13   作者:小小小LIN子   我要评论
这篇文章主要介绍了Redis实现分布式锁和等待序列的方法示例,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

在集群下,经常会因为同时处理发生资源争抢和并发问题,但是我们都知道同步锁 synchronized 、 cas 、 ReentrankLock 这些锁的作用范围都是 JVM ,说白了在集群下没啥用。这时我们就需要能在多台 JVM 之间决定执行顺序的锁了,现在分布式锁主要有 redis 、 Zookeeper 实现的,还有数据库的方式,不过性能太差,也就是需要一个第三方的监管。

背景

最近在做一个消费 Kafka 消息的时候发现,由于线上的消费者过多,经常会遇到,多个机器同时处理一个主键类型的数据的情况发生,如果最后是执行更新操作的话,也就是一个更新顺序的问题,但是如果恰好都需要插入数据的时候,会出现主键重复的问题。这是生产上不被允许的(因为公司有异常监管的机制,扣分啥的),这是就需要个分布式锁了,斟酌后用了 Redis 的实现方式(因为网上例子多)

分析

redis 实现的分布式锁,实现原理是 set 方法,因为多个线程同时请求的时候,只有一个线程可以成功并返回结果,还可以设置有效期,来避免死锁的发生,一切都是这么的完美,不过有个问题,在 set 的时候,会直接返回结果,成功或者失败,不具有阻塞效果,需要我们自己对失败的线程进程处理,有两种方式

  • 丢弃
  • 等待重试 由于我们的系统需要这些数据,那么只能重新尝试获取。这里使用 redis 的 List 类型实现等待序列的作用

代码

直接上代码 其实直接redis的工具类就可以解决了

package com.test
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.List;

/**
 * @desc redis队列实现方式
 * @anthor 
 * @date 
 **/
public class RedisUcUitl {

  private static final String LOCK_SUCCESS = "OK";
  private static final String SET_IF_NOT_EXIST = "NX";
  private static final String SET_WITH_EXPIRE_TIME = "PX";

  private static final Long RELEASE_SUCCESS = 1L;

  private RedisUcUitl() {

  }
  /**
   * logger
   **/

  /**
   * 存储redis队列顺序存储 在队列首部存入
   *
   * @param key  字节类型
   * @param value 字节类型
   */
  public static Long lpush(Jedis jedis, final byte[] key, final byte[] value) {

    return jedis.lpush(key, value);
  
  }

  /**
   * 移除列表中最后一个元素 并将改元素添加入另一个列表中 ,当列表为空时 将阻塞连接 直到等待超时
   *
   * @param srckey
   * @param dstkey
   * @param timeout 0 表示永不超时
   * @return
   */
  public static byte[] brpoplpush(Jedis jedis,final byte[] srckey, final byte[] dstkey, final int timeout) {

    return jedis.brpoplpush(srckey, dstkey, timeout);

  }

  /**
   * 返回制定的key,起始位置的redis数据
   * @param redisKey
   * @param start
   * @param end -1 表示到最后
   * @return
   */
  public static List<byte[]> lrange(Jedis jedis,final byte[] redisKey, final long start, final long end) {
    
    return jedis.lrange(redisKey, start, end);
  }

  /**
   * 删除key
   * @param redisKey
   */
  public static void delete(Jedis jedis, final byte[] redisKey) {
    
     return jedis.del(redisKey);
  }

  /**
   * 尝试加锁
   * @param lockKey key名称
   * @param requestId 身份标识
   * @param expireTime 过期时间
   * @return
   */
  public static boolean tryGetDistributedLock(Jedis jedis,final String lockKey, final String requestId, final int expireTime) {
    String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
    return LOCK_SUCCESS.equals(result);

  }

  /**
   * 释放锁
   * @param lockKey key名称
   * @param requestId 身份标识
   * @return
   */
  public static boolean releaseDistributedLock(Jedis jedis,final String lockKey, final String requestId) {
    final String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));

    return RELEASE_SUCCESS.equals(result);

  }
}

业务逻辑主要代码如下

1.先消耗队列中的

while(true){
  // 消费队列
  try{
    // 被放入redis队列的数据 序列化后的
    byte[] bytes = RedisUcUitl.brpoplpush(keyStr.getBytes(UTF_8), dstKeyStr.getBytes(UTF_8), 1);
    if(bytes == null || bytes.isEmpty()){
      // 队列中没数据时退出
      break;
    }
    // 反序列化对象
    Map<String, Object> singleMap = (Map<String, Object>) ObjectSerialUtil.bytesToObject(bytes);
    // 塞入唯一的值 防止被其他线程误解锁
    String requestId = UUID.randomUUID().toString();
    boolean lockGetFlag = RedisUcUitl.tryGetDistributedLock(keyStr,requestId, 100);
    if(lockGetFlag){
      // 成功获取锁 进行业务处理
      //TODO
      // 处理完毕释放锁 
      boolean freeLock = RedisUcUitl.releaseDistributedLock(keyStr, requestId);

    }else{
      // 未能获得锁放入等待队列
     RedisUcUitl.lpush(keyStr.getBytes(UTF_8), ObjectSerialUtil.objectToBytes(param));
  
    }
    
  }catch(Exception e){
    break;
  }
  
}

2.处理最新接到的数据

同样是走尝试获取锁,获取不到放入队列的流程

一般序列化用 fastJson 之列的就可以了,这里用的是 JDK 自带的,工具类如下

public class ObjectSerialUtil {

  private ObjectSerialUtil() {
//    工具类
  }

  /**
   * 将Object对象序列化为byte[]
   *
   * @param obj 对象
   * @return byte数组
   * @throws Exception
   */
  public static byte[] objectToBytes(Object obj) throws IOException {
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    ObjectOutputStream oos = new ObjectOutputStream(bos);
    oos.writeObject(obj);
    byte[] bytes = bos.toByteArray();
    bos.close();
    oos.close();
    return bytes;
  }


  /**
   * 将bytes数组还原为对象
   *
   * @param bytes
   * @return
   * @throws Exception
   */
  public static Object bytesToObject(byte[] bytes) {
    try {
      ByteArrayInputStream bin = new ByteArrayInputStream(bytes);
      ObjectInputStream ois = new ObjectInputStream(bin);
      return ois.readObject();
    } catch (Exception e) {
      throw new BaseException("反序列化出错!", e);
    }
  }
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • redis源码分析教程之压缩链表ziplist详解

    redis源码分析教程之压缩链表ziplist详解

    ziplist结构在redis运用非常广泛,是列表、字典等数据类型的底层结构之一。ziplist的优点在于能够一定程度地节约内存。下面这篇文章主要给大家介绍了关于redis源码分析教程之压缩链表ziplist的相关资料,需要的朋友可以参考借鉴,下面来一起看看吧。
    2017-12-12
  • redis集群规范详解

    redis集群规范详解

    这篇文章主要介绍了redis集群规范详解,涉及节点失效检测、集群状态检测、从节点选举等相关内容,比较详细,需要的朋友可以参考。
    2017-10-10
  • Redis 对比 Memcached 并在 CentOS 下进行安装配置详解

    Redis 对比 Memcached 并在 CentOS 下进行安装配置详解

    Redis 是一个开源、支持网络、基于内存、键值对的 Key-Value 数据库,本篇文章主要介绍了Redis 对比 Memcached 并在 CentOS 下进行安装配置详解,有兴趣的可以了解一下。
    2016-11-11
  • Redis配置文件redis.conf详细配置说明

    Redis配置文件redis.conf详细配置说明

    本文列出了Redis的配置文件redis.conf的各配置项的详细说明,简单易懂
    2018-03-03
  • redis通过位图法记录在线用户的状态详解

    redis通过位图法记录在线用户的状态详解

    这篇文章主要给大家介绍了关于redis如何通过位图法记录在线用户的状态的相关资料,文中先对位图进行了一个简单的介绍,而后通过示例代码将实现的方法介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2018-11-11
  • 安装Redis就那么几步,很简单

    安装Redis就那么几步,很简单

    Redis是一种非关系型数据库(NoSQL),NoSQL是以key-value的形式存储,和传统的关系型数据库不一样,不一定遵循传统数据库的一些基本要求,本文重点给大家介绍安装Redis的步骤,需要的朋友参考下吧
    2018-11-11
  • Redis主从实现读写分离

    Redis主从实现读写分离

    这篇文章主要为大家详细介绍了Redis主从实现读写分离的相关资料 ,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2016-10-10
  • Redis中事件驱动模型示例详解

    Redis中事件驱动模型示例详解

    Redis这个数据库相信不用过多介绍了,大家应该都知道,下面这篇文章主要给大家介绍了关于Redis中事件驱动模型的相关资料,文中通过示例代码介绍的非常详细,需要的朋友可以参考借鉴,下面随着小编来一起学习学习吧。
    2018-03-03
  • Redis事务涉及的watch、multi等命令详解

    Redis事务涉及的watch、multi等命令详解

    这篇文章主要介绍了Redis事务涉及的watch、multi等命令,本文给大家介绍的非常详细,具有一定的参考借鉴价值 ,需要的朋友可以参考下
    2018-10-10
  • windows 64位下redis安装教程

    windows 64位下redis安装教程

    这篇文章主要为大家详细介绍了windows 64位下redis安装教程,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2017-10-10

最新评论