Tair分布式锁 实践经验(160805更新)

前言

在JVM中，怎么样防止因多线程并发造成的数据不一致或逻辑混乱？大家一定都能想到锁，能想到java.util.concurrent包下的各种各样的用法。
但在分布式环境下，不同的机器、不同的jvm，那些好用的并发锁工具，并不能满足需要。

所以我们需要有一个分布式锁，来解决分布式环境下的并发问题。而本文正是在这条道路上，走出的一些经验的总结。

我们按照待解决问题的场景，一步一步看下去。

问题一：如何实现一个分布式锁；

锁，大多是基于一个只能被1个线程占用、得到的资源来实现的。
JVM的锁，是基于CPU对于寄存器的修改指令cmpxchg，来保证旧值改为新值的操作，只有一个能成功。这里的旧值，就是这个被争夺的资源，大多数情况，并发时第一个线程对旧值修改成功后，其他线程就没有机会了。（当然，ABA是另外一个话题，这里就不说了。。）

所以，在分布式环境下，要实现一个锁，我们就要找个一个只能被1个线程占用的资源。
有经验的开发很快能想到，共享磁盘文件、缓存、mysql数据库，这些分布式环境下，数据表现为单份的，都应当能满足需求。

然而，基于文件、DB会遭遇各式各样的问题，性能，经常也会是瓶颈。
因此，我们这里使用的，是淘系用的最多的缓存中间件产品--Tair。

查看com.taobao.tair.TairManager接口，发现有以下几个接口或许适合使用：

  • TairManager.incr 加法计数器。首次使用时，返回值能等于默认值，而不被+n的机会，只有一个。貌似可以用。
  • TairManager.lock 看起来名字像是这个意思，姑且拿来一试。
  • TairManager.put 传入version版本进行校验，cas原则会保证只有一个能成功。貌似可以用。

（注：mdb有可能丢、且invalid时不保证跨机房一致性，所以这个锁肯定需要用ldb来实现的。）

在线上多机房情况下，做了一下测试，测试程序核心代码大约是：

void test(){           // 2个机房的jvm实例，每个实例n个线程同时执行本方法；
    while(true){
        if(tryLock()){     // tryLock\unLock 的实现对应有3套；
            try{
                logger.info("Got lock! Hostname:{} ThreadName:{}",getHostname(),Thread.currentThread().getName());
                Thread.sleep(1000);
            }finally {
                unLock();
                logger.info("Release lock! Hostname:{} ThreadName:{}",getHostname(),Thread.currentThread().getName());
            }
        }
    }
}

测试的结论如下：

  • TairManager.incr 初始几次锁的获取和释放没有问题，但是后来返回值很大，就谁也拿不到锁；怀疑和接口超时、或invalid机制有关；

    • 补充： 后面发现incr做锁的一种可靠方案是，使用值限定参数：int lowBound, int upperBound
      比如：lowBound=0, upperBound=1，则value一直在0与1之间切换，用过多次，还是很靠谱的。
  • TairManager.lock 完全不行，lock接口看来根本不是做这个用的；真正的使用场景是锁住一个key不容许更新，不是锁机制。

    0

  • TairManager.put 非常稳定、靠谱；有个现象值得关注：A机房invalid之后，B机房会先拿到锁；因为invalid先从远程机房开始；

最后，给出ldb put实现的分布式锁的核心代码（__后面都基于ldb put来实现__）：

public boolean trylock(String key) {
    ResultCode code = ldbTairManager.put(NAMESPACE, key, "This is a Lock.", 2, 0);
    if (ResultCode.SUCCESS.equals(code))
        return true;
    else
        return false;
}
public boolean unlock(String key) {
    ldbTairManager.invalid(NAMESPACE, key);
}

问题二：lock之后，程序发布\或者进程crash，trylock就永远false了；

每次发布，总发现有些异常数据，拿不到锁，不能继续向前走；
仔细分析，原来tair的lock，一直没能释放；

