提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档
文章目录- zk实现分布式锁
- 1.zk中锁的种类：
- 2.zk如何上读锁
- 3.zk如何上写锁
- 4.⽺群效应
- 可以调整成链式监听。解决这个问题。
- 5.curator实现读写锁
- 分布式锁案例
- 案例分析
- 依赖：
- 分布式锁实现
- 测试
- 对比单体模式下---ReentrantLock
- Curator框架实现分布式锁案例
- 依赖
- 获取客户端连接
- 测试案例
zk实现分布式锁 1.zk中锁的种类：
- 读锁：⼤家都可以读，要想上读锁的前提：之前的锁没有写锁
- 写锁：只有得到写锁的才能写。要想上写锁的前提是，之前没有任何锁。
1.创建⼀个临时序号节点，节点的数据是read，表示是读锁
2.获取当前zk中序号⽐⾃⼰⼩的所有节点
3.判断最⼩节点是否是读锁：
- 如果不是读锁的话，则上锁失败，为最⼩节点设置监听。阻塞等待，zk的watch机制
会当最⼩节点发⽣变化时通知当前节点，于是再执⾏第⼆步的流程 - 如果是读锁的话，则上锁成功
1.创建⼀个临时序号节点，节点的数据是write，表示是 写锁
2.获取zk中所有的⼦节点
3.判断⾃⼰是否是最⼩的节点：
- 如果是，则上写锁成功
- 如果不是，说明前⾯还有锁，则上锁失败，监听最⼩的节点，如果最⼩节点有变化， 则回到第⼆步。
如果⽤上述的上锁⽅式，只要有节点发⽣变化，就会触发其他节点的监听事件，这样的话对
zk的压⼒⾮常⼤，——⽺群效应。
可以调整成链式监听。解决这个问题。5.curator实现读写锁
import org.apache.curator.framework.Curatorframework;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
public class TestReadWriteLock {
@Autowired
private Curatorframework client;
@Test
void testGetReadLock() throws Exception {
// 读写锁
InterProcessReadWriteLock interProcessReadWriteLock=new InterProcessReadWriteLock(client, "/lock1");
// 获取读锁对象
InterProcessLock interProcessLock=interProcessReadWriteLock.readLock();
System.out.println("等待获取读锁对象!");
// 获取锁
interProcessLock.acquire();
for (int i = 1; i <= 100; i++) {
Thread.sleep(3000);
System.out.println(i);
}
// 释放锁
interProcessLock.release();
System.out.println("等待释放锁!");
}
@Test
void testGetWriteLock() throws Exception {
// 读写锁
InterProcessReadWriteLock interProcessReadWriteLock=new InterProcessReadWriteLock(client, "/lock1");
// 获取写锁对象
InterProcessLock interProcessLock=interProcessReadWriteLock.writeLock();
System.out.println("等待获取写锁对象!");
// 获取锁
interProcessLock.acquire();
for (int i = 1; i <= 100; i++) {
Thread.sleep(3000);
System.out.println(i);
}
// 释放锁
interProcessLock.release();
System.out.println("等待释放锁!");
}
}
分布式锁案例
案例分析
- 比如说 "进程 1"在使用该资源的时候，会先去获得锁在使用该资源的时候，保持独占，这样其他进程就无法访问该资源，
- "进程1"用完该资源以后就将锁释放掉，让其他进程来获得锁，
- 那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。
- 接收到请求后，在/locks节点下创建一个临时顺序节点
- 判断自己是不是当前节点下最小的节点：是，获取到锁；不是，对前一个节点进行监听
- 获取到锁，处理完业务后，delete节点释放锁，然后下面的节点将收到通知，重复第二步判断
分布式锁实现org.apache.zookeeper zookeeper3.5.7
- 获取连接
- 对zk加锁
- 对zk解锁
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class DistributedLock {
private final String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
private final int sessionTimeout = 2000;
private final ZooKeeper zk;
private CountDownLatch connectLatch = new CountDownLatch(1);
private CountDownLatch waitLatch = new CountDownLatch(1);
private String waitPath;
private String currentMode;
public DistributedLock() throws IOException, InterruptedException, KeeperException {
// 获取连接
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
// connectLatch 如果连接上zk 可以释放
if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
connectLatch.countDown();
}
// waitLatch 需要释放
if (watchedEvent.getType()== Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
waitLatch.countDown();
}
}
});
// 等待zk正常连接后，往下走程序
connectLatch.await();
// 判断根节点/locks是否存在
Stat stat = zk.exists("/locks", false);
if (stat == null) {
// 创建一下根节点
zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
}
// 对zk加锁
public void zklock() {
// 创建对应的临时带序号节点
try {
currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// wait一小会, 让结果更清晰一些
Thread.sleep(10);
// 判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点
List children = zk.getChildren("/locks", false);
// 如果children 只有一个值，那就直接获取锁； 如果有多个节点，需要判断，谁最小
if (children.size() == 1) {
return;
} else {
Collections.sort(children);
// 获取节点名称 seq-00000000
String thisNode = currentMode.substring("/locks/".length());
// 通过seq-00000000获取该节点在children集合的位置
int index = children.indexOf(thisNode);
// 判断
if (index == -1) {
System.out.println("数据异常");
} else if (index == 0) {
// 就一个节点，可以获取锁了
return;
} else {
// 需要监听 他前一个节点变化
waitPath = "/locks/" + children.get(index - 1);
zk.getData(waitPath,true,new Stat());
// 等待监听
waitLatch.await();
return;
}
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 解锁
public void unZkLock() {
// 删除节点
try {
zk.delete(this.currentMode,-1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
}
测试
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
public class DistributedLockTest {
public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
final DistributedLock lock1 = new DistributedLock();
final DistributedLock lock2 = new DistributedLock();
new Thread(new Runnable() {
@Override
public void run() {
try {
lock1.zklock();
System.out.println("线程1 启动，获取到锁");
Thread.sleep(5 * 1000);
lock1.unZkLock();
System.out.println("线程1 释放锁");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
lock2.zklock();
System.out.println("线程2 启动，获取到锁");
Thread.sleep(5 * 1000);
lock2.unZkLock();
System.out.println("线程2 释放锁");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
对比单体模式下—ReentrantLock
Curator框架实现分布式锁案例
依赖
获取客户端连接org.apache.curator curator-framework4.3.0 org.apache.curator curator-recipes4.3.0 org.apache.curator curator-client4.3.0
private static Curatorframework getCuratorframework() {
ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
Curatorframework client = CuratorframeworkFactory.builder().connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181")
.connectionTimeoutMs(2000)
.sessionTimeoutMs(2000)
.retryPolicy(policy).build();
// 启动客户端
client.start();
System.out.println("zookeeper 启动成功");
return client;
}
测试案例
import org.apache.curator.framework.Curatorframework;
import org.apache.curator.framework.CuratorframeworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class CuratorLockTest {
public static void main(String[] args) {
// 创建分布式锁1
InterProcessMutex lock1 = new InterProcessMutex(getCuratorframework(), "/locks");
// 创建分布式锁2
InterProcessMutex lock2 = new InterProcessMutex(getCuratorframework(), "/locks");
new Thread(new Runnable() {
@Override
public void run() {
try {
lock1.acquire();
System.out.println("线程1 获取到锁");
lock1.acquire();
System.out.println("线程1 再次获取到锁");
Thread.sleep(5 * 1000);
lock1.release();
System.out.println("线程1 释放锁");
lock1.release();
System.out.println("线程1 再次释放锁");
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
lock2.acquire();
System.out.println("线程2 获取到锁");
lock2.acquire();
System.out.println("线程2 再次获取到锁");
Thread.sleep(5 * 1000);
lock2.release();
System.out.println("线程2 释放锁");
lock2.release();
System.out.println("线程2 再次释放锁");
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
private static Curatorframework getCuratorframework() {
ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
Curatorframework client = CuratorframeworkFactory.builder().connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181")
.connectionTimeoutMs(2000)
.sessionTimeoutMs(2000)
.retryPolicy(policy).build();
// 启动客户端
client.start();
System.out.println("zookeeper 启动成功");
return client;
}
}
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）