分布式锁Demo

桃扇骨 2023-06-10 13:23 6阅读 0赞

**目录**

序列化

模拟订单生成

获取单例订单生成类

创建订单接口

基于zookeeper实现的公平分布式锁

测试入口

--------------------

### 序列化 ###

package test.lock;
    
    import org.I0Itec.zkclient.exception.ZkMarshallingError;
    import org.I0Itec.zkclient.serialize.ZkSerializer;
    
    import java.io.UnsupportedEncodingException;
    
    /**
     * 序列化
     */
    public class MyZkSerializer implements ZkSerializer {
    
    	@Override
    	public byte[] serialize(Object data) throws ZkMarshallingError {
    		String d = (String) data;
    		try {
    			return d.getBytes("UTF-8");
    		} catch (UnsupportedEncodingException e) {
    			e.printStackTrace();
    		}
    		return null;
    	}
    
    	@Override
    	public Object deserialize(byte[] bytes) throws ZkMarshallingError {
    		try {
    			return new String(bytes, "UTF-8");
    		} catch (UnsupportedEncodingException e) {
    			e.printStackTrace();
    		}
    		return null;
    	}
    
    }

### 模拟订单生成 ###

package test.lock;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    /**
     * 订单编号生成类
     * OrderCodeGenerator
     */
    public class OrderCodeGenerator {
    	// 自增长序列
    	private int i = 0;
    
    
    
    	// 按照“年-月-日-小时-分钟-秒-自增长序列”的规则生成订单编号
    	public String getOrderCode() {
    		Date now = new Date();
    		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-");
    		String result = sdf.format(now) + i++;
    		return result;
    	}
    	
    	
    
    	public static void main(String[] args) {
    		OrderCodeGenerator ong = new OrderCodeGenerator();
    		for (int i = 0; i < 10; i++) {
    			System.out.println(ong.getOrderCode());
    		}
    	}
    
    }

### 获取单例订单生成类  ###

package test.lock;
    
    /**
     * 获取单例订单生成类
     * OrderCodeGenerator
     */
    public class OrderCodeGeneratorSingle {
    	
    	static class InstanceHolder {
    		private static OrderCodeGenerator instance = new OrderCodeGenerator();
    	}
    	
    	public static OrderCodeGenerator getInstance() {
    		return InstanceHolder.instance;
    	}
    }

###  创建订单接口 ###

package test.lock;
    
    
    import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.locks.Lock;
    
    
    /**
     * 创建订单接口
     */
    public class OrderServiceImplWithDistributeLock {
    	
    	private OrderCodeGenerator ocg = OrderCodeGeneratorSingle.getInstance();
    
    
    	//公平分布式锁
    	private Lock lock = new ZkFairLock("/distributeLock");
    
    	// 重复编号集合
    	private static Set<String> codeSet = new HashSet<String>();
    
    	// 创建订单接口
    	public void createOrder() {
    
    		String orderCode = null;
    		try {
    			lock.lock();
    			// 获取订单号
    			orderCode = ocg.getOrderCode();
    			
    			if(codeSet.contains(orderCode)) {
    				System.err.println("重复编号："+orderCode);
    			}else{
    				codeSet.add(orderCode);
    				System.out.println("没有重复编码：" + orderCode);
    			}
    
    		} finally {
    			lock.unlock();
    		}
    
    		System.out.println(Thread.currentThread().getName() + " =============>" + orderCode);
    
    		
    
    	}
    
    }

### 基于zookeeper实现的公平分布式锁  ###

