本篇内容主要讲解“ZooKeeper同步框架怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“ZooKeeper同步框架怎么实现”吧!
首先,定义一个同步接口,它有一个execute方法,主要负责同步任务的实现。
Path参数是任务节点(用户),只有相同的节点才会同步工作。想象一下,去银行取钱,如果每个人都有一个专属的柜台,那效率是明显的。
SynchronousProcessor参数用来处理具体的业务。
Synchronous.java
package org.bigmouth.nvwa.zookeeper.concurrent; /** * 同步,支持分布式 * * @author Allen Hu * 2015-4-17 */ public interface Synchronous { /** * 同步执行,根据path标识来区分同步工作。不同的path将不会同步进行。 * * @param处理结果类型 * @param path 任务节点 * e.g. "/project/synchronous/0000001" * @param processor 业务处理器 * @return 处理结果 */T execute(String path, SynchronousProcessorprocessor); }
MutexLockSynchronous.java
Synchronous的实现类,基于普通排它锁的方式实现。
package org.bigmouth.nvwa.zookeeper.concurrent; import org.apache.curator.framework.recipes.locks.InterProcessLock; import org.apache.curator.framework.recipes.locks.InterProcessMutex; import org.apache.zookeeper.common.PathUtils; import org.bigmouth.nvwa.zookeeper.ZkClientHolder; /** * 基于普通排他锁的方式实现同步 * * @author Allen Hu * 2015-4-17 */ public class MutexLockSynchronous implements Synchronous { private final ZkClientHolder zkClientHolder; public MutexLockSynchronous(ZkClientHolder zkClientHolder) { this.zkClientHolder = zkClientHolder; } @Override publicT execute(String path, SynchronousProcessorprocessor) { PathUtils.validatePath(path); InterProcessLock lock = new InterProcessMutex(zkClientHolder.get(), path); try { lock.acquire(); if (null != processor) return processor.process(); } catch (Exception e) { if (null != processor) processor.exceptionCaught(e); } finally { try { lock.release(); } catch (Exception e) { } } return null; } }
SynchronousProcessor.java
任务处理器接口,实现它来完成具体的业务工作
package org.bigmouth.nvwa.zookeeper.concurrent; /** * 同步业务处理器 * * @author Allen Hu * 2015-4-17 */ public interface SynchronousProcessor{ /** * 处理具体的业务 * * @return */ T process(); /** * 异常捕获 * * @param throwable */ void exceptionCaught(Throwable throwable); }
ZkClientHolder.java
当然少不了这个了,继承的父类可以不需要了解,就是定义了两个抽象方法:doInit和doDestroy方法。
package org.bigmouth.nvwa.zookeeper; import org.apache.commons.lang.StringUtils; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.ExponentialBackoffRetry; import org.bigmouth.nvwa.utils.BaseLifeCycleSupport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.base.Preconditions; /** * ZooKeeper client holder * * @author Allen Hu * 2015-4-16 */ public class ZkClientHolder extends BaseLifeCycleSupport { private static final Logger LOGGER = LoggerFactory.getLogger(ZkClientHolder.class); public static final int MAX_RETRIES = 3; public static final int BASE_SLEEP_TIMEMS = 3000; private CuratorFramework zkClient; private final String connectString; private final int sessionTimeout; public ZkClientHolder(String connectString, int sessionTimeout) { Preconditions.checkArgument(StringUtils.isNotBlank(connectString), "connectString cannot be blank"); Preconditions.checkArgument(sessionTimeout >= 10000, "sessionTimeout must be greater than 10000"); this.connectString = connectString; this.sessionTimeout = sessionTimeout; } public CuratorFramework get() { return zkClient; } @Override protected void doInit() { zkClient = CuratorFrameworkFactory.builder() .sessionTimeoutMs(sessionTimeout) .connectString(connectString) .retryPolicy(new ExponentialBackoffRetry(BASE_SLEEP_TIMEMS, MAX_RETRIES)) .build(); zkClient.start(); if (LOGGER.isInfoEnabled()) { LOGGER.info("Connected to ZooKepper server: {}", connectString); } } @Override protected void doDestroy() { if (null != zkClient) zkClient.close(); } }
最后来个测试类,模拟多个用户多线程处理任务的过程,我们达到了相同用户间同步的目的。
package org.bigmouth.nvwa.zookeeper.concurrent; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.apache.curator.utils.ZKPaths; import org.bigmouth.nvwa.zookeeper.ZkClientHolder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * * @author Allen Hu * 2015-4-17 */ public class ConcurrentTest { private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrentTest.class); private ZkClientHolder zkClientHolder = new ZkClientHolder("172.16.3.24:2181", 60000); private Synchronous synchronous; public ConcurrentTest() { zkClientHolder.init(); synchronous = new MutexLockSynchronous(zkClientHolder); } public class Service implements Runnable { private final String id; private final long sleepInMillis; public Service(String id, long sleepInMillis) { this.id = id; this.sleepInMillis = sleepInMillis; } @Override public void run() { synchronous.execute(ZKPaths.makePath("/nvwa/zookeeper/concurrent", id), new SynchronousProcessor() { @Override public String process() { LOGGER.info(id + " star...!"); try { Thread.sleep(sleepInMillis); } catch (InterruptedException e) { e.printStackTrace(); } LOGGER.info(id + " has execution!"); return id; } @Override public void exceptionCaught(Throwable throwable) { throwable.printStackTrace(); } }); } } static ExecutorService executor = Executors.newCachedThreadPool(); public static void main(String[] args) { ConcurrentTest ct = new ConcurrentTest(); executor.submit(ct.new Service("1", 5000)); // 1号 处理5秒 executor.submit(ct.new Service("1", 2000)); // 1号 处理2秒 executor.submit(ct.new Service("2", 5000)); // 2号 处理5秒 executor.submit(ct.new Service("3", 10000)); // 3号 处理10秒 executor.submit(ct.new Service("3", 500)); // 3号 处理0.5秒 } }
输出结果,1、2、3任务并行,而相同的任务串行。如:第二个1号等第一个1号执行完才开始。
到此,相信大家对“ZooKeeper同步框架怎么实现”有了更深的了解,不妨来实际操作一番吧!这里是亿速云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。