分布式锁实现机制_13.锁的实现机制

分布式锁实现机制

---

介绍说明

单点应用下,并发场景相对还比较好控制,可以借助java.util下的并发包工具能
够解决大部分问题。但是在多节点分布式场景
下,java.util.locks.ReentrantLock可能就并不能发挥多大作用了,此时我们
需要借助分布式锁来控制并发。
1
2
3
4

解决思路

分布式场景下之所以不能够使用并发包下的锁解决并发问题，那是因为多节点是每个应用都有相互独立的进程,他们没有共享内存资源内存因此很难控制并发。

要想控制分布式应用并发问题

1、内存共享(同一台数据库服务器mysql、同一台缓存数据库memcached/redis)
2、彼此进行消息通信/彼此可见(zookeeper)
备注:其实分布式问题最终都要把它简化成单点问题来解决,否则就很难解决。
1
2
3

常用的分布式锁实现策略

1、基于数据库(mysql)的表锁(比如对某个插入字段orderId作唯一索引约束)或增加一个版本号字段乐观锁实现

基于mysql锁表(唯一索引)

该实现方式完全是依靠数据库的唯一索引来控制分布式并发，可扩展性和应用场景有限，某些场景下可以使用。
1

---

表结构如下

CREATE TABLE `order_info` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `order_id` int(11) DEFAULT NULL,
  `insert_time` datetime DEFAULT NULL,
  `order_type` varchar(10) DEFAULT NULL,
  `update_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `order_id` (`order_id`)//对order_id建立唯一索引
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
//当多节点分布式场景下对同一个订单执行插入操作时，就可以利用数据库服务器提供的唯一索引来控制住该场景下的并发问题。
1
2
3
4
5
6
7
8
9
10

基于mysql的乐观锁,比如在表中增加一个version版本号来控制

表结构

CREATE TABLE `order_info` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `order_id` int(11) DEFAULT NULL,
  `insert_time` datetime DEFAULT NULL,
  `order_type` varchar(10) DEFAULT NULL,
  `update_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  `version` int(11) DEFAULT NULL,//根据这个版本号来判断更新之前有没有其它线程对它更新过
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
1
2
3
4
5
6
7
8
9

示例代码

//1、先查询出该记录对应的版本号
int versionCode=jdbc.selectVersion(sql);

//2、执行更新操作
//sql大概是这样的
update order_info set version=versionCode+1 where version=versionCode; 
jdbc.updateOrderInfo(sql);
//如果update能够影响一条语句，那么说明它是占到资源更新成功，如果没有那么表示资源已经被别的线程占用了。

//每次插入或更新的时候先检查查出的version是否发生了变化。
1
2
3
4
5
6
7
8
9
10

备注:

基于数据库实现比较简单,可扩展性不好，数据库锁实现只能是非阻塞锁，即应该
为tryLock，是尝试获得锁，如果无法获得则会返回失败。该锁机制没有过期时间。
1
2

---

2、基于内存服务器的原子性操作或compare and set(检查比较再赋值),比如memcached/redis缓存服务器实现

示例代码

memcached实现的锁工具类

/**
 * 尝试获得锁资源
 * 
 * @param id
 * @return
 */
public static boolean tryLock(String id) {

    MemcachedClient memcachedClient = MemcachedUtils.getMemcachedClient();
    try {
        return memcachedClient.add(id, 100, "value");
        //最好给锁资源设置一个有效时间
    } catch (Exception e) {
        e.printStackTrace();
    }
    return false;
}

/**
 * 尝试释放锁资源
 * 
 * @param id
 * @return
 */
