标签：Thread lock ZooKeeper 线程 应用 org import public 分布式
一、Zookeeper是什么
Zookeeper是一个高性能的分布式系统的协调服务。它在一个简单的接口里暴露公共服务：像命名、配置管理、同步、和群组服务，所以你没有必要从头开始实现它们。你可以使用现成的Zookeeper去实现共识、群组管理、领导人选举和业务协议。并且你可以在它的基础之上建立自己特定的需求。
二、Zookeeper分布式锁的实现原理
利用临时顺序节点实现Zookeeper分布式锁。
1、首先为一个lock场景，在zookeeper中指定对应的一个根节点，用于记录资源竞争的内容，称之为/lock_node
2、每个lock创建后，会lazy在zookeeper中创建一个node节点，表明对应的资源竞争标识。 (小技巧：node节点为EPHEMERAL_SEQUENTIAL，自增长的临时节点)。比如有两个客户端创建znode，那分别为/lock_node/lock-1、/lock_node/lock-2
3、进行lock操作时，获取对应lock根节点下的所有字节点，也即处于竞争中的资源标识
4、按照Fair竞争的原则，按照对应的自增内容做排序，取出编号最小的一个节点做为lock的owner，判断自己的节点id是否就为owner id，如果是则返回，lock成功。
5、如果自己非owner id，按照排序的结果找到序号比自己前一位的id，关注它锁释放的操作(也就是exist watcher)，形成一个链式的触发过程。
unlock过程
6、将自己id对应的节点删除即可，对应的下一个排队的节点就可以收到Watcher事件，从而被唤醒得到锁后退出
ZooKeeper的几个特性让它非常合适作为分布式锁服务
zookeeper支持watcher机制，这样实现阻塞锁，可以watch锁数据，等到数据被删除，zookeeper会通知客户端去重新竞争锁。
zookeeper的数据可以支持临时节点的概念，即客户端写入的数据是临时数据，在客户端宕机后，临时数据会被删除，这样就实现了锁的异常释放。使用这样的方式，就不需要给锁增加超时自动释放的特性了。
三、Zookeeper分布式锁应用
Curator是Netflix公司开源的一个Zookeeper客户端，与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。
使用场景
1、创建ZooKeeperConnector.java
package com.robot.zookeeper.components;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author: 会跳舞的机器人
* @date: 2017/6/21 18:07
* @description:ZooKeeper连接器
*/
public class ZooKeeperConnector {
private Logger logger = LoggerFactory.getLogger(this.getClass());
/**
* 连接IP端口信息，格式为：10.1.74.75:2281,10.1.74.75:2282,10.1.74.75:2283
*/
private String hosts;
private CuratorFramework client;
private static final int DEFAULT_SESSION_TIMEOUT_MS = 30 * 1000;
private static final int DEFAULT_CONNECTION_TIMEOUT_MS = 10 * 1000;
private int sessionTimeout = DEFAULT_SESSION_TIMEOUT_MS;
private int connectionTimeout = DEFAULT_CONNECTION_TIMEOUT_MS;
/**
* 连接ZooKeeper
*/
public void connect() {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
client = CuratorFrameworkFactory.newClient(hosts, sessionTimeout, connectionTimeout, retryPolicy);
client.start();
logger.info("Successfully connected to Zookeeper [{}] ", hosts);
}
/**
* 关闭ZooKeeper的连接
*/
public void close() {
CloseableUtils.closeQuietly(client);
}
/**
* 重连
*
* @return
*/
public CuratorFramework reConnect() {
connect();
return client;
}
public String getHosts() {
return hosts;
}
public void setHosts(String hosts) {
this.hosts = hosts;
}
public CuratorFramework getClient() {
if (client == null) {
connect();
}
return client;
}
public void setClient(CuratorFramework client) {
this.client = client;
}
public int getSessionTimeout() {
return sessionTimeout;
}
public void setSessionTimeout(int sessionTimeout) {
this.sessionTimeout = sessionTimeout;
}
public int getConnectionTimeout() {
return connectionTimeout;
}
public void setConnectionTimeout(int connectionTimeout) {
this.connectionTimeout = connectionTimeout;
}
}2、创建AccountLock.java
package com.robot.zookeeper.utils;
import com.robot.zookeeper.components.ZooKeeperConnector;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
/**
* @author: 会跳舞的机器人
* @date: 2017/6/22 10:16
* @description:账户分布式锁
*/
public class AccountLock {
private Logger logger = LoggerFactory.getLogger(this.getClass());
/**
* 共享锁
*/
InterProcessMutex lock;
/**
* 是否获取到锁
*/
boolean wasAcquired = false;
/**
* 构造器
*
* @param path
* @param zooKeeperConnector
*/
public AccountLock(String path, ZooKeeperConnector zooKeeperConnector) {
lock = new InterProcessMutex(zooKeeperConnector.getClient(), path);
}
/**
* 尝试加锁并等待
*
* @param timeOut 超时时间（秒），-1表示不等待
* @return true表示获取锁成功，false表示获取锁失败
*/
public boolean acquire(int timeOut) {
try {
if (timeOut == -1) {
wasAcquired = lock.acquire(-1, null);
} else {
wasAcquired = lock.acquire(timeOut, TimeUnit.SECONDS);
}
} catch (Exception e) {
logger.error("Get lock time out error", e.getMessage());
wasAcquired = false;
}
return wasAcquired;
}
/**
* 尝试加锁并等待
*
* @param timeOut timeOut 超时时间，-1表示不等待
* @param timeUnit 超时时间单位
* @return true表示获取锁成功，false表示获取锁失败
*/
public boolean acquire(int timeOut, TimeUnit timeUnit) {
try {
wasAcquired = lock.acquire(timeOut, timeUnit);
} catch (Exception e) {
logger.error("Get lock time out error", e.getMessage());
wasAcquired = false;
}
return wasAcquired;
}
/**
* 释放锁
*/
public void release() {
if (wasAcquired) {
try {
lock.release();
} catch (Exception e) {
logger.error("release lock error", e.getMessage());
}
}
}
}3、测试类
package com.robot.zookeeper;
import com.robot.zookeeper.components.ZooKeeperConnector;
import com.robot.zookeeper.utils.AccountLock;
/**
* @author: 会跳舞的机器人
* @date: 2017/6/22 10:34
* @description:
*/
public class Test {
public static void main(String[] args) {
final ZooKeeperConnector zooKeeperConnector = new ZooKeeperConnector();
zooKeeperConnector.setHosts("192.168.133.128:2181,192.168.133.129:2182,192.168.133.130:2183");
zooKeeperConnector.connect();
/**
* 创建4个线程去获取锁
*/
for (int i = 1; i < 5; i++) {
Thread thread = new Thread() {
@Override
public void run() {
AccountLock accountLock = new AccountLock("/ACCOUNT/221890", zooKeeperConnector);
boolean wasAcquired = accountLock.acquire(10);
if (wasAcquired) {
System.out.println("线程" + Thread.currentThread().getName() + "获取到锁");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
accountLock.release();
System.out.println("线程" + Thread.currentThread().getName() + "释放锁");
}
}
};
thread.start();
}
}
}输出结果：
线程Thread-3获取到锁
线程Thread-3释放锁
线程Thread-4获取到锁
线程Thread-4释放锁
线程Thread-2获取到锁
线程Thread-2释放锁
线程Thread-1获取到锁
线程Thread-1释放锁账户加锁的时候，我们针对用户的ID进行加锁，在测试类中，我们创建了4个线程去获取锁，从输出结果可以看出每次只有一个线程能获取到锁，并且在该线程释放锁之后，其他的线程才能获取到锁。
当然，测试类中的ZooKeeperConnector的初始化一般都是通过Spring进行管理
<beans>
<bean id="zkConnector" class="com.baibei.component.zk.ZooKeeperConnector"
init-method="connect" lazy-init="false">
<property name="hosts" value="#{environment['ZOOKEEPER.CONNECTION.HOSTS']}" />
</bean>
</beans>Demo中所需要的maven配置如下：
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.10.0</version>
</dependency>
<!-- curator的内嵌包版本存在问题，所以用这个版本来替代-->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>18.0</version>
</dependency>原文转自：https://www.jianshu.com/p/11b81630990b
作者：会跳舞的机器人
标签：Thread,lock,ZooKeeper,线程,应用,org,import,public,分布式 来源： https://blog.51cto.com/mageedu/2532104
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。