标签:att for 分组 复合 scheme cat lis channel member
dubbo学习小结
参考:
https://blog.csdn.net/paul_wei2008/article/details/19355681
https://blog.csdn.net/liweisnake/article/details/63251252
https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/
https://www.jianshu.com/p/b9f3f6a16911
Dubbo基本原理机制
分布式服务框架:
–高性能和透明化的RPC远程服务调用方案
–SOA服务治理方案
-Apache MINA 框架基于Reactor模型通信框架,基于tcp长连接
Dubbo缺省协议采用单一长连接和NIO异步通讯,
适合于小数据量大并发的服务调用,以及服务消费者机器数远大于服务提供者机器数的情况
分析源代码,基本原理如下:
zookeeper入门
zookeeper可谓是目前使用最广泛的分布式组件了。其功能和职责单一,但却非常重要。
在现今这个年代,介绍zookeeper的书和文章可谓多如牛毛,本人不才,试图通过自己的理解来介绍zookeeper,希望通过一个初学者的视角来学习zookeeper,以期让人更加深入和平稳的理解zookeeper。其中参考了不少教程和书,相关书目列在文末,也感谢这些作者。
学习新的框架,先让我们搞清楚他是什么,这是它的内涵,然后再介绍它能做什么,这是它的外延,内涵和外延共同来定义框架本身,会对框架有较为深刻的理解,在应用层面上知道如何用。其次再搞清楚zookeeper相关的理论基础,其目的是知道zookeeper是如何被发明的,是否能够借鉴以便今后自己能够用到其他地方。最后搞清楚zookeeper中一些设计的原理和细节,目的也是搞清来龙去脉,学会“术”从而应用到别的地方。当然了,加深的理解同样能够帮助认识zookeeper本身,在使用时才知道为什么这样用。
首先,
zookeeper到底是什么?
zookeeper实际上是yahoo开发的,用于分布式中一致性处理的框架。最初其作为研发hadoop时的副产品。由于分布式系统中一致性处理较为困难,其他的分布式系统没有必要 费劲重复造轮子,故随后的分布式系统中大量应用了zookeeper,以至于zookeeper成为了各种分布式系统的基础组件,其地位之重要,可想而知。著名的hadoop,kafka,dubbo 都是基于zookeeper而构建。
要想理解zookeeper到底是做啥的,那首先得理解清楚,什么是一致性。
所谓的一致性,实际上就是围绕着“看见”来的。谁能看见?能否看见?什么时候看见?举个例子:淘宝后台卖家,在后台上架一件大促的商品,通过服务器A提交到主数据库,假设刚提交后立马就有用户去通过应用服务器B去从数据库查询该商品,就会出现一个现象,卖家已经更新成功了,然而买家却看不到;而经过一段时间后,主数据库的数据同步到了从数据库,买家就能查到了。
假设卖家更新成功之后买家立马就能看到卖家的更新,则称为强一致性;
如果卖家更新成功后买家不能看到卖家更新的内容,则称为弱一致性;
而卖家更新成功后,买家经过一段时间最终能看到卖家的更新,则称为最终一致性。
更多的一致性例子可以参考文献2,里面列举了10种一致性的例子,如果要给一致性下个定义,可以是分布式系统中状态或数据保持同步和一致。特别需要注意一致性跟事务的区别,可以记得学习数据库时特别强调ACID,故而满足ACID的数据库能够做事务,其中C即是一致性,因此,事务是一致性的一种特例,比起一致性更难达成。
如何保证在分布式环境下数据的最终一致,这个就是zookeeper需要解决的问题。对于这些问题,有哪些挑战,zookeeper又是如何解决这些挑战的,下一篇文章将会主要涉及这个主题。
一些常见的解决一致性问题的方式:
1. 查询重试补偿。对于分布式应用中不确定的情况,先使用查询接口查询到当前状态,如果当前状态不一致则采用补偿接口对状态进行重试推进,或者回滚接口对业务做回滚。典型的场景如银行跟支付宝之间的交互。支付宝发送一个转账请求到银行,如一直未收到响应,则可以通过银行的查询接口查询该笔交易的状态,如该笔交易对方未收到,则采取补偿的模式进行推送。
2. 定时任务推送。对于上面的情况,有可能一次推送搞不定,于是需要2次,3次推送。不要怀疑,支付宝内最初掉单率很高,全靠后续不断的定时任务推送增加成功率。
3. TCC。try-confirm-cancel。实际上是两阶段协议,第二阶段的可以实现提交操作或是逆操作。
zookeeper到底能做什么?
在业界的实际应用是什么?了解这些应用,会对zookeeper能够做的事有更直观的认识。
hadoop:
鼻祖级应用,ResourceManager在整个hadoop中算是单点,为了实现其高可用,分为主备ResourceManager,zookeeper在其中管理整个ResourceManager。
可以想象,主备ResourceManager最初是主RM提供服务,如果一切安好,则zookeeper无用武之地。然而,总归会出现主RM提供不了服务的情况。于是会出现主备切换的情况,而zookeeper正是为主备切换保驾护航。
先来推理一下,主备切换会出现什么问题。传统的主备切换,可以让主备之间维持心跳连接,一旦备机发现主机心跳检测不到了,则自己切换为主机,原来的主机等待救援。这种方式有两个问题,一是由于网络抖动,负载过大等问题,备机检测不到心跳并不能说明主机一定挂了,有可能一定时间后主机或网络恢复,这时候主机并不知道备机已经切换为主机,2台主机互相争用,可能造成脑裂;二是如果一些数据集中在主机上面,则备机切换时由于同步延时势必会损失掉一部分的数据。
如何解决这些问题?早期的方式提供了不少解决方案,比如备机一旦切换为主机,则通过电源控制直接切断主机电源,简单粗暴,但是此刻备机已经是单点,如果主机是因为量撑不住而挂,那备机有可能会重蹈覆辙,最终导致整个服务不可用。
zookeeper又是如何解决这个问题的呢?
1. zookeeper作为第三方集群参与到主备节点中去,当主备启动时会在zookeeper上竞争创建一个临时锁节点,争用成功者则充当主机,其余备机
2. 所有备机会监听该临时锁节点,一旦主机与zookeeper间session失效,则临时节点被删除
3. 一旦临时节点被删除,备机开始重新申请创建临时锁节点,重新争用为主机;
4. 用zookeeper如何解决脑裂?实际上主机争用到节点后通过对根节点做一个ACL权限控制,则其他抢占的机器由于无法更新临时锁节点,只有放弃成为备机。
zookeeper使用了非常简单又现成的方式来解决的这个问题,比起其他方案方便不少,这也是为啥zookeeper流行的原因。说白了,就是把复杂操作封装化精简化
dubbo:
作为业界知名的分布式soa框架,dubbo的主要的服务注册发现功能便是由zookeeper来提供的。
对于一个服务框架,注册中心是其核心中的核心,虽然暂时挂掉并不会导致整个服务出问题,但是一旦挂掉,整体风险就很高。考虑一般情况,注册中心就是单台机器的时候,其实现很容易,所有机器起来都去注册服务给它,并且所有调用方都跟它保持长连接,一旦服务有变,即通过长连接来通知到调用方。但是当服务集群规模扩大时,这事情就不简单了,单机保持连接数有限,而且容易故障。
作为一个稳定的服务化框架,dubbo可以选择并推荐zookeeper作为注册中心。其底层将zookeeper常用的客户端zkclient和curator封装成为ZookeeperClient。
1. 当服务提供者服务启动时,向zookeeper注册一个节点
2. 服务消费者则订阅其父节点的变化,诸如启动停止都能够通过节点创建删除得知,异常情况比如被调用方掉线也可以通过临时节点session 断开自动删除得知
3. 服务消费方同时也会将自己订阅的服务以节点创建的方式放到zookeeper
4. 于是可以得到映射关系,诸如谁提供了服务,谁订阅了谁提供的服务,基于这层关系再做监控,就能轻易得知整个系统情况。
zookeeper的基本数据模型
一句话,类似linux文件系统的节点模型
其节点有如下有趣而又重要的特性:
1. 同一时刻多台机器创建同一个节点,只有一个会争抢成功。利用这个特性可以做分布式锁。
2. 临时节点的生命周期与会话一致,会话关闭则临时节点删除。这个特性经常用来做心跳,动态监控,负载等动作
3. 顺序节点保证节点名全局唯一。这个特性可以用来生成分布式环境下的全局自增长id
通过zookeeper提供的原语服务,可以对zookeeper能做的事情有个精确和直观的认识
zookeeper提供的原语服务
1. 创建节点。
2. 删除节点
3. 更新节点
4. 获取节点信息
5. 权限控制
6. 事件监听
实际上,就是对节点的增删查改加上权限控制与事件监听,但是通过对这些原语的组合以及不同场景的使用,可以实现很多用法。参考文献5
1. 数据发布订阅。即注册中心,见上面dubbo用法。主要通过对节点管理做到发布以及事件监听做到订阅
2. 负载均衡。见上面kafka用法
3. 命名服务。zookeeper的节点结构天然支持命名服务,即把信息集中存储,并以树状管理,方便统一查阅
4. 分布式协调通知。协调通知实际上与发布订阅类似,由于引入的第三方的zookeeper,实际上对很多种协调通知做了解耦,比如参考文献4中提到的消息推送,心跳检测等
5. 集群管理与master选举。通过上面的第二点特性,可以轻易得知集群机器存活状况,从而轻松管理集群;通过上面第一点特性,可以做出master争抢。
6. 分布式锁。实际上就是第一点特性的应用。
7. 分布式队列。实际上就是第三点特性的应用。
8. 分布式的并发等待。类似于多线程的join问题,主任务的执行依赖于其他子任务全部执行完毕,在单机多线程里可以用join,但是分布式环境下如何实现呢。利用zookeeper,可以创建一个主任务节点,旗下子任务一旦执行完毕,则在主任务节点下挂一个子任务节点,等节点数量足够,则认为主任务可以开始执行。
可以发现,所有的原语就是zookeeper的基础,而其他的用法总结无非是将原语放到不同场景下的归类罢了。
相信到这里你对zookeeper应该有个初步的了解和大致的印象了。
本系列文章分为:
zookeeper入门系列-理论基础-zab协议
zookeeper入门系列-理论基础-raft协议
zookeeper入门系列-设计细节
参考文献:
保证分布式系统数据一致性的6种方案 http://weibo.com/ttarticle/p/show?id=2309403965965003062676
解决分布式系统的一致性问题,我们需要了解哪些理论? http://mp.weixin.qq.com/s/hGnpHfn7a7yxjPBP78i4bg
分布式系统的事务处理 http://coolshell.cn/articles/10910.html
ZooKeeper典型应用场景一览 http://jm.taobao.org/2011/10/08/1232/
zookeeper中的基本概念 http://www.hollischuang.com/archives/1280
zookeeper入门使用 http://www.importnew.com/23025.html
本文介绍的 Zookeeper 是以 3.2.2 这个稳定版本为基础,最新的版本可以通过官网 http://hadoop.apache.org/zookeeper/来获取,Zookeeper 的安装非常简单,下面将从单机模式和集群模式两个方面介绍 Zookeeper 的安装和配置。
单机安装非常简单,只要获取到 Zookeeper 的压缩包并解压到某个目录如:/home/zookeeper-3.2.2 下,Zookeeper 的启动脚本在 bin 目录下,Linux 下的启动脚本是 zkServer.sh,在 3.2.2 这个版本 Zookeeper 没有提供 windows 下的启动脚本,所以要想在 windows 下启动 Zookeeper 要自己手工写一个,如清单 1 所示:
1
2
3
4
5
6
7
8
9
10
11
12
13
|
setlocal set ZOOCFGDIR=%~dp0%..\conf set ZOO_LOG_DIR=%~dp0%.. set ZOO_LOG4J_PROP=INFO,CONSOLE set CLASSPATH=%ZOOCFGDIR% set CLASSPATH=%~dp0..\*;%~dp0..\lib\*;%CLASSPATH% set CLASSPATH=%~dp0..\build\classes;%~dp0..\build\lib\*;%CLASSPATH% set ZOOCFG=%ZOOCFGDIR%\zoo.cfg set ZOOMAIN=org.apache.zookeeper.server.ZooKeeperServerMain java "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %* endlocal |
在你执行启动脚本之前,还有几个基本的配置项需要配置一下,Zookeeper 的配置文件在 conf 目录下,这个目录下有 zoo_sample.cfg 和 log4j.properties,你需要做的就是将 zoo_sample.cfg 改名为 zoo.cfg,因为 Zookeeper 在启动时会找这个文件作为默认配置文件。下面详细介绍一下,这个配置文件中各个配置项的意义。
1
2
3
|
tickTime=2000 dataDir=D:/devtools/zookeeper-3.2.2/build clientPort=2181 |
当这些配置项配置好后,你现在就可以启动 Zookeeper 了,启动后要检查 Zookeeper 是否已经在服务,可以通过 netstat – ano 命令查看是否有你配置的 clientPort 端口号在监听服务。
Zookeeper 不仅可以单机提供服务,同时也支持多机组成集群来提供服务。实际上 Zookeeper 还支持另外一种伪集群的方式,也就是可以在一台物理机上运行多个 Zookeeper 实例,下面将介绍集群模式的安装和配置。
Zookeeper 的集群模式的安装和配置也不是很复杂,所要做的就是增加几个配置项。集群模式除了上面的三个配置项还要增加下面几个配置项:
1
2
3
4
|
initLimit=5 syncLimit=2 server.1=192.168.211.1:2888:3888 server.2=192.168.211.2:2888:3888 |
除了修改 zoo.cfg 配置文件,集群模式下还要配置一个文件 myid,这个文件在 dataDir 目录下,这个文件里面就有一个数据就是 A 的值,Zookeeper 启动时会读取这个文件,拿到里面的数据与 zoo.cfg 里面的配置信息比较从而判断到底是那个 server。
Zookeeper 会维护一个具有层次关系的数据结构,它非常类似于一个标准的文件系统,如图 1 所示:
Zookeeper 这种数据结构有如下这些特点:
Zookeeper 作为一个分布式的服务框架,主要用来解决分布式集群中应用系统的一致性问题,它能提供基于类似于文件系统的目录节点树方式的数据存储,但是 Zookeeper 并不是用来专门存储数据的,它的作用主要是用来维护和监控你存储的数据的状态变化。通过监控这些数据状态的变化,从而可以达到基于数据的集群管理,后面将会详细介绍 Zookeeper 能够解决的一些典型问题,这里先介绍一下,Zookeeper 的操作接口和简单使用示例。
客户端要连接 Zookeeper 服务器可以通过创建 org.apache.zookeeper. ZooKeeper 的一个实例对象,然后调用这个类提供的接口来和服务器交互。
前面说了 ZooKeeper 主要是用来维护和监控一个目录节点树中存储的数据的状态,所有我们能够操作 ZooKeeper 的也和操作目录节点树大体一样,如创建一个目录节点,给某个目录节点设置数据,获取某个目录节点的所有子目录节点,给某个目录节点设置权限和监控这个目录节点的状态变化。
这些接口如下表所示:
除了以上这些上表中列出的方法之外还有一些重载方法,如都提供了一个回调类的重载方法以及可以设置特定 Watcher 的重载方法,具体的方法可以参考 org.apache.zookeeper. ZooKeeper 类的 API 说明。
下面给出基本的操作 ZooKeeper 的示例代码,这样你就能对 ZooKeeper 有直观的认识了。下面的清单包括了创建与 ZooKeeper 服务器的连接以及最基本的数据操作:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
// 创建一个与服务器的连接 ZooKeeper zk = new ZooKeeper("localhost:" + CLIENT_PORT, ClientBase.CONNECTION_TIMEOUT, new Watcher() { // 监控所有被触发的事件 public void process(WatchedEvent event) { System.out.println("已经触发了" + event.getType() + "事件!"); } }); // 创建一个目录节点 zk.create("/testRootPath", "testRootData".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); // 创建一个子目录节点 zk.create("/testRootPath/testChildPathOne", "testChildDataOne".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); System.out.println(new String(zk.getData("/testRootPath",false,null))); // 取出子目录节点列表 System.out.println(zk.getChildren("/testRootPath",true)); // 修改子目录节点数据 zk.setData("/testRootPath/testChildPathOne","modifyChildDataOne".getBytes(),-1); System.out.println("目录节点状态:["+zk.exists("/testRootPath",true)+"]"); // 创建另外一个子目录节点 zk.create("/testRootPath/testChildPathTwo", "testChildDataTwo".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); System.out.println(new String(zk.getData("/testRootPath/testChildPathTwo",true,null))); // 删除子目录节点 zk.delete("/testRootPath/testChildPathTwo",-1); zk.delete("/testRootPath/testChildPathOne",-1); // 删除父目录节点 zk.delete("/testRootPath",-1); // 关闭连接 zk.close(); |
输出的结果如下:
1
2
3
4
5
6
7
8
|
已经触发了 None 事件! testRootData [testChildPathOne] 目录节点状态:[5,5,1281804532336,1281804532336,0,1,0,0,12,1,6] 已经触发了 NodeChildrenChanged 事件! testChildDataTwo 已经触发了 NodeDeleted 事件! 已经触发了 NodeDeleted 事件! |
当对目录节点监控状态打开时,一旦目录节点的状态发生变化,Watcher 对象的 process 方法就会被调用。
Zookeeper 从设计模式角度来看,是一个基于观察者模式设计的分布式服务管理框架,它负责存储和管理大家都关心的数据,然后接受观察者的注册,一旦这些数据的状态发生变化,Zookeeper 就将负责通知已经在 Zookeeper 上注册的那些观察者做出相应的反应,从而实现集群中类似 Master/Slave 管理模式,关于 Zookeeper 的详细架构等内部细节可以阅读 Zookeeper 的源码
下面详细介绍这些典型的应用场景,也就是 Zookeeper 到底能帮我们解决那些问题?下面将给出答案。
分布式应用中,通常需要有一套完整的命名规则,既能够产生唯一的名称又便于人识别和记住,通常情况下用树形的名称结构是一个理想的选择,树形的名称结构是一个有层次的目录结构,既对人友好又不会重复。说到这里你可能想到了 JNDI,没错 Zookeeper 的 Name Service 与 JNDI 能够完成的功能是差不多的,它们都是将有层次的目录结构关联到一定资源上,但是 Zookeeper 的 Name Service 更加是广泛意义上的关联,也许你并不需要将名称关联到特定资源上,你可能只需要一个不会重复名称,就像数据库中产生一个唯一的数字主键一样。
Name Service 已经是 Zookeeper 内置的功能,你只要调用 Zookeeper 的 API 就能实现。如调用 create 接口就可以很容易创建一个目录节点。
配置的管理在分布式应用环境中很常见,例如同一个应用系统需要多台 PC Server 运行,但是它们运行的应用系统的某些配置项是相同的,如果要修改这些相同的配置项,那么就必须同时修改每台运行这个应用系统的 PC Server,这样非常麻烦而且容易出错。
像这样的配置信息完全可以交给 Zookeeper 来管理,将配置信息保存在 Zookeeper 的某个目录节点中,然后将所有需要修改的应用机器监控配置信息的状态,一旦配置信息发生变化,每台应用机器就会收到 Zookeeper 的通知,然后从 Zookeeper 获取新的配置信息应用到系统中。
Zookeeper 能够很容易的实现集群管理的功能,如有多台 Server 组成一个服务集群,那么必须要一个“总管”知道当前集群中每台机器的服务状态,一旦有机器不能提供服务,集群中其它集群必须知道,从而做出调整重新分配服务策略。同样当增加集群的服务能力时,就会增加一台或多台 Server,同样也必须让“总管”知道。
Zookeeper 不仅能够帮你维护当前的集群中机器的服务状态,而且能够帮你选出一个“总管”,让这个总管来管理集群,这就是 Zookeeper 的另一个功能 Leader Election。
它们的实现方式都是在 Zookeeper 上创建一个 EPHEMERAL 类型的目录节点,然后每个 Server 在它们创建目录节点的父目录节点上调用 getChildren(String path, boolean watch) 方法并设置 watch 为 true,由于是 EPHEMERAL 目录节点,当创建它的 Server 死去,这个目录节点也随之被删除,所以 Children 将会变化,这时 getChildren上的 Watch 将会被调用,所以其它 Server 就知道已经有某台 Server 死去了。新增 Server 也是同样的原理。
Zookeeper 如何实现 Leader Election,也就是选出一个 Master Server。和前面的一样每台 Server 创建一个 EPHEMERAL 目录节点,不同的是它还是一个 SEQUENTIAL 目录节点,所以它是个 EPHEMERAL_SEQUENTIAL 目录节点。之所以它是 EPHEMERAL_SEQUENTIAL 目录节点,是因为我们可以给每台 Server 编号,我们可以选择当前是最小编号的 Server 为 Master,假如这个最小编号的 Server 死去,由于是 EPHEMERAL 节点,死去的 Server 对应的节点也被删除,所以当前的节点列表中又出现一个最小编号的节点,我们就选择这个节点为当前 Master。这样就实现了动态选择 Master,避免了传统意义上单 Master 容易出现单点故障的问题。
这部分的示例代码如下,完整的代码请看附件:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
void findLeader() throws InterruptedException { byte[] leader = null; try { leader = zk.getData(root + "/leader", true, null); } catch (Exception e) { logger.error(e); } if (leader != null) { following(); } else { String newLeader = null; try { byte[] localhost = InetAddress.getLocalHost().getAddress(); newLeader = zk.create(root + "/leader", localhost, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); } catch (Exception e) { logger.error(e); } if (newLeader != null) { leading(); } else { mutex.wait(); } } } |
共享锁在同一个进程中很容易实现,但是在跨进程或者在不同 Server 之间就不好实现了。Zookeeper 却很容易实现这个功能,实现方式也是需要获得锁的 Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点,然后调用 getChildren方法获取当前的目录节点列表中最小的目录节点是不是就是自己创建的目录节点,如果正是自己创建的,那么它就获得了这个锁,如果不是那么它就调用 exists(String path, boolean watch) 方法并监控 Zookeeper 上目录节点列表的变化,一直到自己创建的节点是列表中最小编号的目录节点,从而获得锁,释放锁很简单,只要删除前面它自己所创建的目录节点就行了。
同步锁的实现代码如下,完整的代码请看附件:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
void getLock() throws KeeperException, InterruptedException{ List< String > list = zk.getChildren(root, false); String[] nodes = list.toArray(new String[list.size()]); Arrays.sort(nodes); if(myZnode.equals(root+"/"+nodes[0])){ doAction(); } else{ waitForLock(nodes[0]); } } void waitForLock(String lower) throws InterruptedException, KeeperException { Stat stat = zk.exists(root + "/" + lower,true); if(stat != null){ mutex.wait(); } else{ getLock(); } } |
Zookeeper 可以处理两种类型的队列:
同步队列用 Zookeeper 实现的实现思路如下:
创建一个父目录 /synchronizing,每个成员都监控标志(Set Watch)位目录 /synchronizing/start 是否存在,然后每个成员都加入这个队列,加入队列的方式就是创建 /synchronizing/member_i 的临时目录节点,然后每个成员获取 / synchronizing 目录的所有目录节点,也就是 member_i。判断 i 的值是否已经是成员的个数,如果小于成员个数等待 /synchronizing/start 的出现,如果已经相等就创建 /synchronizing/start。
用下面的流程图更容易理解:
同步队列的关键代码如下,完整的代码请看附件:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
void addQueue() throws KeeperException, InterruptedException{ zk.exists(root + "/start",true); zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); synchronized (mutex) { List< String > list = zk.getChildren(root, false); if (list.size() < size) { mutex.wait(); } else { zk.create(root + "/start", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } } } |
当队列没满是进入 wait(),然后会一直等待 Watch 的通知,Watch 的代码如下:
1
2
3
4
5
6
7
8
|
public void process(WatchedEvent event) { if(event.getPath().equals(root + "/start") && event.getType() == Event.EventType.NodeCreated){ System.out.println("得到通知"); super.process(event); doAction(); } } |
FIFO 队列用 Zookeeper 实现思路如下:
实现的思路也非常简单,就是在特定的目录下创建 SEQUENTIAL 类型的子目录 /queue_i,这样就能保证所有成员加入队列时都是有编号的,出队列时通过 getChildren( ) 方法可以返回当前所有的队列中的元素,然后消费其中最小的一个,这样就能保证 FIFO。
下面是生产者和消费者这种队列形式的示例代码,完整的代码请看附件:
1
2
3
4
5
6
7
8
9
|
boolean produce(int i) throws KeeperException, InterruptedException{ ByteBuffer b = ByteBuffer.allocate(4); byte[] value; b.putInt(i); value = b.array(); zk.create(root + "/element", value, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL); return true; } |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
int consume() throws KeeperException, InterruptedException{ int retvalue = -1; Stat stat = null; while (true) { synchronized (mutex) { List< String > list = zk.getChildren(root, true); if (list.size() == 0) { mutex.wait(); } else { Integer min = new Integer(list.get(0).substring(7)); for(String s : list){ Integer tempValue = new Integer(s.substring(7)); if(tempValue < min) min = tempValue; } byte[] b = zk.getData(root + "/element" + min,false, stat); zk.delete(root + "/element" + min, 0); ByteBuffer buffer = ByteBuffer.wrap(b); retvalue = buffer.getInt(); return retvalue; } } } } |
Zookeeper 作为 Hadoop 项目中的一个子项目,是 Hadoop 集群管理的一个必不可少的模块,它主要用来控制集群中的数据,如它管理 Hadoop 集群中的 NameNode,还有 Hbase 中 Master Election、Server 之间状态同步等。
本文介绍的 Zookeeper 的基本知识,以及介绍了几个典型的应用场景。这些都是 Zookeeper 的基本功能,最重要的是 Zoopkeeper 提供了一套很好的分布式集群管理的机制,就是它这种基于层次型的目录树的数据结构,并对树中的节点进行有效管理,从而可以设计出多种多样的分布式的数据管理模型,而不仅仅局限于上面提到的几个常用应用场景。
Netty 是一个利用 Java 的高级网络的能力,隐藏其背后的复杂性而提供一个易于使用的 API 的客户端/服务器框架。
Netty 是一个广泛使用的 Java 网络编程框架(Netty 在 2011 年获得了Duke‘s Choice Award,见https://www.java.net/dukeschoice/2011)。它活跃和成长于用户社区,像大型公司 Facebook 和 Instagram 以及流行 开源项目如 Infinispan, HornetQ, Vert.x, Apache Cassandra 和 Elasticsearch 等,都利用其强大的对于网络抽象的核心代码。
以上是摘自《Essential Netty In Action》这本书,本文的内容也是本人读了这本书之后的一些整理心得,如有不当之处欢迎大虾们指正
Netty和Tomcat最大的区别就在于通信协议,Tomcat是基于Http协议的,他的实质是一个基于http协议的web容器,但是Netty不一样,他能通过编程自定义各种协议,因为netty能够通过codec自己来编码/解码字节流,完成类似redis访问的功能,这就是netty和tomcat最大的不同。
有人说netty的性能就一定比tomcat性能高,其实不然,tomcat从6.x开始就支持了nio模式,并且后续还有arp模式——一种通过jni调用apache网络库的模式,相比于旧的bio模式,并发性能得到了很大提高,特别是arp模式,而netty是否比tomcat性能更高,则要取决于netty程序作者的技术实力了。
如第一部分所述,netty是一款收到大公司青睐的框架,在我看来,netty能够受到青睐的原因有三:
Netty是一款基于NIO(Nonblocking I/O,非阻塞IO)开发的网络通信框架,对比于BIO(Blocking I/O,阻塞IO),他的并发性能得到了很大提高,两张图让你了解BIO和NIO的区别:
Selector
。
除了BIO和NIO之外,还有一些其他的IO模型,下面这张图就表示了五种IO模型的处理流程:
以上摘自Linux IO模式及 select、poll、epoll详解
Netty的传输快其实也是依赖了NIO的一个特性——零拷贝。我们知道,Java的内存有堆内存、栈内存和字符串常量池等等,其中堆内存是占用内存空间最大的一块,也是Java对象存放的地方,一般我们的数据如果需要从IO读取到堆内存,中间需要经过Socket缓冲区,也就是说一个数据会被拷贝两次才能到达他的的终点,如果数据量大,就会造成不必要的资源浪费。
Netty针对这种情况,使用了NIO中的另一大特性——零拷贝,当他需要接收数据的时候,他会在堆内存之外开辟一块内存,数据就直接从IO读到了那块内存中去,在netty里面通过ByteBuf可以直接对这些数据进行直接操作,从而加快了传输速度。
下两图就介绍了两种拷贝方式的区别,摘自Linux 中的零拷贝技术,第 1 部分
上文介绍的ByteBuf是Netty的一个重要概念,他是netty数据处理的容器,也是Netty封装好的一个重要体现,将在下一部分做详细介绍。
要说Netty为什么封装好,这种用文字是说不清的,直接上代码:
public class PlainOioServer {
public void serve(int port) throws IOException {
final ServerSocket socket = new ServerSocket(port); //1
try {
for (;;) {
final Socket clientSocket = socket.accept(); //2
System.out.println("Accepted connection from " + clientSocket);
new Thread(new Runnable() { //3
@Override
public void run() {
OutputStream out;
try {
out = clientSocket.getOutputStream();
out.write("Hi!\r\n".getBytes(Charset.forName("UTF-8"))); //4
out.flush();
clientSocket.close(); //5
} catch (IOException e) {
e.printStackTrace();
try {
clientSocket.close();
} catch (IOException ex) {
// ignore on close
}
}
}
}).start(); //6
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
public class PlainNioServer {
public void serve(int port) throws IOException {
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
ServerSocket ss = serverChannel.socket();
InetSocketAddress address = new InetSocketAddress(port);
ss.bind(address); //1
Selector selector = Selector.open(); //2
serverChannel.register(selector, SelectionKey.OP_ACCEPT); //3
final ByteBuffer msg = ByteBuffer.wrap("Hi!\r\n".getBytes());
for (;;) {
try {
selector.select(); //4
} catch (IOException ex) {
ex.printStackTrace();
// handle exception
break;
}
Set<SelectionKey> readyKeys = selector.selectedKeys(); //5
Iterator<SelectionKey> iterator = readyKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
try {
if (key.isAcceptable()) { //6
ServerSocketChannel server =
(ServerSocketChannel)key.channel();
SocketChannel client = server.accept();
client.configureBlocking(false);
client.register(selector, SelectionKey.OP_WRITE |
SelectionKey.OP_READ, msg.duplicate()); //7
System.out.println(
"Accepted connection from " + client);
}
if (key.isWritable()) { //8
SocketChannel client =
(SocketChannel)key.channel();
ByteBuffer buffer =
(ByteBuffer)key.attachment();
while (buffer.hasRemaining()) {
if (client.write(buffer) == 0) { //9
break;
}
}
client.close(); //10
}
} catch (IOException ex) {
key.cancel();
try {
key.channel().close();
} catch (IOException cex) {
// 在关闭时忽略
}
}
}
}
}
}
public class NettyOioServer {
public void server(int port) throws Exception {
final ByteBuf buf = Unpooled.unreleasableBuffer(
Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
EventLoopGroup group = new OioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); //1
b.group(group) //2
.channel(OioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {//3
@Override
public void initChannel(SocketChannel ch)
throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() { //4
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);//5
}
});
}
});
ChannelFuture f = b.bind().sync(); //6
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully().sync(); //7
}
}
}
从代码量上来看,Netty就已经秒杀传统Socket编程了,但是这一部分博大精深,仅仅贴几个代码岂能说明问题,在这里给大家介绍一下Netty的一些重要概念,让大家更理解Netty。
Channel
数据传输流,与channel相关的概念有以下四个,上一张图让你了解netty里面的Channel。
他有三种使用模式:
除此之外,他还提供一大堆api方便你使用,在这里我就不一一列出了,具体参见ByteBuf字节缓存。
Netty入门教程2——动手搭建HttpServer
Netty入门教程3——Decoder和Encoder
Netty入门教程4——如何实现长连接
以上就是我对《Netty实战》这本书的一些心得和书外的一些相关知识整理,如果有不同的见解,欢迎讨论!
标签:att for 分组 复合 scheme cat lis channel member
原文地址:https://www.cnblogs.com/xuwc/p/8975776.html