package test.lock;
    
    import org.I0Itec.zkclient.IZkDataListener;
    import org.I0Itec.zkclient.ZkClient;
    import org.I0Itec.zkclient.exception.ZkNodeExistsException;
    
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    
    /**
     * 基于zookeeper实现的公平锁
     */
    public class ZkFairLock implements Lock {
    
    	/*
    	 * 利用临时顺序节点来实现分布式锁
    	 * 获取锁：取排队号（创建自己的临时顺序节点），然后判断自己是否是最小号，如是，则获得锁；不是，则注册前一节点的watcher,阻塞等待
    	 * 释放锁：删除自己创建的临时顺序节点
    	 */
    	private String lockPath;
    
    	private ZkClient client;
    	
    	private ThreadLocal<String> currentPath = new ThreadLocal<String>();
    
    	private ThreadLocal<String> beforePath = new ThreadLocal<String>();
    	// 锁重入计数器
    	private ThreadLocal<Integer> reenterCount = ThreadLocal.withInitial(()->0);
    
    	public ZkFairLock(String lockPath) {
    		if(lockPath == null || lockPath.trim().equals("")) {
    			throw new IllegalArgumentException("patch不能为空字符串");
    		}
    		this.lockPath = lockPath;
    		client = new ZkClient("127.0.0.1:2181");
    		client.setZkSerializer(new MyZkSerializer());
    		if (!this.client.exists(lockPath)) {
    			try {
    				this.client.createPersistent(lockPath, true);
    			} catch (ZkNodeExistsException e) {
    
    			}
    		}
    	}
    
    	@Override
    	public boolean tryLock() {
    		System.out.println(Thread.currentThread().getName() + "-----尝试获取分布式锁");
    		if (this.currentPath.get() == null || !client.exists(this.currentPath.get())) {
    			String node = this.client.createEphemeralSequential(lockPath + "/", "locked");
    			currentPath.set(node);
    			reenterCount.set(0);
    		}
    		
    		// 获得所有的子
    		List<String> children = this.client.getChildren(lockPath);
    
    		// 排序list
    		Collections.sort(children);
    
    		// 判断当前节点是否是最小的
    		if (currentPath.get().equals(lockPath + "/" + children.get(0))) {
    			// 锁重入计数
    			reenterCount.set(reenterCount.get() + 1);
    			System.out.println(Thread.currentThread().getName() + "-----获得分布式锁");
    			return true;
    		} else {
    			// 取到前一个
    			// 得到字节的索引号
    			int curIndex = children.indexOf(currentPath.get().substring(lockPath.length() + 1));
    			String node = lockPath + "/" + children.get(curIndex - 1);
    			beforePath.set(node);
    		}
    		return false;
    	}
    
    	@Override
    	public void lock() {
    		if (!tryLock()) {
    			// 阻塞等待
    			waitForLock();
    			// 再次尝试加锁
    			lock();
    		}
    	}
    
    	private void waitForLock() {
    
    		CountDownLatch cdl = new CountDownLatch(1);
    
    		// 注册watcher
    		IZkDataListener listener = new IZkDataListener() {
    
    			@Override
    			public void handleDataDeleted(String dataPath) throws Exception {
    				System.out.println(Thread.currentThread().getName() + "-----监听到节点被删除，分布式锁被释放");
    				cdl.countDown();
    			}
    
    			@Override
    			public void handleDataChange(String dataPath, Object data) throws Exception {
    				
    			}
    		};
    
    		client.subscribeDataChanges(this.beforePath.get(), listener);
    
    		// 怎么让自己阻塞
    		if (this.client.exists(this.beforePath.get())) {
    			try {
    				System.out.println(Thread.currentThread().getName() + "-----分布式锁没抢到，进入阻塞状态");
    				cdl.await();
    				System.out.println(Thread.currentThread().getName() + "-----释放分布式锁，被唤醒");
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		// 醒来后，取消watcher
    		client.unsubscribeDataChanges(this.beforePath.get(), listener);
    	}
    
    	@Override
    	public void unlock() {
    		System.out.println(Thread.currentThread().getName() + "-----释放分布式锁");
    		if(reenterCount.get() > 1) {
    			// 重入次数减1，释放锁
    			reenterCount.set(reenterCount.get() - 1);
    			return;
    		}
    		// 删除节点
    		if(this.currentPath.get() != null) {
    			this.client.delete(this.currentPath.get());
    			this.currentPath.set(null);
    			this.reenterCount.set(0);
    		}
    	}
    
    	@Override
    	public void lockInterruptibly() throws InterruptedException {
    		
    	}
    
    	@Override
    	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    		
    		return false;
    	}
    
    	@Override
    	public Condition newCondition() {
    		return null;
    	}
    }

### 测试入口 ###

package test.lock;
    
    import com.alibaba.fastjson.annotation.JSONType;
    import lombok.Data;
    
    import java.util.HashSet;
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 分布式锁测试类入口
     * @author nachuan
     * @create 2019-10-13 09:35
     */
    public class ZkFairLockTest {
    
    
    
    
        public static void main(String[] args) {
    
            // 服务集群数
            int service = 5;
            // 并发数
            int requestSize = 10;
    
            CyclicBarrier requestBarrier = new CyclicBarrier(requestSize * service);
            // 多线程模拟高并发
            for (int i = 0; i < service; i++) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        // 模拟分布式集群的场景
                        OrderServiceImplWithDistributeLock t5 = new OrderServiceImplWithDistributeLock();
                        System.out.println(Thread.currentThread().getName() + "---------我准备好---------------");
    
                        for(int i =0; i < requestSize; i++) {
                            new Thread(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        // 等待service台服务，requestSize个请求 一起出发
                                        requestBarrier.await();
                                    } catch (InterruptedException | BrokenBarrierException e) {
                                        e.printStackTrace();
                                    }
                                    t5.createOrder();
                                }
    
                            }).start();
                        }
                    }
                }).start();
    
            }
    
           /* try {
                Thread.currentThread().join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/
    
    
        }
    
    
    
    }