惠州网站制作费用,网站建设是哪个专业,工作作风方面对照检查材料,舞蹈培训机构网站模板上一篇#xff1a;02-Zookeeper实战
1. 项目构建
zookeeper 官方的客户端没有和服务端代码分离#xff0c;他们为同一个jar 文件#xff0c;所以我们直接引入zookeeper的maven即可#xff0c; 这里版本请保持与服务端版本一致#xff0c;不然会有很多兼容性的问题
02-Zookeeper实战
1. 项目构建
zookeeper 官方的客户端没有和服务端代码分离他们为同一个jar 文件所以我们直接引入zookeeper的maven即可 这里版本请保持与服务端版本一致不然会有很多兼容性的问题
dependencygroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactIdversion3.5.8/version
/dependency2. 创建客户端实例
为了便于测试直接在初始化方法中创建zookeeper实例
Slf4j
public class ZookeeperClientTest {private static final String ZK_ADDRESS192.168.109.200:2181;private static final int SESSION_TIMEOUT 5000;private static ZooKeeper zooKeeper;private static final String ZK_NODE/zk-node;Beforepublic void init() throws IOException, InterruptedException {final CountDownLatch countDownLatchnew CountDownLatch(1);zooKeepernew ZooKeeper(ZK_ADDRESS, SESSION_TIMEOUT, event - {if (event.getState() Watcher.Event.KeeperState.SyncConnected event.getType() Watcher.Event.EventType.None){countDownLatch.countDown();log.info(连接成功);}});log.info(连接中....);countDownLatch.await();}
}创建Zookeeper实例的方法
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[])
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[], boolean, HostProvider)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[], boolean, HostProvider, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[], boolean)参数名称含义connectStringZooKeeper服务器列表由英文逗号分开的host:port字符串组成每一个都代表一台ZooKeeper机器如host1:port1,host2:port2,host3:port3。另外也可以在connectString中设置客户端连接上ZooKeeper后的根目录方法是在host:port字符串之后添加上这个根目录例如,host1:port1,host2:port2,host3:port3/zk-base,这样就指定了该客户端连接上ZooKeeper服务器之后,所有对ZooKeeper的操作都会基于这个根目录。例如客户端对/sub-node 的操作最终创建 /zk-node/sub-node, 这个目录也叫Chroot即客户端隔离命名空间。sessionTimeout会话的超时时间是一个以“毫秒”为单位的整型值。在ZooKeeper中有会话的概念在一个会话周期内ZooKeeper客户端和服务器之间会通过心跳检测机制来维持会话的有效性一旦在sessionTimeout时间内没有进行有效的心跳检测会话就会失效。watcherZooKeeper允许客户端在构造方法中传入一个接口 watcher (org.apache. zookeeper.Watcher)的实现类对象来作为默认的 Watcher事件通知处理器。当然该参数可以设置为null 以表明不需要设置默认的 Watcher处理器。canBeReadOnly这是一个boolean类型的参数用于标识当前会话是否支持“read-only(只读)”模式。默认情况下在ZooKeeper集群中一个机器如果和集群中过半及以上机器失去了网络连接那么这个机器将不再处理客户端请求包括读写请求)。但是在某些使用场景下当ZooKeeper服务器发生此类故障的时候我们还是希望ZooKeeper服务器能够提供读服务当然写服务肯定无法提供——这就是 ZooKeeper的“read-only”模式。sessionId和 sessionPasswd分别代表会话ID和会话秘钥。这两个参数能够唯一确定一个会话同时客户端使用这两个参数可以实现客户端会话复用从而达到恢复会话的效果。具体使用方法是第一次连接上ZooKeeper服务器时通过调用ZooKeeper对象实例的以下两个接口即可获得当前会话的ID和秘钥:long getSessionId();byte[]getSessionPasswd( );荻取到这两个参数值之后就可以在下次创建ZooKeeper对象实例的时候传入构造方法了
同步创建节点
Test
public void createTest() throws KeeperException, InterruptedException {String path zooKeeper.create(ZK_NODE, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);log.info(created path: {},path);
}异步创建节点
Test
public void createAsycTest() throws InterruptedException {zooKeeper.create(ZK_NODE, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT,(rc, path, ctx, name) - log.info(rc {},path {},ctx {},name {},rc,path,ctx,name),context);TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
}修改节点数据
Test
public void setTest() throws KeeperException, InterruptedException {Stat stat new Stat();byte[] data zooKeeper.getData(ZK_NODE, false, stat);log.info(修改前: {},new String(data));zooKeeper.setData(ZK_NODE, changed!.getBytes(), stat.getVersion());byte[] dataAfter zooKeeper.getData(ZK_NODE, false, stat);log.info(修改后: {},new String(dataAfter));
}3. 什么是 Curator
Curator 是一套由netflix 公司开源的Java 语言编程的 ZooKeeper 客户端框架Curator项目是现在ZooKeeper 客户端中使用最多对ZooKeeper 版本支持最好的第三方客户端并推荐使用Curator 把我们平时常用的很多 ZooKeeper 服务开发功能做了封装例如 Leader 选举、分布式计数器、分布式锁。这就减少了技术人员在使用 ZooKeeper 时的大部分底层细节开发工作。在会话重新连接、Watch 反复注册、多种异常处理等使用场景中用原生的 ZooKeeper 处理比较复杂。而在使用 Curator 时由于其对这些功能都做了高度的封装使用起来更加简单不但减少了开发时间而且增强了程序的可靠性。
4. Curator 实战
这里我们以 Maven 工程为例首先要引入Curator 框架相关的开发包这里为了方便测试引入了junit lombok由于Zookeeper本身以来了 log4j 日志框架所以这里可以创建对应的log4j配置文件后直接使用。 如下面的代码所示我们通过将 Curator 相关的引用包配置到 Maven 工程的 pom 文件中将 Curaotr 框架引用到工程项目里在配置文件中分别引用了两个 Curator 相关的包第一个是 curator-framework 包该包是对 ZooKeeper 底层 API 的一些封装。另一个是 curator-recipes 包该包封装了一些 ZooKeeper 服务的高级特性如Cache 事件监听、选举、分布式锁、分布式Barrier。
dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversion5.0.0/versionexclusionsexclusiongroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactId/exclusion/exclusions
/dependency
dependencygroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactIdversion3.5.8/version
/dependency
dependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.13/version
/dependency
dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.12/version
/dependency
dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversion5.0.0/versionexclusionsexclusiongroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactId/exclusion/exclusions
/dependency
dependencygroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactIdversion3.5.8/version
/dependency
dependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.13/version
/dependency
dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId会话创建
要进行客户端服务器交互第一步就要创建会话 Curator 提供了多种方式创建会话比如用静态工厂方式创建
// 重试策略
RetryPolicy retryPolicy new ExponentialBackoffRetry(1000, 3)
CuratorFramework client CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
client.start();或者使用 fluent 风格创建
RetryPolicy retryPolicy new ExponentialBackoffRetry(1000, 3);
CuratorFramework client CuratorFrameworkFactory.builder().connectString(192.168.128.129:2181).sessionTimeoutMs(5000) // 会话超时时间.connectionTimeoutMs(5000) // 连接超时时间.retryPolicy(retryPolicy).namespace(base) // 包含隔离名称.build();
client.start();这段代码的编码风格采用了流式方式最核心的类是 CuratorFramework 类该类的作用是定义一个 ZooKeeper 客户端对象并在之后的上下文中使用。在定义 CuratorFramework 对象实例的时候我们使用了 CuratorFrameworkFactory 工厂方法并指定了 connectionString 服务器地址列表、retryPolicy 重试策略 、sessionTimeoutMs 会话超时时间、connectionTimeoutMs 会话创建超时时间。下面我们分别对这几个参数进行讲解
connectionString服务器地址列表在指定服务器地址列表的时候可以是一个地址也可以是多个地址。如果是多个地址那么每个服务器地址列表用逗号分隔, 如 host1:port1,host2:port2,host3port3 。
retryPolicy重试策略当客户端异常退出或者与服务端失去连接的时候可以通过设置客户端重新连接 ZooKeeper 服务端。而 Curator 提供了 一次重试、多次重试等不同种类的实现方式。在 Curator 内部可以通过判断服务器返回的 keeperException 的状态代码来判断是否进行重试处理如果返回的是 OK 表示一切操作都没有问题而 SYSTEMERROR 表示系统或服务端错误。
策略名称描述ExponentialBackoffRetry重试一组次数重试之间的睡眠时间增加RetryNTimes重试最大次数RetryOneTime只重试一次RetryUntilElapsed在给定的时间结束之前重试
超时时间Curator 客户端创建过程中有两个超时时间的设置。一个是 sessionTimeoutMs 会话超时时间用来设置该条会话在 ZooKeeper 服务端的失效时间。另一个是 connectionTimeoutMs 客户端创建会话的超时时间用来限制客户端发起一个会话连接到接收 ZooKeeper 服务端应答的时间。sessionTimeoutMs 作用在服务端而 connectionTimeoutMs 作用在客户端。
创建节点
创建节点的方式如下面的代码所示回顾我们之前课程中讲到的内容描述一个节点要包括节点的类型即临时节点还是持久节点、节点的数据信息、节点是否是有序节点等属性和性质。 Test
public void testCreate() throws Exception {String path curatorFramework.create().forPath(/curator-node);// curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath(/curator-node,some-data.getBytes())log.info(curator create node :{} successfully.,path);
}在 Curator 中可以使用 create 函数创建数据节点并通过 withMode 函数指定节点类型持久化节点临时节点顺序节点临时顺序节点持久化顺序节点等默认是持久化节点之后调用 forPath 函数来指定节点的路径和数据信息。
一次性创建带层级结构的节点
Test
public void testCreateWithParent() throws Exception {String pathWithParent/node-parent/sub-node-1;String path curatorFramework.create().creatingParentsIfNeeded().forPath(pathWithParent);log.info(curator create node :{} successfully.,path);
}获取数据
Test
public void testGetData() throws Exception {byte[] bytes curatorFramework.getData().forPath(/curator-node);log.info(get data from node :{} successfully.,new String(bytes));
}更新节点
我们通过客户端实例的 setData() 方法更新 ZooKeeper 服务上的数据节点在setData 方法的后边通过 forPath 函数来指定更新的数据节点路径以及要更新的数据。
Test
public void testSetData() throws Exception {curatorFramework.setData().forPath(/curator-node,changed!.getBytes());byte[] bytes curatorFramework.getData().forPath(/curator-node);log.info(get data from node /curator-node :{} successfully.,new String(bytes));
}删除节点
Test
public void testDelete() throws Exception {String pathWithParent/node-parent;curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(pathWithParent);
}guaranteed该函数的功能如字面意思一样主要起到一个保障删除成功的作用其底层工作方式是只要该客户端的会话有效就会在后台持续发起删除请求直到该数据节点在 ZooKeeper 服务端被删除。
deletingChildrenIfNeeded指定了该函数后系统在删除该数据节点的时候会以递归的方式直接删除其子节点以及子节点的子节点。
异步接口
Curator 引入了BackgroundCallback 接口用来处理服务器端返回来的信息这个处理过程是在异步线程中调用默认在 EventThread 中调用也可以自定义线程池。
public interface BackgroundCallback
{/*** Called when the async background operation completes** param client the client* param event operation result details* throws Exception errors*/public void processResult(CuratorFramework client, CuratorEvent event) throws Exception;
}如上接口主要参数为 client 客户端 和 服务端事件 event inBackground 异步处理默认在EventThread中执行
Test
public void test() throws Exception {curatorFramework.getData().inBackground((item1, item2) - {log.info( background: {}, item2);}).forPath(ZK_NODE);TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
}指定线程池
Test
public void test() throws Exception {ExecutorService executorService Executors.newSingleThreadExecutor();curatorFramework.getData().inBackground((item1, item2) - {log.info( background: {}, item2);},executorService).forPath(ZK_NODE);TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
}5. Curator 监听器
/*** Receives notifications about errors and background events*/
public interface CuratorListener
{/*** Called when a background task has completed or a watch has triggered** param client client* param event the event* throws Exception any errors*/public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception;
}针对 background 通知和错误通知。使用此监听器之后调用inBackground 方法会异步获得监听
Curator Caches: Curator 引入了 Cache 来实现对 Zookeeper 服务端事件监听Cache 事件监听可以理解为一个本地缓存视图与远程 Zookeeper 视图的对比过程。Cache 提供了反复注册的功能。Cache 分为两类注册类型节点监听和子节点监听。
node cache: NodeCache 对某一个节点进行监听
public NodeCache(CuratorFramework client,String path)
Parameters:
client - the client
path - path to cache可以通过注册监听器来实现对当前节点数据变化的处理
public void addListener(NodeCacheListener listener)Add a change listener
Parameters:
listener - the listenerSlf4j
public class NodeCacheTest extends AbstractCuratorTest{public static final String NODE_CACHE/node-cache;Testpublic void testNodeCacheTest() throws Exception {createIfNeed(NODE_CACHE);NodeCache nodeCache new NodeCache(curatorFramework, NODE_CACHE);nodeCache.getListenable().addListener(new NodeCacheListener() {Overridepublic void nodeChanged() throws Exception {log.info({} path nodeChanged: ,NODE_CACHE);printNodeData();}});nodeCache.start();}public void printNodeData() throws Exception {byte[] bytes curatorFramework.getData().forPath(NODE_CACHE);log.info(data: {},new String(bytes));}
}path cache: PathChildrenCache 会对子节点进行监听但是不会对二级子节点进行监听
public PathChildrenCache(CuratorFramework client,String path,boolean cacheData)
Parameters:
client - the client
path - path to watch
cacheData - if true, node contents are cached in addition to the stat可以通过注册监听器来实现对当前节点的子节点数据变化的处理
public void addListener(PathChildrenCacheListener listener)Add a change listener
Parameters:
listener - the listenerSlf4j
public class PathCacheTest extends AbstractCuratorTest{public static final String PATH/path-cache;Testpublic void testPathCache() throws Exception {createIfNeed(PATH);PathChildrenCache pathChildrenCache new PathChildrenCache(curatorFramework, PATH, true);pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {log.info(event: {},event);}});// 如果设置为true则在首次启动时就会缓存节点内容到Cache中pathChildrenCache.start(true);}
}tree cache: TreeCache 使用一个内部类TreeNode来维护这个一个树结构。并将这个树结构与ZK节点进行了映射。所以TreeCache 可以监听当前节点下所有节点的事件。
public TreeCache(CuratorFramework client,String path,boolean cacheData)
Parameters:
client - the client
path - path to watch
cacheData - if true, node contents are cached in addition to the stat可以通过注册监听器来实现对当前节点的子节点及递归子节点数据变化的处理
public void addListener(TreeCacheListener listener)Add a change listener
Parameters:
listener - the listenerSlf4j
public class TreeCacheTest extends AbstractCuratorTest{public static final String TREE_CACHE/tree-path;Testpublic void testTreeCache() throws Exception {createIfNeed(TREE_CACHE);TreeCache treeCache new TreeCache(curatorFramework, TREE_CACHE);treeCache.getListenable().addListener(new TreeCacheListener() {Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {log.info( tree cache: {},event);}});treeCache.start();}
}下一篇04-Zookeeper集群详解
