zookeeper分布式锁

首先搭建zookeeper集群docker-compose.yml

version: '2'networks:  zk:services:  zk1:    image: zookeeper:3.4    container_name: zk1    networks:        - zk    ports:        - "21811:2181"    environment:      ZOO_MY_ID: 1      ZOO_SERVERS: server.1=0.0.0.0:2888:3888 server.2=zk2:2888:3888 server.3=zk3:2888:3888  zk2:    image: zookeeper:3.4    container_name: zk2    networks:        - zk    ports:        - "21812:2181"    environment:      ZOO_MY_ID: 2      ZOO_SERVERS: server.1=zk1:2888:3888 server.2=0.0.0.0:2888:3888 server.3=zk3:2888:3888  zk3:    image: zookeeper:3.4    container_name: zk3    networks:        - zk    ports:        - "21813:2181"    environment:      ZOO_MY_ID: 3      ZOO_SERVERS: server.1=zk1:2888:3888 server.2=zk2:2888:3888 server.3=0.0.0.0:2888:3888

docker-compose up -d 创建并启动
检查状态

[root@localhost test]# docker exec -it zk1 bash ./bin/zkServer.sh statusZooKeeper JMX enabled by defaultUsing config: /conf/zoo.cfgMode: follower[root@localhost test]# docker exec -it zk2 bash ./bin/zkServer.sh statusZooKeeper JMX enabled by defaultUsing config: /conf/zoo.cfgMode: follower[root@localhost test]# docker exec -it zk3 bash ./bin/zkServer.sh statusZooKeeper JMX enabled by defaultUsing config: /conf/zoo.cfgMode: leader

下面我们来程序验证：

我们用现成的curator来操作zk分布式锁

import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.api.GetDataBuilder;import org.apache.curator.framework.recipes.locks.InterProcessMutex;import org.apache.curator.retry.ExponentialBackoffRetry;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.sql.Time;import java.util.Date;import java.util.concurrent.TimeUnit;public class DistributedLock {        public static Logger log = LoggerFactory.getLogger(DistributedLock.class);        private InterProcessMutex interProcessMutex;  //可重入排它锁        private String lockName;  //竞争资源标志        private String root = "/distributed/lock/";//根节点        private static CuratorFramework curatorFramework;        private static String ZK_URL = "127.0.0.1:21811,127.0.0.1:21812,127.0.0.1:21813";        static{            curatorFramework= CuratorFrameworkFactory.newClient(ZK_URL,new ExponentialBackoffRetry(1000,3));            curatorFramework.start();        }        /**         * 实例化         * @param lockName         */        public DistributedLock(String lockName){            try {                this.lockName = lockName;                interProcessMutex = new InterProcessMutex(curatorFramework, root + lockName);            }catch (Exception e){                log.error("initial InterProcessMutex exception="+e);            }        }        /**         * 获取锁         */        public void acquireLock(){            int flag = 0;            try {                //重试N次，每次最大等待1s                while (!interProcessMutex.acquire(1, TimeUnit.SECONDS)){                    flag++;                    if(flag>5){  //重试次                        break;                    }                }            } catch (Exception e) {                log.error("distributed lock acquire exception="+e);            }            if(flag>5){                log.info("Thread:"+Thread.currentThread().getId()+" acquire distributed lock  busy"+ new Date().getTime());            }else{                log.info("Thread:"+Thread.currentThread().getId()+" acquire distributed lock  success"+ new Date().getTime());            }        }        /**         * 释放锁         */        public void releaseLock(){            try {                if(interProcessMutex != null && interProcessMutex.isAcquiredInThisProcess()){                    interProcessMutex.release();                    curatorFramework.delete().inBackground().forPath(root+lockName);                    //byte[] data = curatorFramework.getData().forPath(root + lockName);                    log.info("Thread:"+Thread.currentThread().getId()+" release distributed lock  success"+ new Date().getTime());                }            }catch (Exception e){                log.info("Thread:"+Thread.currentThread().getId()+" release distributed lock  exception="+e);            }        }    }

接下来我们开2个线程来竞争分布式锁：

public class TestLock {    public static void main(String[] args) throws InterruptedException {        String lockName = "lock1";        DistributedLock lockFoo = new DistributedLock(lockName);        //lockFoo.acquireLock();        //lockFoo.releaseLock();        //        System.out.println("主线程ID是：" + Thread.currentThread().getId());        Thread thread1 = new MyThread("thread1",lockFoo);        Thread thread2 = new MyThread("thread2",lockFoo);        thread1.start();        Thread.sleep(1000);        thread2.start();    }}/** * 自定义线程 */class MyThread extends Thread {    /*线程名称*/    private String name;    private DistributedLock lockFoo;    public MyThread(String name,DistributedLock lockFoo) {        this.name = name;        this.lockFoo = lockFoo;    }    @Override    public void run() {        if(this.name.equals("thread1")) {            this.lockFoo.acquireLock();        }        System.out.println("名称" + name + "的线程ID是：" + Thread.currentThread().getId());        try {            Thread.sleep(3000);        } catch (InterruptedException e) {            e.printStackTrace();        }        if(this.name.equals("thread2")) {            this.lockFoo.releaseLock();        }    }}

程序中thread1获取了锁后，thread2解锁失败。

更多相关文章

随机推荐