Curator 分布式锁的入门示例代码

待我称王封你为后i 2022-02-02 14:27 185阅读 0赞

### 文章目录 ###

*   *  1、单机版的并发同步的示例
     *  2、分布式锁：
     *   *  2.1、分布式锁的核心接口：
         *  2.2、分布式锁的示例代码：

## 1、单机版的并发同步的示例 ##

package com.aop8.curator.lock;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class Lock1 { 
    
    	static ReentrantLock reentrantLock = new ReentrantLock();
    	static int count = 10;
    	public static void genarNo(){ 
    		try { 
    			reentrantLock.lock();
    			TimeUnit.SECONDS.sleep(1);
    			System.out.println(count);
    		} catch (InterruptedException e) { 
    			e.printStackTrace();
    		} finally { 
    			reentrantLock.unlock();
    		}
    	}
    	
    	public static void main(String[] args) throws Exception{ 
    		
    		final CountDownLatch countdown = new CountDownLatch(1);
    		for(int i = 0; i < 10; i++){ 
    			new Thread(new Runnable() { 
    				@Override
    				public void run() { 
    					try { 
    						countdown.await();
    						genarNo();
    					} catch (Exception e) { 
    						e.printStackTrace();
    					} finally { 
    					}
    				}
    			},"t" + i).start();
    		}
    		Thread.sleep(50);
    		countdown.countDown();
    		
    	}
    }

运行结果：

9
    8
    7
    6
    5
    4
    3
    2
    1
    0

## 2、分布式锁： ##

### 2.1、分布式锁的核心接口： ###

public void acquire() throws Exception;
    
    public void release() throws Exception;

### 2.2、分布式锁的示例代码： ###

package com.aop8.curator.lock;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    import org.apache.curator.RetryPolicy;
    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;
    
    import com.aop8.CommonsUtils;
    
    
    public class Lock2 { 
    
    	/** session超时时间 */
    	static final int SESSION_OUTTIME = 5000;//ms 
    	
    	static int count = 10;
    	public static void genarNo(){ 
    		try { 
    			count--;
    			System.out.println(count);
    		} finally { 
    		
    		}
    	}
    	
    	public static void main(String[] args) throws Exception { 
    		
    		//1 重试策略：初次间隔时间为1s 重试10次
    		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
    		//2 通过工厂创建连接
    		CuratorFramework cf = CuratorFrameworkFactory.builder()
    					.connectString("127.0.0.1:2181")
    					.sessionTimeoutMs(SESSION_OUTTIME)
    					.retryPolicy(retryPolicy)
    // .namespace("super")
    					.build();
    		//3 开启连接
    		cf.start();
    		
    		//4 分布式锁
    		final InterProcessMutex lock = new InterProcessMutex(cf, "/super");
    		final CountDownLatch countdown = new CountDownLatch(1);
    		
    		for(int i = 0; i < 10; i++){ 
    			new Thread(new Runnable() { 
    				@Override
    				public void run() { 
    					try { 
    						countdown.await();
    						//加锁
    						lock.acquire();
    						System.out.println(Thread.currentThread().getName()+" 开始执行业务逻辑...");
    						
    						TimeUnit.SECONDS.sleep(2);//业务处理中...
    						
    					} catch (Exception e) { 
    						e.printStackTrace();
    					} finally { 
    						try { 
    							
    							System.out.println(Thread.currentThread().getName()+" 结束\n");
    							//释放
    							lock.release();
    						} catch (Exception e) { 
    							e.printStackTrace();
    						}
    					}
    				}
    			},"t" + i).start();
    		}
    		Thread.sleep(100);
    		countdown.countDown();
    		 
    	}
    }

运行结果：（每次运行结果是不一样，但是可以保证每次只能有一个线程进行业务逻辑操作）

t8 开始执行业务逻辑...
    t8 结束
    
    t4 开始执行业务逻辑...
    t4 结束
    
    t6 开始执行业务逻辑...
    t6 结束
    
    t2 开始执行业务逻辑...
    t2 结束
    
    t1 开始执行业务逻辑...
    t1 结束
    
    t0 开始执行业务逻辑...
    t0 结束
    
    t7 开始执行业务逻辑...
    t7 结束
    
    t9 开始执行业务逻辑...
    t9 结束
    
    t3 开始执行业务逻辑...
    t3 结束
    
    t5 开始执行业务逻辑...
    t5 结束