package com..zookeeper.GWdemo; 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 InterprocessLock { public static void main(String[] args) { CuratorFramework zkClient = getZkClient(); String lockPath = "/lock"; InterProcessMutex lock = new InterProcessMutex(zkClient, lockPath); //模拟50个线程抢锁 for (int i = 0; i < 50; i++) { new Thread(new TestThread(i, lock)).start(); } } static class TestThread implements Runnable { private Integer threadFlag; private InterProcessMutex lock; public TestThread(Integer threadFlag, InterProcessMutex lock) { this.threadFlag = threadFlag; this.lock = lock; } @Override public void run() { try { lock.acquire(); System.out.println("第"+threadFlag+"线程获取到了锁"); //等到1秒后释放锁 Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); }finally { try { lock.release(); } catch (Exception e) { e.printStackTrace(); } } } } private static CuratorFramework getZkClient() { String zkServerAddress = "127.0.0.1:2181"; ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000); CuratorFramework zkClient = CuratorFrameworkFactory.builder() .connectString(zkServerAddress) .sessionTimeoutMs(5000) .connectionTimeoutMs(5000) .retryPolicy(retryPolicy) .build(); zkClient.start(); return zkClient; } }
起20个线程,对数据库做-1操作,数据库值为15
package com..zookeeper.zkLock; 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 wfc.service.database.RecordSet; import wfc.service.database.SQL; import java.util.concurrent.Executor; import java.util.concurrent.Executors; /** * 简单理解 */ public class CuratorLockDemo { public static void main (String[] args) { String servers = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183"; CuratorFramework curator = CuratorFrameworkFactory.builder() .retryPolicy(new ExponentialBackoffRetry(10000, 3)) .connectString(servers).build(); curator.start(); final InterProcessMutex lock = new InterProcessMutex(curator, "/global_lock"); Executor pool = Executors.newFixedThreadPool(10); for (int i = 0; i < 20; i ++) { pool.execute(new Runnable() { public void run() { try { lock.acquire(); int count = 1; String st_fj_id = "keys"; System.out.println(Thread.currentThread().getId()+" ==="); String insertSql = "update dangan_fj set count = count-? where st_fj_id = ? and count>=1"; Object[] insertObject = new Object[] {count,st_fj_id}; RecordSet rs = SQL.execute(insertSql,insertObject); int number = rs.TOTAL_RECORD_COUNT; //影响行数 System.out.println(Thread.currentThread().getId()+" 数据库影响行数 "+number ); } catch (Exception e) { e.printStackTrace(); }finally{ try { lock.release(); } catch (Exception e) { e.printStackTrace(); } } } }); } } }
运行结果 :