在线精品99_中国九九盗摄偷拍偷看_91免费版在线观看_91.app_91高清视频在线_99热最新网站

ZooKeeper同步框架怎么实现

118次阅读
没有评论

共计 5479 个字符,预计需要花费 14 分钟才能阅读完成。

本篇内容主要讲解“ZooKeeper 同步框架怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让丸趣 TV 小编来带大家学习“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 同步框架怎么实现”有了更深的了解,不妨来实际操作一番吧!这里是丸趣 TV 网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!

正文完
 
丸趣
版权声明:本站原创文章,由 丸趣 2023-08-16发表,共计5479字。
转载说明:除特殊说明外本站除技术相关以外文章皆由网络搜集发布,转载请注明出处。
评论(没有评论)
主站蜘蛛池模板: 91精品综合 | 202z日本高清免费视频 | 成年人黄色毛片 | 日本一级毛片免费播 | 无码人妻丰满熟妇啪啪网站 | 亚欧免费无码aⅴ在线观看 亚欧日韩 | 亚洲国产精品一区 | 国产极品美女高潮无套在线观看 | 国产精品无卡无在线播放 | 色悠久久久久综合网小说 | 日本真人边吃奶边做爽电影 | 国产精品成人第一区 | 国产一区二区三区视频 | 国产亚洲精品久久久久久 | 久久精品视频免费播放 | 看久久| 欧美一区中文字幕 | a级国产乱理伦片在线播放 a级国产乱理伦片在线观看 | 天堂а√在线最新版中文在线 | 在线视频激情 | 欧美日韩三级 | 亚洲国产精品无码久久一区二区 | 国产精品久久成人影院 | 亚洲国产精品特色大片观看完整版 | 久久久久久久久女黄 | 久久成人国产精品 | 国产亚洲一本大道中文在线 | 久久国产精品久久久久久小说 | 欧美换爱交换乱理伦片不卡片 | 亚洲成在人色婷婷 | 无码办公室丝袜ol中文字幕 | 四虎精品成人免费影视 | 亚洲熟妇无码久久精品 | 国产精品永久在线观看 | 午夜 在线| se94se最新网站 | 成年美女在线观看免费视频 | 黄色一级性生活视频 | 国产黄色免费网站 | 正在播放重口老熟女露脸 | 国产成 人 综合 亚洲网 |