要解决这个问题，可以先不管原因，无脑的给tair put加上超时时间就行，这样业务至少可以自行恢复。
但是，这个超时时间需要仔细考虑把握，需要在业务承受范围之内。

注： 像程序发布、进程crash，这种情况，是无可避免的让锁没机会释放。还有其他可能性，大多是bug了。。

public boolean trylock(String key, int timeout) {
    ResultCode code = ldbTairManager.put(NAMESPACE, key, "This is a Lock.", 2, timeout);
    if (ResultCode.SUCCESS.equals(code))
        return true;
    else
        return false;
}

问题二：tair存在让人苦恼的超时问题，即使千分之1，本业务有时也不能容忍

tair的大神给的回复很简单：超时请重试；

仔细想一下，put超时分两种情况：

  1. 上一次put其实已经成功了；那么重试肯定会失败；
  2. 上一次put其实失败了；那么若这一次顺利，就会成功；

那如何解决上面的第1点呢？

其实，锁应该是能够经得起复查的（类似偏向锁）：A拿到的锁，没有unlock之前，无论A重试检查多少次，都是A的！

怎么实现？

既然用的是ldb缓存，它是key-value结构的，前面version控制等，都只用到了key。
这里，我们可以从tair value里做文章：让value包含机器ip+线程name，trylock内先get value做检查

于是，实现变为：

public boolean trylock(String key, int timeout) {
    Result<DataEntry> result = locker.ldbTairManager.get(NAMESPACE, lockKey);
    if (result == null) return null;
    if (ResultCode.DATANOTEXSITS.equals(result.getRc())) { // means lock is free
        ResultCode code = ldbTairManager.put(NAMESPACE, key, getLockValue(), 2, timeout);
        if (ResultCode.SUCCESS.equals(code))
            return true;
    }else if(result.getValue() != null && getLockValue().equals(result.getValue().getValue())){
        return true;
    }
    return false;
}

private String getLockValue(){
    return Utils.getHostname() + ":" + Thread.currentThread().getName();
}

注意： 其实这里线程复用时，ThreadName有相同风险，可以改为uuid逻辑，需复用锁时传入uuid。

public boolean trylock(String key, int timeout, String uuid){ ... }

问题三：新方案其实多了一次get操作，若是get也超时怎么办？

超时无法避免，还是要靠重试！（前提是逻辑可以重试）

public boolean trylock(String key, int timeout) {
    Result<DataEntry> result = locker.ldbTairManager.get(NAMESPACE, key);
    if (result == null || ResultCode.CONNERROR.equals(result.getRc()) || ResultCode.TIMEOUT.equals(result.getRc()) || ResultCode.UNKNOW.equals(result.getRc())) // get timeout retry case
        result = locker.ldbTairManager.get(NAMESPACE, key);
    if (result == null || ResultCode.CONNERROR.equals(result.getRc()) || ResultCode.TIMEOUT.equals(result.getRc()) || ResultCode.UNKNOW.equals(result.getRc())){ // 还是超时，则留下日志痕迹
        logger.error("ldb tair get timeout. key:{}.",key);
        return false；
    }
    if (ResultCode.DATANOTEXSITS.equals(result.getRc())) { // means lock is free
        ResultCode code = ldbTairManager.put(NAMESPACE, key, getLockValue(), 2, timeout);
        if (ResultCode.SUCCESS.equals(code))
            return true;
        else if(code==null || ResultCode.CONNERROR.equals(result.getRc()) || ResultCode.TIMEOUT.equals(result.getRc()) || ResultCode.UNKNOW.equals(result.getRc())){  // put timeout retry case
            return trylock(key, timeout); // 递归尝试
        }
    }else if(result.getValue() != null && getLockValue().equals(result.getValue().getValue())){
        return true;
    }
    return false;
}

// 【注意】其实这里线程复用时，ThreadName有相同风险，可以改为uuid逻辑，复用锁传入uuid。
private String getLockValue(){
    return Utils.getHostname() + ":" + Thread.currentThread().getName();
}

