黑马程序员技术交流社区

标题: 【广州校区】+【原创】+ 分布式锁控制多线程有序输出 [打印本页]

---

作者: 小李哥    时间: 2019-7-25 11:37
标题: 【广州校区】+【原创】+ 分布式锁控制多线程有序输出
当很多进程需要访问共享资源时，我们可以通过zk来实现分布式锁。主要步骤是：
1.建立一个节点，假如名为：lock 。节点类型为持久节点（PERSISTENT）
2.每当进程需要访问共享资源时，会调用分布式锁的lock()或tryLock()方法获得锁，这个时候会在第一步创建的lock节点下建立相应的顺序子节点，节点类型为临时顺序节点（EPHEMERAL_SEQUENTIAL），通过组成特定的名字name+lock+顺序号。
3.在建立子节点后，对lock下面的所有以name开头的子节点进行排序，判断刚刚建立的子节点顺序号是否是最小的节点，假如是最小节点，则获得该锁对资源进行访问。
4.假如不是该节点，就获得该节点的上一顺序节点，并给该节点是否存在注册监听事件。同时在这里阻塞。等待监听事件的发生，获得锁控制权。
5.当调用完共享资源后，调用unlock（）方法，关闭zk，进而可以引发监听事件，释放该锁。
实现的分布式锁是严格的按照顺序访问的并发锁。


以下代码实现：
package com.itheima.jishutie;

public interface ExtLock {
   
    //ExtLock基于zk实现分布式锁
    public void  getLock() throws InterruptedException;
   
    //释放锁
    public void unLock();
   
}


package com.itheima.jishutie;

import org.I0Itec.zkclient.ZkClient;

//将重复代码抽象到子类中（模板方法设计模式）
public abstract class ZookeeperAbstractLock implements ExtLock {
    public static final String CONNECTION="192.168.12.131:2181";
    protected ZkClient zkClient = new ZkClient(CONNECTION);
    public  String lockPath="/lockPath";


     //获取锁
      public void getLock() throws InterruptedException {
          //1、连接zkClient 创建一个/lock的临时节点
          // 2、 如果节点创建成果，直接执行业务逻辑，如果节点创建失败，进行等待
          if (tryLock()) {
            System.out.println("#####成功获取锁######");
        }else {
            //进行等待
            waitLock();
        }
     
        //3、使用事件通知监听该节点是否被删除    ，如果是，重新进入获取锁的资源  
        
    }
      
   //创建失败 进行等待
    abstract void waitLock() throws InterruptedException;


    abstract boolean tryLock();
     
     
    //释放锁
      public void unLock() {
        //执行完毕 直接连接
          if (zkClient != null) {
              System.out.println("######释放锁完毕######");
            zkClient.close();

        }
        
    }
      
}





package com.itheima.jishutie;

import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.IZkDataListener;

public class ZookeeperDistrbuteLock extends ZookeeperAbstractLock {
    private static  ThreadLocal<CountDownLatch> countDownLatch = new ThreadLocal<CountDownLatch>() {
        @Override
        protected CountDownLatch initialValue() {
            return new CountDownLatch(1);
        }
    };
    @Override
    boolean tryLock() {
        try {
            zkClient.createEphemeral(lockPath);
//            System.out.println("#########获取锁######");
            return true;
        } catch (Exception e) {
            // 如果失败 直接catch
            return false;
        }
    }

    @Override
    void waitLock()  {

        IZkDataListener iZkDataListener = new IZkDataListener() {

            // 节点被删除
            public void handleDataDeleted(String arg0) throws Exception {
                if (countDownLatch.get() != null) {
                    countDownLatch.get().countDown(); // 计数器为0的情况，await 后面的继续执行
                }

            }

            // 节点被修改
            public void handleDataChange(String arg0, Object arg1) throws Exception {

            }
        };

        // 监听事件通知
        zkClient.subscribeDataChanges(lockPath, iZkDataListener);
        // 控制程序的等待
        if (zkClient.exists(lockPath)) {  //如果 检查出 已经被创建了 就new 然后进行等待



            if (countDownLatch.get() != null) {
                countDownLatch.remove();
            }
            countDownLatch.set(new CountDownLatch(1));
            try {
                countDownLatch.get().wait(); //等待时候 就不往下走了   当为0 时候 后面的继续执行
            } catch (InterruptedException e) {
           //     e.printStackTrace();
            }


        }
        //后面代码继续执行
        //为了不影响程序的执行 建议删除该事件监听 监听完了就删除掉
        zkClient.unsubscribeDataChanges(lockPath, iZkDataListener);

    }

}
package com.itheima.jishutie;

import java.text.SimpleDateFormat;
import java.util.Date;

//生成订单号 时间戳
public class OrderNumGenerator {
  //区分不同的订单号
    private static int count = 0;
//单台服务器，多个线程 同事生成订单号
    public String getNumber(){

        SimpleDateFormat simpt = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");  
        return simpt.format(new Date()) + "-" + ++count;  //时间戳后面加了 count

    }
   
}
package com.itheima.jishutie;

public class OrderService implements Runnable {

    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator(); // 定义成全局的
    private ExtLock lock = new ZookeeperDistrbuteLock();

    public void run() {
        getNumber();
    }

    public synchronized void getNumber() { // 加锁 保证线程安全问题 让一个线程操作
        try {
            lock.getLock();
            String number = orderNumGenerator.getNumber();

            System.out.println("順序输出:" + number);

        } catch (Exception e) {

        } finally {
            lock.unLock();
        }
    }

    public static void main(String[] args) {
//        OrderService orderService = new OrderService();
        for (int i = 0; i < 100; i++) { // 开启100个线程
            //模拟分布式锁的场景
            new Thread(new OrderService()).start();
        }
    }

}


控制台输出结果：######成功获取锁######
順序输出:2019-07-25-11-30-16-1######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-2######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-3######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-4######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-5######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-6######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-7######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-8######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-9######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-10######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-11######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-12######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-13######释放锁完毕###########成功获取锁######順序输出:2019-07-25-11-30-16-14

---

欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2