public static boolean releaseLock(String id) {
    MemcachedClient memcachedClient = MemcachedUtils.getMemcachedClient();
    try {
        return memcachedClient.delete(id);
    } catch (Exception e) {
        e.printStackTrace();
    }
    return false;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

使用方式AppMain

    String id = "xxx";
    try {
        boolean getLock = MemcachedLock.tryLock(id);
        if (getLock) {//如果获得锁资源
            // doSomething()...
        }
    } finally {
        MemcachedLock.releaseLock(id);
        //不论执行业务执行成功与否都要释放资源
    }
    //使用锁一定要将代码块放入try..catch..finally代码块中
1
2
3
4
5
6
7
8
9
10
11

redis等实现分布式锁大致相同。

备注:

基于内存服务器实现方式比较常用,扩展性比较好并且实现简单性能较好,缺点就
是内存服务器数据通常没有持久化功能,发生宕机时不好追踪问题。
1
2

---

3、基于消息通信的协调服务的zookeeper实现

示例代码:

/**
 * 分布式锁方案实现
 * Meilele.com Inc.
 * Copyright (c) 2008-2015 All Rights Reserved.
 * @author xuyi3
 * @version $Id: DistributedSharedLock.java, v 0.1 2015年12月21日 下午1:46:47 xuyi3 Exp $
 */
public class DistributedSharedLock implements Watcher {

    /** SLF4J日志*/
    private static final Logger logger       = LoggerFactory.getLogger(DistributedSharedLock.class);
    /** 连接zookeeper的服务器地址 */
    private static final String ADDR         = Configure.ZOO_KEEPER_XS;
    /** 创建子节点名称*/
    private static final String LOCK_NODE    = Configure.LOCK_NODE;
    /** 锁目录名称*/
    private String              rootLockNode = Configure.LOCK_PATH;                                 // 锁目录
    /** zookeeper对象*/
    private ZooKeeper           zk           = null;
    /** 互斥标识*/
    private Integer             mutex;
    /** 当前锁标识*/
    private Integer             currentLock;

    /**
     * 构造函数实现 连接zk服务器 创建zk锁目录
     *
     * @param rootLockNode
     */
    public DistributedSharedLock() {

        try {
            // 连接zk服务器
            zk = new ZooKeeper(ADDR, Configure.TIME_OUT, this);
        } catch (IOException e) {
            e.printStackTrace();
        }
        mutex = new Integer(-1);
        if (zk != null) {
            try {
                // 建立根目录节点
                Stat s = zk.exists(rootLockNode, false);
                if (s == null) {
                    zk.create(rootLockNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
                }
            } catch (KeeperException e) {
                logger.info("Keeper exception when instantiating queue:{}", e.toString());
            } catch (InterruptedException e) {
                logger.info("InterruptedException{}", e.toString());
            }
        }
    }

    /**
     * 请求zk服务器，获得锁
     * 基于创建临时节点,节点最小的获得锁持有权的算法。
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void acquire() throws KeeperException, InterruptedException {
        ByteBuffer b = ByteBuffer.allocate(4);
        byte[] value;
        b.putInt(ThreadLocalRandom.current().nextInt(10));
        value = b.array();

        // 创建锁节点
        String lockName = zk.create(rootLockNode + "/" + LOCK_NODE, value,
            ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        synchronized (mutex) {
            while (true) {
                // 获得当前锁节点的number，和所有的锁节点比较
                Integer acquireLock = new Integer(lockName.substring(lockName.lastIndexOf('-') + 1));

                List<String> childLockNode = zk.getChildren(rootLockNode, true);

                //将创建的子节点放入TreeSet中进行排序
                SortedSet<Integer> sortedLock = new TreeSet<Integer>();
                for (String temp : childLockNode) {
                    Integer tempLockNumber = new Integer(temp.substring(temp.lastIndexOf('-') + 1));
                    sortedLock.add(tempLockNumber);
                }
                //获取最小节点,创建该节点的连接持有锁资源。
                currentLock = sortedLock.first();

                // 如果当前创建的锁的序号是最小的那么认为这个客户端获得了锁
                if (currentLock >= acquireLock) {
                    return;
                } else {
                    // 没有获得锁则等待下次事件的发生
                    mutex.wait();
                }
            }
        }
    }

    /**
     * 释放锁
     *
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void release() throws KeeperException, InterruptedException {
        String lockName = String.format("%010d", currentLock);
        zk.delete(rootLockNode + "/" + LOCK_NODE + lockName, -1);
        //释放锁的时候应该主动关闭和zookeeper的连接
        if (zk.getState() == States.CONNECTED) {
            zk.close();
        }

    }

    /**
     * 监听器必须实现的方法
     * 
     */
    public void process(WatchedEvent arg0) {
        synchronized (mutex) {
            mutex.notifyAll();
        }
    }
}
//是基于zookeeper临时节点实现的分布式锁。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

备注:

基于zookeeper实现方式是最合适的,zookeeper支持watcher机制，这样实现阻
塞锁，可以watch锁数据，等到数据被删除，zookeeper会通知客户端去重新竞争
锁。zookeeper的数据可以支持临时节点的概念，即客户端写入的数据是临时数
据，在客户端宕机后，临时数据会被删除，这样就实现了锁的异常释放。但是使用
zookeeper分布式锁搭建比较繁琐，如果企业内部本身就已经搭建了zookeeper
服务那么推荐使用,否则推荐第二种实现方式。
1
2
3
4
5
6

参考

http://blog.chinaunix.net/uid-26111972-id-3759540.html
http://www.importnew.com/20307.html
http://ju.outofmemory.cn/entry/75844
http://www.jianshu.com/p/e3a1637f79a1