进一步的，我们还可以对get/put的retry做次数控制；
真实线上的情况，一般一次retry就能解决问题，次数多了，反而可能导致雪崩，需要慎重；

多一次Get的性能影响

有代码洁癖、性能洁癖的人可能会想：普通tair锁，一次put就能搞定，这里却要先get再put，浪费啊。。。

这里梳理一下：

  • 若是锁已经被持有，那么get之后，麻烦发现被持有，直接返回失败；这时，并不会再次put，开销是一样的；（甚至get的开销，比put要小，至少不会占用put的限流阈值）
  • 若是没人持有锁，确实这时get有些浪费的，但是为了锁可以复查这个特性（可重试）、为了能解决超时这个问题，我认为还是值得的。在实际场景中，开发者自己可以评估是否需要。比如：拿前面的uuid的样例API讲，若不需要这个特性时，就不传入uuid，那么实现代码里，可以自动降级为只有一个put的锁实现；

问题四：批量锁

批量锁，主要注意拿锁的顺序和释放锁相反，伪代码如下：

if(trylock("A") && trylock("B") && trylock("C")){
    try{
        // do something
    }finally{  // 注意这里的顺序要反过来
        unlock("C");
        unlock("B");
        unlock("A");
    }
}

最后总结下，给一个完整代码 :smile:

import com.taobao.tair.DataEntry;
import com.taobao.tair.Result;
import com.taobao.tair.ResultCode;
import com.taobao.tair.TairManager;
import org.apache.commons.lang.NotImplementedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.helpers.FormattingTuple;
import org.slf4j.helpers.MessageFormatter;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class CommonLocker {

    private static final Logger logger = LoggerFactory.getLogger(CommonLocker.class);

    @Resource
    private TairManager ldbTairManager;

    private static final short NAMESPACE = 1310;

    private static CommonLocker locker;

    public void init() {
        if (locker != null) return;
        synchronized (CommonLocker.class) {
            if (locker == null)
                locker = this;
        }
    }

    public static Lock newLock(String format, Object... argArray) {
        FormattingTuple ft = MessageFormatter.arrayFormat(format, argArray);
        return newLock(ft.getMessage());
    }

    public static Lock newLock(String strKey) {
        String key = "_tl_" + strKey;
        return new TairLock(key, CommonConfig.lock_default_timeout);
    }

    public static Lock newLock(String strKey, int timeout) {
        String key = "_tl_" + strKey;
        return new TairLock(key, timeout);
    }

    private static class TairLock implements Lock {
        private String lockKey;
        private boolean gotLock = false;
        private int retryGet = 0;
        private int retryPut = 0;
        private int timeout;

        public TairLock(String key, int timeout) {
            this.lockKey = tokey(key);
            this.timeout = timeout;
        }

        public boolean tryLock() {
            return tryLock(timeout);
        }

        /**
         * need finally do unlock
         *
         * @return
         */
        public boolean tryLock(int timeout) {
            Result<DataEntry> result = locker.ldbTairManager.get(NAMESPACE, lockKey);
            while (retryGet++ < CommonConfig.lock_get_max_retry &&
                    (result == null || ResultCode.CONNERROR.equals(result.getRc()) || ResultCode.TIMEOUT.equals(result.getRc()) || ResultCode.UNKNOW.equals(result.getRc()))) // 重试一次
                result = locker.ldbTairManager.get(NAMESPACE, lockKey);
            if (ResultCode.DATANOTEXSITS.equals(result.getRc())) { // lock is free
                // 已验证version 2表示为空，若不是为空，则返回version error
                ResultCode code = locker.ldbTairManager.put(NAMESPACE, lockKey, locker.getValue(), 2, timeout);
                if (ResultCode.SUCCESS.equals(code)) {
                    gotLock = true;
                    return true;
                } else if (retryPut++ < CommonConfig.lock_put_max_retry &&
                        (code == null || ResultCode.CONNERROR.equals(result.getRc()) || ResultCode.TIMEOUT.equals(result.getRc()) || ResultCode.UNKNOW.equals(result.getRc()))) {
                    return tryLock(timeout);
                }
            } else if (result.getValue() != null && locker.getValue().equals(result.getValue().getValue())) {
// 【注意】其实这里线程复用时，ThreadName有相同风险，可以改为uuid逻辑，复用锁传入uuid。
                // 若是自己的锁，自己继续用
                gotLock = true;
                return true;
            }
            // 到这里表示没有拿到锁
            return false;
        }

        public void unlock() {
            if (gotLock) {
                ResultCode invalidCode = locker.ldbTairManager.invalid(NAMESPACE, lockKey);
                gotLock = false;
            }
        }

        public void lock() {
            throw new NotImplementedException();
        }

        public void lockInterruptibly() throws InterruptedException {
            throw new NotImplementedException();
        }

        public boolean tryLock(long l, TimeUnit timeUnit) throws InterruptedException {
            throw new NotImplementedException();
        }

        public Condition newCondition() {
            throw new NotImplementedException();
        }
    }

// 【注意】其实这里线程复用时，ThreadName有相同风险，可以改为uuid逻辑，复用锁传入uuid。
    private String getValue() {
        return getHostname() + ":" + Thread.currentThread().getName();
    }

    /**
     * 获得机器名
     *
     * @return
     */
    public static String getHostname() {
        try {
            return InetAddress.getLocalHost().getHostName();
        } catch (UnknownHostException e) {
            return "[unknown]";
        }
    }

    public void setLdbTairManager(TairManager ldbTairManager) {
        this.ldbTairManager = ldbTairManager;
    }
}

