分布式锁的实现与应用场景对比

use test;
CREATE TABLE `DistributedLock` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `name` varchar(64) NOT NULL DEFAULT '' COMMENT '锁名',
  `desc` varchar(1024) NOT NULL DEFAULT '备注信息',
  `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '保存数据时间，自动生成',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uidx_name` (`name`) 
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';

//数据库中的每一条记录就是一把锁，利用的mysql唯一索引的排他性

lock(name,desc){
	insert into DistributedLock(`name`,`desc`) values (#{name},#{desc});
}

unlock(name){
	delete from DistributedLock where name = #{name}
}

获取：再次获取锁的同时更新count（+1）.
释放：更新count-1，当count==0删除记录。

boolean lock(){
    connection.setAutoCommit(false)
    while(true){
        try{
            result = select ... from DistributedLock where name=lock for update;
            if(result==null){
                return true;
            }
        }catch(Exception e){
		connection.commit();
        }
        sleep(*);
    }
    return false;
}

void unlock(){
    connection.commit();
}

 select ...,version 
 update  table set version+1 where version=xx

update personal_bank set account=200  				
where id="xxx" and account=oldAccount

	/**
	 * 加锁代码摘录
	 */
 	@Override
	public boolean tryLock(String lockName, long timeout, TimeUnit unit) throws InterruptedException {

		LockInfo lockInfo = new LockInfo(Thread.currentThread(), lockName);

		// 先判断重入锁
		if (reentrantIfNeed(lockInfo)) {
			return true;
		} else {
			return tryAcquire(lockInfo, timeout, unit);
		}
	}

	/**
	 * @param lockInfo
	 * @param timeout
	 * @param unit
	 * @return
	 * @throws InterruptedException
	 */
	private boolean tryAcquire(LockInfo lockInfo, long timeout, TimeUnit unit) throws InterruptedException {
		// lock first time
		Long loopTimeout = (unit != null) ? unit.toMillis(timeout) : timeout;
		long startMillis = System.currentTimeMillis();
		boolean isAcqired = false;
		do {
			String result = getCache().set(lockInfo.getName(), lockInfo.getValue(), SET_IF_NOT_EXIST,
					SET_EXPIRE_MILLISECONDS, LOCK_EXPIRE_MILLSECOND);
			if ("OK".equals(result)) {
				isAcqired = true;
				break;
			}
			TimeUnit.NANOSECONDS.sleep(RETRY_INTERVAL);

		} while (System.currentTimeMillis() - startMillis > loopTimeout);

		if (isAcqired) {
			allLockInfo.add(lockInfo);
		}
		return isAcqired;
	}


	/**
	 * 解锁代码摘录
	 */


	protected static final String UNLOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return -1 end";

	@Override
	public void unlock(String lockName) {
		Thread currentThread = Thread.currentThread();

		LockInfo lockInfo = new LockInfo(currentThread, lockName);

		LockInfo locked = getExistedLock(lockInfo);

		if (null == locked) {
			throw new IllegalMonitorStateException("currentThread : [ " + currentThread + " ] unlock : [ " + lockName
					+ " ] fail because of not owner.");
		}
		int reentrantTimes = locked.reentrantMinusOne();
		if (reentrantTimes > 0) {
			return;
		} else if (reentrantTimes < 0) {
			throw new IllegalMonitorStateException("currentThread : [ " + currentThread + " ] unlock : [ " + lockName
					+ " ] fail because of reentrant can't be negative.");
		} else {
			Object result = getCache().eval(UNLOCK_LUA_SCRIPT, Collections.singletonList(lockName),
					Collections.singletonList(locked.getValue()));
			if (!UNLOCK_FAIL.equals(result)) {
				allLockInfo.remove(locked);
			}
		}
	}

/**
* @author zhangsh
*/
public class RedisWatchLock {
private static final String redisHost = "127.0.0.1";
private static final int port = 6379;
private static JedisPoolConfig config;
private static JedisPool pool;
private static ExecutorService service;
private static int count = 10;
private static CountDownLatch latch;
private static AtomicInteger Countor = new AtomicInteger(0);
static {
config = new JedisPoolConfig();
config.setMaxIdle(10);
config.setMaxWaitMillis(1000);
config.setMaxTotal(30);
pool = new JedisPool(config, redisHost, port);
service = Executors.newFixedThreadPool(10);
latch = new CountDownLatch(count);
}
public static void main(String args[]) {
int count = 10;
String ThreadNamePrefix = "thread-";
Jedis cli = pool.getResource();
cli.del("redis_inc_key");// 先删除既定的key
cli.set("redis_inc_key", String.valueOf(1));// 设定默认值
for (int i = 0; i < count; i++) {
Thread th = new Thread(new TestThread(pool));
th.setName(ThreadNamePrefix + i);
System.out.println(th.getName() + "inited...");
service.submit(th);
}
service.shutdown();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("all sub thread sucess");
System.out.println("countor is " + Countor.get());
String countStr = cli.get("redis_inc_key");
System.out.println(countStr);
}
public static class TestThread implements Runnable {
private String incKeyStr = "redis_inc_key";
private Jedis cli;
private JedisPool pool;
public TestThread(JedisPool pool) {
cli = pool.getResource();
this.pool = pool;
}
public void run() {
try {
for (int i = 0; i < 100; i++) {
actomicAdd();//生产环境中批量操作尽量使用redisPipeLine!!
}
} catch (Exception e) {
pool.returnBrokenResource(cli);
} finally {
pool.returnResource(cli);
latch.countDown();
}
}
/**
* 0 watch key 
* 1 multi
* 2 set key value(queued)
* 3 exec
* 
* return null：fail
* reurn  "ok": succeed
* 
* watch每次都需要执行(注册)
*/
public void actomicAdd() {
cli.watch(incKeyStr);// 0.watch key
boolean flag = true;
while (flag) {
String countStr = cli.get("redis_inc_key");
int countInt = Integer.parseInt(countStr);
int expect = countInt + 1;
Transaction tx = cli.multi(); // 1.multi
tx.set(incKeyStr, String.valueOf(expect));// 2.set key value
// (queued)
List<Object> list = tx.exec();// 3.exec
if (list == null) {
System.out.println("fail");
continue;
} else {
flag = false;
System.out.println("succeed");
}
System.out.println("my expect num is " + expect);
System.out.println("seting....");
}
Countor.incrementAndGet();
}
}
}