下面说一下分布式实现的几种方式：java
1、数据库悲观锁node
所谓的悲观锁：顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，因此每次拿数据的时候都会上锁。这样别人拿数据的时候就要等待直到锁的释放。redis
这里是采用oracle的 select ...... where id=1 for update 来实现分布式锁，建议加上nowait，或者wait 以及 of数据库
下面是demo：apache
select * from table where id=1 for update nowait;//当锁被占用，不等待直接报错 select * from table where id=1 for update wait 6;//当锁被占用，等待6s select * from table where id=1 for update of columns nowait;//锁定执行的列，反之则锁全部列。
该方案，在高并发时显然不适用，依赖于数据库的性能以及锁机制，会形成锁没法释放。api
2、数据库乐观锁缓存
所谓的乐观锁：就是很乐观，每次去拿数据的时候都认为别人不会修改，因此不会上锁，可是在更新的时候会判断一下在此期间别人有没有去更新这个数据。通常的方案都是加一个版本号字段（version），在查询数据时将版本号带出来，更新后将版本号+1，若是版本号一致才更新，并获取影响行数，若是没更新则报错。服务器
3、redis的setnxsession
因为redis是单线程工做的，因此它存取key-value 的时候是单线程工做的，而且多个线程请求redis并不存在竞争问题。因此能够设置一个标识来做为一把锁，只有获取了该锁以后，才能对共享的资源进行操做，没有拿到锁的线程处于不断去取锁的状态，直到等到上一个线程释放锁（即后一个线程能够取到锁）或者超过规定超时时间再也不取锁。并发
import com.test.core.base.utils.ApplicationUtil;
import com.test.redis.api.RedisStringOperationService;
/**
* redis分布式锁
* @author LIPENG
* @date 2017年9月22日 下午4:25:56
* @version V1.0
*/
public class RedisDistributionLock {
private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;
/**
* Lock key path.
*/
private String lockKey;
/**
* 锁超时时间，防止线程在入锁之后，无限的执行等待
*/
private int expireMsecs = 60 * 1000;
/**
* 锁等待时间，防止线程饥饿
*/
private int timeoutMsecs = 10 * 1000;
private volatile boolean locked = false;
/**
* Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
*
* @param lockKey lock key (ex. account:1, ...)
*/
public RedisDistributionLock(String lockKey) {
this.lockKey = lockKey + "_lock";
}
/**
* Detailed constructor with default lock expiration of 60000 msecs.
*
*/
public RedisDistributionLock(String lockKey, int timeoutMsecs) {
this(lockKey);
this.timeoutMsecs = timeoutMsecs;
}
/**
* Detailed constructor.
*
*/
public RedisDistributionLock(String lockKey, int timeoutMsecs, int expireMsecs) {
this(lockKey, timeoutMsecs);
this.expireMsecs = expireMsecs;
}
/**
* @return lock key
*/
public String getLockKey() {
return lockKey;
}
/**
* 得到 lock.
* 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
* reids缓存的key是锁的key,全部的共享, value是锁的到期时间(注意:这里把过时时间放在value了,没有时间上设置其超时时间)
* 执行过程:
* 1.经过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功得到锁
* 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
*
* @return true if lock is acquired, false acquire timeouted
* @throws InterruptedException in case of thread interruption
*/
public synchronized boolean lock() throws InterruptedException {
int timeout = timeoutMsecs;
while (timeout >= 0) {
long expires = System.currentTimeMillis() + expireMsecs + 1;
String expiresStr = String.valueOf(expires); //锁到期时间
if (getRedisStringOperationService().setNX(lockKey, expiresStr)) {
// lock acquired
locked = true;
return true;
}
String currentValueStr = getRedisStringOperationService().get(lockKey); //redis里的时间
if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
//判断是否为空，不为空的状况下，若是被其余线程设置了值，则第二个条件判断是过不去的
// lock is expired
String oldValueStr = getRedisStringOperationService().getAndSet(lockKey, expiresStr);
//获取上一个锁到期时间，并设置如今的锁到期时间，
//只有一个线程才能获取上一个线上的设置时间，由于jedis.getSet是同步的
if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
//防止误删（覆盖，由于key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，可是由于什么相差了不多的时间，因此能够接受
//[分布式的状况下]:如过这个时候，多个线程刚好都到了这里，可是只有一个线程的设置值和当前值相同，他才有权利获取锁
// lock acquired
locked = true;
return true;
}
}
timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;
/*
延迟100 毫秒, 这里使用随机时间可能会好一点,能够防止饥饿进程的出现,即,当同时到达多个进程,
只会有一个进程得到锁,其余的都用一样的频率进行尝试,后面有来了一些进行,也以一样的频率申请锁,这将可能致使前面来的锁得不到知足.
使用随机的等待时间能够必定程度上保证公平性
*/
Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
}
return false;
}
/**
* 释放锁
*/
public synchronized void unlock() {
if (locked) {
getRedisStringOperationService().delete(lockKey);
locked = false;
}
}
private RedisStringOperationService getRedisStringOperationService(){
return ApplicationUtil.getBean(RedisStringOperationService.class);
}
}
调用方式：
RedisDistributionLock lock = new RedisDistributionLock("key", 10000, 20000);
try {
if (lock.lock()) {
// 须要加锁的代码
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
4、使用zookeeper
当不少进程须要访问共享资源时，咱们能够经过zk来实现分布式锁。主要步骤是：
1.创建一个节点，假如名为：lock 。节点类型为持久节点（PERSISTENT）
2.每当进程须要访问共享资源时，会调用分布式锁的lock()或tryLock()方法得到锁，这个时候会在第一步建立的lock节点下创建相应的顺序子节点，节点类型为临时顺序节点（EPHEMERAL_SEQUENTIAL），经过组成特定的名字name+lock+顺序号。
3.在创建子节点后，对lock下面的全部以name开头的子节点进行排序，判断刚刚创建的子节点顺序号是不是最小的节点，假如是最小节点，则得到该锁对资源进行访问。
4.假如不是该节点，就得到该节点的上一顺序节点，并给该节点是否存在注册监听事件。同时在这里阻塞。等待监听事件的发生，得到锁控制权。
5.当调用完共享资源后，调用unlock（）方法，关闭zk，进而能够引起监听事件，释放该锁。
实现的分布式锁是严格的按照顺序访问的并发锁。
package cn.wpeace.zktest;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;
/**
* @author peace
*
*/
public class DistributedLock implements Lock, Watcher{
private ZooKeeper zk;
private String root = "/locks";//根
private String lockName;//竞争资源的标志
private String waitNode;//等待前一个锁
private String myZnode;//当前锁
private CountDownLatch latch;//计数器
private CountDownLatch connectedSignal=new CountDownLatch(1);
private int sessionTimeout = 30000;
/**
* 建立分布式锁,使用前请确认config配置的zookeeper服务可用
* @param config 192.168.1.127:2181
* @param lockName 竞争资源标志,lockName中不能包含单词_lock_
*/
public DistributedLock(String config, String lockName){
this.lockName = lockName;
// 建立一个与服务器的链接
try {
zk = new ZooKeeper(config, sessionTimeout, this);
connectedSignal.await();
Stat stat = zk.exists(root, false);//此去不执行 Watcher
if(stat == null){
// 建立根节点
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
} catch (IOException e) {
throw new LockException(e);
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
}
/**
* zookeeper节点的监视器
*/
public void process(WatchedEvent event) {
//创建链接用
if(event.getState()==KeeperState.SyncConnected){
connectedSignal.countDown();
return;
}
//其余线程放弃锁的标志
if(this.latch != null) {
this.latch.countDown();
}
}
public void lock() {
try {
if(this.tryLock()){
System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true");
return;
}
else{
waitForLock(waitNode, sessionTimeout);//等待锁
}
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
}
public boolean tryLock() {
try {
String splitStr = "_lock_";
if(lockName.contains(splitStr))
throw new LockException("lockName can not contains \\u000B");
//建立临时子节点
myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(myZnode + " is created ");
//取出全部子节点
List<String> subNodes = zk.getChildren(root, false);
//取出全部lockName的锁
List<String> lockObjNodes = new ArrayList<String>();
for (String node : subNodes) {
String _node = node.split(splitStr)[0];
if(_node.equals(lockName)){
lockObjNodes.add(node);
}
}
Collections.sort(lockObjNodes);
if(myZnode.equals(root+"/"+lockObjNodes.get(0))){
//若是是最小的节点,则表示取得锁
System.out.println(myZnode + "==" + lockObjNodes.get(0));
return true;
}
//若是不是最小的节点，找到比本身小1的节点
String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);//找到前一个子节点
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
return false;
}
public boolean tryLock(long time, TimeUnit unit) {
try {
if(this.tryLock()){
return true;
}
return waitForLock(waitNode,time);
} catch (Exception e) {
e.printStackTrace();
}
return false;
}
private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException {
Stat stat = zk.exists(root + "/" + lower,true);//同时注册监听。
//判断比本身小一个数的节点是否存在,若是不存在则无需等待锁,同时注册监听
if(stat != null){
System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower);
this.latch = new CountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);//等待，这里应该一直等待其余线程释放锁
this.latch = null;
}
return true;
}
public void unlock() {
try {
System.out.println("unlock " + myZnode);
zk.delete(myZnode,-1);
myZnode = null;
zk.close();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
public void lockInterruptibly() throws InterruptedException {
this.lock();
}
public Condition newCondition() {
return null;
}
public class LockException extends RuntimeException {
private static final long serialVersionUID = 1L;
public LockException(String e){
super(e);
}
public LockException(Exception e){
super(e);
}
}
}
调用方法：
DistributedLock lock = new DistributedLock("192.168.1.127:2181","lock");
lock.lock();
//共享资源
if(lock != null)
lock.unlock();