使用样例

Lock lockA = CommonLocker.newLock("hs_room_{}_uid_{}", roomDo.getUuid(), roomDo.getMaster().getUid());
Lock lockB = CommonLocker.newLock("hs_room_{}_uid_{}", roomDo.getUuid(), roomDo.getPartnerList().get(0).getUid());
try {
    if (lockA.tryLock() && lockB.tryLock()) {// 分布式锁定本任务
        // do something....
    }
} finally {
    lockB.unlock();
    lockA.unlock();
}

祝大家用的愉快，Happy Coding!!

时间： 2024-10-29 06:27:16

Tair分布式锁 实践经验(160805更新)的相关文章

ZooKeeper场景实践：（7） 分布式锁

1.基本介绍 分布式锁是控制分布式系统之间同步访问共享资源的一种方式,需要互斥来防止彼此干扰来保证一致性.利用Zookeeper的强一致性可以完成锁服务.Zookeeper的官方文档是列举了两种锁,独占锁和共享锁.独占锁保证任何时候都只有一个进程能或者资源的读写权限.共享锁可以同时有多个读,但是同一时刻最多只能有一个写,读和写是互斥的. 2.场景分析 我们准备来实现互斥的锁,按照官网的思路,给定一个锁的路径,如/Lock,所有要申请这个锁的进程都在/Lock目录下创建一个/Lock/lock-的

基于zookeeper的分布式并发锁实践

基于zookeeper的分布式并发锁实践 http://www.365yg.com/item/6421663785764782593/

C#和.NET中的分布式锁服务

背景 分布式锁服务在大家的项目中或许用的不多,因为大家都把排他放在数据库那一层来挡.当大量的行锁.表锁.事务充斥着数据库的时候,不如换个角度思考问题.一般web应用很多的瓶颈都在数据库上,这里给大家介绍的是减轻数据库锁负担的一种方案. 简介 如果我们的需求很简单,例如对于用户的账户资金,要保证原子性操作.并且不同的客户端在同一时间内只能提交一个对象操作.lock.单例?!在单台上还可以,但是大型web项目上,负载均衡是常用的技术手段手段,同一意义的对象可能存在不同的副本,这时我们又如何保证排他操

