本篇内容主要讲解“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同步框架怎么实现”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
