怎么分析ZooKeeper分布式任务调度中心

2023-06-04 13:06:05 678人浏览八月长安

摘要

怎么分析ZooKeeper分布式任务调度中心，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。一：背景软件系统中，定时任务往往不可或缺，大家可能会采用spring qu

怎么分析ZooKeeper分布式任务调度中心，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。

一：背景

软件系统中，定时任务往往不可或缺，大家可能会采用spring quartz 建立定时任务模块。当任务模块进行了分布式部署，通常会出现定时任务重复执行的情况。怎么避免这种情况呢，是否可以构建一个任务注册中心，Quartz负责注册任务，但不具体执行任务内的业务逻辑。

任务注册中心解决任务重复注册的问题，同时将任务分配给若干处理器进行具体的业务处理，保证在同一个时间内，一个任务只会被一个处理器进行处理。如下图

怎么分析ZooKeeper分布式任务调度中心

二：任务注册中心与调度中心

采用ZooKeeper的客户端框架Curator, 利用ZooKeeper的Master选举机制实现。注册任务就相当于在ZooKeeper中创建或更新一个节点。通过更新节点的内容，来记录任务的执行状态。

三：代码模拟

public class ZookeeperDemo5 {
private static String PATH = "/zkbk2";
private static String FINISH_FLAG="finish";
private static CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("10.243.3.18:2181,10.243.3.17:2181,10.243.3.27:2181").sessionTimeoutMs(5000)
.retryPolicy(new ExponentialBackoffRetry(100, 3)).build();
public static void main(String[] args) throws Exception {
client.start();
PathChildrenCache cache = new PathChildrenCache(client, PATH, true);
cache.start(StartMode.POST_INITIALIZED_EVENT);
cache.getListenable().addListener(new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
switch (event.getType()) {
case CHILD_ADDED:
System.out.println("CHILD_ADDED," + event.getData().getPath());
deal(event);
break;
case CHILD_UPDATED:
System.out.println("CHILD_UPDATED," + event.getData().getPath());
deal(event);
break;
case CHILD_REMOVED:
System.out.println("CHILD_REMOVED," + event.getData().getPath());
break;
default:
break;
}
}
});
Thread.sleep(Integer.MAX_VALUE);
}
private static void deal( PathChildrenCacheEvent event)
{
final LeaderLatch latch = new LeaderLatch(client, event.getData().getPath());
latch.addListener(new LeaderLatchListener() {
@Override
public void notLeader() {
}
@Override
public void isLeader() {
try {
System.out.println("======begin deal==========" + event.getData().getPath());
String data = new String(client.getData().forPath(event.getData().getPath()));
if (!data.contains("finish")) {
Thread.sleep(10000L);
System.out.println("");
System.out.println("");
System.out.println("**************************************");
System.out.println("**************************************");
System.out.println("**************************************");
System.out.println("**************************************");
System.out.println("");
System.out.println("");
client.setData().forPath(event.getData().getPath(), FINISH_FLAG.getBytes());
}
System.out.println("======finish deal==========" + event.getData().getPath());
latch.close();
}
catch (Exception e) {
e.printStackTrace();
}
}
});
try {
latch.start();
}
catch (Exception e) {
e.printStackTrace();
}
}
}