DockOne微信分享（八十一）：唯品会数据库备份恢复容器化项目实践经验总结

本文讲的是DockOne微信分享(八十一):唯品会数据库备份恢复容器化项目实践经验总结[编者的话]本文分享了唯品会数据库Docker的异地容灾项目实践经验,项目中针对用户数据库的异地恢复场景的需求进行开发和测试,整合了网络,存储.调度.监控,镜像等多个模块.在实施完成后,从技术上总结关于选型.开发.踩坑.测试等方面的经验. 项目背景 数据库Docker的异地备份恢复容灾项目,针对用户数据库的异地备份恢复场景的需求进行开发和测试,整合了容器网络.存储.调度.监控.镜像等多个模块.同时针对数据库的日

基于Zookeeper的分布式锁

这篇文章只需要你10分钟的时间. 实现分布式锁目前有三种流行方案,分别为基于数据库.Redis.Zookeeper的方案,其中前两种方案网络上有很多资料可以参考,本文不做展开.我们来看下使用Zookeeper如何实现分布式锁. 什么是Zookeeper? Zookeeper(业界简称zk)是一种提供配置管理.分布式协同以及命名的中心化服务,这些提供的功能都是分布式系统中非常底层且必不可少的基本功能,但是如果自己实现这些功能而且要达到高吞吐.低延迟同时还要保持一致性和可用性,实际上非常困难.因此z

redis分布式锁的运用和理解

问题描述 redis分布式锁的运用和理解 redis是单进程单线程模式 ,为什么还需要分布式锁?跨jvm是指多个服务器上在运行同一个服务吗? 网上看了很多说跨jvm会需要锁 ,但还是不怎么理解 解决方案 redis分布式锁redis分布式锁-SETNX实现Redis实现分布式锁 解决方案二: redis 3.0也有集群模式了,可以多个机器组成一个整体的cache 解决方案三: redis分布式锁和zookeeper分布式锁都在在企业服务中常用的知识.譬如:你有一个服务程序,需要部署在5台独立的机

基于Redis的分布式锁真的安全吗？（下）

自从我写完这个话题的上半部分之后,就感觉头脑中出现了许多细小的声音,久久挥之不去.它们就像是在为了一些鸡毛蒜皮的小事而相互争吵个不停.的确,有关分布式的话题就是这样,琐碎异常,而且每个人说的话听起来似乎都有道理.   今天,我们就继续探讨这个话题的后半部分.本文中,我们将从Antirez反驳Martin Kleppmann的观点开始讲起,然后会涉及到Hacker News上出现的一些讨论内容,接下来我们还会讨论到基于Zookeeper和Chubby的分布式锁是怎样的,并和Redlock进行一些对

redis 如何实现分布式锁

问题描述 redis 如何实现分布式锁 redis 使用 GETSET命令实现的分布式锁 必须设置超时时间来防止死锁的发生,但是如果一个线程的执行时间超过了设定的超时时间,锁会被另外一个线程获取,但是当前线程还没执行结束,这样就造成了 多个线程同时可以访问一个资源的问题,请问有办法解决吗?

【推荐】如何使用好阿里云的网络安全隔离？深入分享阿里云ECS安全组实践经验

近期针对安全的话题非常火爆.特别邀请阿里云专家撰写一组文章,深入分享阿里云ECS安全组实践经验. 阿里云的ECS有经典网络和专有网络(VPC)两种网络类型.经典网络基于阿里云基础网络,通过安全组(类似虚拟防火墙)的一系列数据链路层.网络层访问控制策略实现资源隔离.安全组策略是白名单的方式,不同租户的服务器会处于不同的安全组,默认情况下安全组仅会放行同租户同安全组内的访问请求. 因此安全组和防火墙一样,一般建议配置最小开放原则,关闭不需要外部访问的端口,仅放行必要的入网地址,避免配置的过于开放而给