深入浅出Zookeeper之四Create请求和处理_小组_ThinkSAAS
深入浅出Zookeeper之四Create请求和处理
深入浅出Zookeeper之四Create请求和处理
客户端接口
public String create(final String path, byte data[], List&ACL& acl,
CreateMode createMode)
throws KeeperException, InterruptedException
final String clientPath =
PathUtils.validatePath(clientPath, createMode.isSequential());
final String serverPath = prependChroot(clientPath);
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.create);
CreateRequest request = new CreateRequest();
//CREATE请求需要server端响应
CreateResponse response = new CreateResponse();
request.setData(data);
//node类型
request.setFlags(createMode.toFlag());
request.setPath(serverPath);
if (acl != null && acl.size() == 0) {
throw new KeeperException.InvalidACLException();
request.setAcl(acl);
//同步提交
ReplyHeader r = cnxn.submitRequest(h, request, response, null);
//异常情况
if (r.getErr() != 0) {
throw KeeperException.create(KeeperException.Code.get(r.getErr()),
clientPath);
//真实路径，对于SEQUENTIAL NODE后面会加上序号
if (cnxn.chrootPath == null) {
return response.getPath();
return response.getPath().substring(cnxn.chrootPath.length());
请求提交过程和之前的exists一样，都是通过sendthread写出去，都会进入pengding队列等待server端返回。
server端处理也是一样，变化的只是RequestProcessor的业务逻辑。也就是说transport层是通用的，变化的是上层的业务层。
server端执行处理链，PrepRequestProcessor
switch (request.type) {
case OpCode.create:
//反序列化的对象
CreateRequest createRequest = new CreateRequest();
//zxid递增
pRequest2Txn(request.type, zks.getNextZxid(), request, createRequest, true);
request.zxid = zks.getZxid();
nextProcessor.processRequest(request);
protected void pRequest2Txn(int type, long zxid, Request request, Record record, boolean deserialize)
throws KeeperException, IOException, RequestProcessorException
//构造内部的事务头
request.hdr = new TxnHeader(request.sessionId, request.cxid, zxid,
zks.getTime(), type);
switch (type) {
case OpCode.create:
//检查session是否还有效
zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
CreateRequest createRequest = (CreateRequest)
//反序列化
if(deserialize)
ByteBufferInputStream.byteBuffer2Record(request.request, createRequest);
String path = createRequest.getPath();
//路径规则检查
int lastSlash = path.lastIndexOf('/');
if (lastSlash == -1 || path.indexOf('') != -1 || failCreate) {
("Invalid path"+ path +"with session 0x"+
Long.toHexString(request.sessionId));
throw new KeeperException.BadArgumentsException(path);
//权限去重
List&ACL& listACL = removeDuplicates(createRequest.getAcl());
if (!fixupACL(request.authInfo, listACL)) {
throw new KeeperException.InvalidACLException(path);
String parentPath = path.substring(0, lastSlash);
//父节点的ChangeRecord
ChangeRecord parentRecord = getRecordForPath(parentPath);
//权限校验
checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE,
request.authInfo);
int parentCVersion = parentRecord.stat.getCversion();
CreateMode createMode =
CreateMode.fromFlag(createRequest.getFlags());
//SEQUENCE节点修改路径
if (createMode.isSequential()) {
path = path + String.format(Locale.ENGLISH,"%010d", parentCVersion);
PathUtils.validatePath(path);
} catch(IllegalArgumentException ie) {
("Invalid path"+ path +"with session 0x"+
Long.toHexString(request.sessionId));
throw new KeeperException.BadArgumentsException(path);
//当前节点ChangeRecord，一并校验节点是否已经存在
if (getRecordForPath(path) != null) {
throw new KeeperException.NodeExistsException(path);
} catch (KeeperException.NoNodeException e) {
// ignore this one
//EPHEMERAL节点不允许创建子节点
boolean ephemeralParent = parentRecord.stat.getEphemeralOwner() != 0;
if (ephemeralParent) {
throw new KeeperException.NoChildrenForEphemeralsException(path);
//递增Cversion
int newCversion = parentRecord.stat.getCversion()+1;
//构造事务体，后续会被写入log
request.txn = new CreateTxn(path, createRequest.getData(),
createMode.isEphemeral(), newCversion);
StatPersisted s = new StatPersisted();
//如果是临时节点，则owner为sessionId
if (createMode.isEphemeral()) {
s.setEphemeralOwner(request.sessionId);
parentRecord = parentRecord.duplicate(request.hdr.getZxid());
//递增parent的child数
parentRecord.childCount++;
//递增parent的cversion
parentRecord.stat.setCversion(newCversion);
//添加changeRecord到冗余队列
addChangeRecord(parentRecord);
addChangeRecord(new ChangeRecord(request.hdr.getZxid(), path, s,
0, listACL));
SyncRequestProcessor处理，主要是log写入，和之前的分析类似，不赘述
FinalRequestProcessor处理，修改内存中的datatree结构
switch (header.getType()) {
case OpCode.create:
CreateTxn createTxn = (CreateTxn)
rc.path = createTxn.getPath();
createNode(
createTxn.getPath(),
createTxn.getData(),
createTxn.getAcl(),
createTxn.getEphemeral() ? header.getClientId() : 0,
createTxn.getParentCVersion(),
header.getZxid(), header.getTime());
public String createNode(String path, byte data[], List&ACL& acl,
long ephemeralOwner, int parentCVersion, long zxid, long time)
throws KeeperException.NoNodeException,
KeeperException.NodeExistsException {
//各种参数设置
int lastSlash = path.lastIndexOf('/');
String parentName = path.substring(0, lastSlash);
String childName = path.substring(lastSlash + 1);
StatPersisted stat = new StatPersisted();
stat.setCtime(time);
stat.setMtime(time);
stat.setCzxid(zxid);
stat.setMzxid(zxid);
stat.setPzxid(zxid);
stat.setVersion(0);
stat.setAversion(0);
stat.setEphemeralOwner(ephemeralOwner);
DataNode parent = nodes.get(parentName);
if (parent == null) {
throw new KeeperException.NoNodeException();
//修改parent节点内容
synchronized (parent) {
Set&String& children = parent.getChildren();
if (children != null) {
if (children.contains(childName)) {
throw new KeeperException.NodeExistsException();
if (parentCVersion == -1) {
parentCVersion = parent.stat.getCversion();
parentCVersion++;
parent.stat.setCversion(parentCVersion);
parent.stat.setPzxid(zxid);
Long longval = convertAcls(acl);
//添加目标节点
DataNode child = new DataNode(parent, data, longval, stat);
parent.addChild(childName);
nodes.put(path, child);
//添加ephemeral类型节点
if (ephemeralOwner != 0) {
HashSet&String& list = ephemerals.get(ephemeralOwner);
if (list == null) {
list = new HashSet&String&();
ephemerals.put(ephemeralOwner, list);
synchronized (list) {
list.add(path);
//dataWatches和childWatches事件触发
dataWatches.triggerWatch(path, Event.EventType.NodeCreated);
childWatches.triggerWatch(parentName.equals("") ?"/": parentName,
Event.EventType.NodeChildrenChanged);
事件触发过程
public Set&Watcher& triggerWatch(String path, EventType type, Set&Watcher& supress) {
WatchedEvent e = new WatchedEvent(type,
KeeperState.SyncConnected, path);
HashSet&Watcher&
synchronized (this) {
//拿节点对应的watcher列表，只通知一次
watchers = watchTable.remove(path);
for (Watcher w : watchers) {
HashSet&String& paths = watch2Paths.get(w);
if (paths != null) {
paths.remove(path);
//挨个通知
for (Watcher w : watchers) {
if (supress != null && supress.contains(w)) {
w.process(e);
具体处理，NIOServerCncx
synchronized public void process(WatchedEvent event) {
//事件通知的响应头
ReplyHeader h = new ReplyHeader(-1, -1L, 0);
// Convert WatchedEvent to a type that can be sent over the wire
WatcherEvent e = event.getWrapper();
//发送响应
sendResponse(h, e,"notification");
具体发送：
synchronized public void sendResponse(ReplyHeader h, Record r, String tag) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
// Make space for length
BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
baos.write(fourBytes);
bos.writeRecord(h,"header");
if (r != null) {
bos.writeRecord(r, tag);
baos.close();
} catch (IOException e) {
LOG.error("Error serializing response");
//转换成byte数组
byte b[] = baos.toByteArray();
ByteBuffer bb = ByteBuffer.wrap(b);
//最后写入长度
bb.putInt(b.length - 4).rewind();
//通过channel异步写
sendBuffer(bb);
if (h.getXid() & 0) {
synchronized(this){
outstandingRequests--;
// check throttling
synchronized (this.factory) {
if (zkServer.getInProcess() & outstandingLimit
|| outstandingRequests & 1) {
sk.selector().wakeup();
enableRecv();
} catch(Exception e) {
LOG.warn("Unexpected exception. Destruction averted.", e);
server端就处理完了，接下来client端SendThread收到响应
if (replyHdr.getXid() == -1) {
// -1 means notification
if (LOG.isDebugEnabled()) {
LOG.debug("Got notification sessionid:0x"
+ Long.toHexString(sessionId));
//反序列化
WatcherEvent event = new WatcherEvent();
event.deserialize(bbia,"response");
// convert from a server path to a client path
//路径切换，如果client用了相对路径的话
if (chrootPath != null) {
String serverPath = event.getPath();
pareTo(chrootPath)==0)
event.setPath("/");
else if (serverPath.length() & chrootPath.length())
event.setPath(serverPath.substring(chrootPath.length()));
LOG.warn("Got server path"+ event.getPath()
+"which is too short for chroot path"
+ chrootPath);
//通过eventThread派发事件，通知之前注册的watcher
WatchedEvent we = new WatchedEvent(event);
if (LOG.isDebugEnabled()) {
LOG.debug("Got"+ we +"for sessionid 0x"
+ Long.toHexString(sessionId));
eventThread.queueEvent( we );
eventThread端处理和之前类似
case NodeCreated:
//通知dataWatcher，只一次
synchronized (dataWatches) {
addTo(dataWatches.remove(clientPath), result);
//通知existWatcher，只一次
synchronized (existWatches) {
addTo(existWatches.remove(clientPath), result);
if (event instanceof WatcherSetEventPair) {
// each watcher will process the event
WatcherSetEventPair pair = (WatcherSetEventPair)
//挨个通知
for (Watcher watcher : pair.watchers) {
watcher.process(pair.event);
} catch (Throwable t) {
LOG.error("Error while calling watcher", t);
以上就完成了create操作的处理，之后的delete，set，get操作都是类似的，大家有兴趣的直接阅读相关代码即可:)
PHP开发框架
服务器环境
ThinkSAAS商业授权：
ThinkSAAS为用户提供有偿个性定制开发服务
ThinkSAAS将为商业授权用户提供二次开发指导和技术支持
手机客户端
ThinkSAAS接收任何功能的Iphone(IOS)和Android手机的客户端定制开发服务
让ThinkSAAS更好，把建议拿来。Zookeeper全解析
先说Paxos，它是一个基于消息传递的一致性算法，Leslie Lamport在1990年提出，近几年被广泛应用于分布式计算中，Google的Chubby，Apache的Zookeeper都是基于它的理论来实现的，Paxos还被认为是到目前为止唯一的分布式一致性算法，其它的算法都是Paxos的改进或简化。有个问题要提一下，Paxos有一个前提：没有拜占庭将军问题。就是说Paxos只有在一个可信的计算环境中才能成立，这个环境是不会被入侵所破坏的。
关于Paxos的具体描述可以在Wiki中找到：http://zh.wikipedia.org/zh-cn/Paxos算法。网上关于Paxos分析的文章也很多。这里希望用最简单的方式加以描述并建立起Paxos和ZK Server的对应关系。
Paxos描述了这样一个场景，有一个叫做Paxos的小岛(Island)上面住了一批居民，岛上面所有的事情由一些特殊的人决定，他们叫做议员(Senator)。议员的总数(Senator Count)是确定的，不能更改。岛上每次环境事务的变更都需要通过一个提议(Proposal)，每个提议都有一个编号(PID)，这个编号是一直增长的，不能倒退。每个提议都需要超过半数((Senator Count)/2 +1)的议员同意才能生效。每个议员只会同意大于当前编号的提议，包括已生效的和未生效的。如果议员收到小于等于当前编号的提议，他会拒绝，并告知对方：你的提议已经有人提过了。这里的当前编号是每个议员在自己记事本上面记录的编号，他不断更新这个编号。整个议会不能保证所有议员记事本上的编号总是相同的。现在议会有一个目标：保证所有的议员对于提议都能达成一致的看法。
好，现在议会开始运作，所有议员一开始记事本上面记录的编号都是0。有一个议员发了一个提议：将电费设定为1元/度。他首先看了一下记事本，嗯，当前提议编号是0，那么我的这个提议的编号就是1，于是他给所有议员发消息：1号提议，设定电费1元/度。其他议员收到消息以后查了一下记事本，哦，当前提议编号是0，这个提议可接受，于是他记录下这个提议并：我接受你的1号提议，同时他在记事本上记录：当前提议编号为1。发起提议的议员收到了超过半数的，立即给所有人发通知：1号提议生效！收到的议员会修改他的记事本，将1好提议由记录改成正式的法令，当有人问他电费为多少时，他会查看法令并告诉对方：1元/度。
现在看冲突的解决：假设总共有三个议员S1-S3，S1和S2同时发起了一个提议:1号提议，设定电费。S1想设为1元/度, S2想设为2元/度。结果S3先收到了S1的提议，于是他做了和前面同样的操作。紧接着他又收到了S2的提议，结果他一查记事本，咦，这个提议的编号小于等于我的当前编号1，于是他拒绝了这个提议：对不起，这个提议先前提过了。于是S2的提议被拒绝，S1正式发布了提议: 1号提议生效。S2向S1或者S3打听并更新了1号法令的内容，然后他可以选择继续发起2号提议。
好，我觉得Paxos的精华就这么多内容。现在让我们来对号入座，看看在ZK Server里面Paxos是如何得以贯彻实施的。
小岛(Island)——ZK Server Cluster
议员(Senator)——ZK Server
提议(Proposal)——ZNode Change(Create/Delete/SetData…)
提议编号(PID)——Zxid(ZooKeeper Transaction Id)
正式法令——所有ZNode及其数据
貌似关键的概念都能一一对应上，但是等一下，Paxos岛上的议员应该是人人平等的吧，而ZK Server好像有一个Leader的概念。没错，其实Leader的概念也应该属于Paxos范畴的。如果议员人人平等，在某种情况下会由于提议的冲突而产生一个“活锁”（所谓活锁我的理解是大家都没有死，都在动，但是一直解决不了冲突问题）。Paxos的作者Lamport在他的文章”The Part-Time Parliament“中阐述了这个问题并给出了解决方案——在所有议员中设立一个总统，只有总统有权发出提议，如果议员有自己的提议，必须发给总统并由总统来提出。好，我们又多了一个角色：总统。
总统——ZK Server Leader
又一个问题产生了，总统怎么选出来的？oh, my god! It’s a long story. 在淘宝核心系统团队的Blog上面有一篇文章是介绍如何选出总统的，有兴趣的可以去看看：/blog/cs/?p=162
现在我们假设总统已经选好了，下面看看ZK Server是怎么实施的。
屁民甲(Client)到某个议员(ZK Server)那里询问(Get)某条法令的情况(ZNode的数据)，议员毫不犹豫的拿出他的记事本(local storage)，查阅法令并告诉他结果，同时声明：我的数据不一定是最新的。你想要最新的数据？没问题，等着，等我找总统Sync一下再告诉你。
屁民乙(Client)到某个议员(ZK Server)那里要求政府归还欠他的一万元钱，议员让他在办公室等着，自己将问题反映给了总统，总统询问所有议员的意见，多数议员表示欠屁民的钱一定要还，于是总统发表声明，从国库中拿出一万元还债，国库总资产由100万变成99万。屁民乙拿到钱回去了(Client函数返回)。
总统突然挂了，议员接二连三的发现联系不上总统，于是各自发表声明，推选新的总统，总统大选期间政府停业，拒绝屁民的请求。
，到此为止吧，当然还有很多其他的情况，但这些情况总是能在Paxos的算法中找到原型并加以解决。这也正是我们认为Paxos是Zookeeper的灵魂的原因。当然ZK Server还有很多属于自己特性的东西：Session, Watcher，Version等等等等，需要我们花更多的时间去研究和学习。
/2010/08/zookeeper%e5%85%a8%e8%a7%a3%e6%9e%90%e2%80%94%e2%80%94paxos%e7%9a%84%e7%81%b5%e9%ad%82/
更多相关文章
原计划在介绍完ZK Client之后就着手ZK Server的介绍,但是发现ZK Server所包含的内容实在太多,并不是简简单单一篇Blog就能搞定的.于是决定从基础搞起比较好. 那么ZK Server最基础的东西是什么呢?我想应该是Paxos了.所以本文会介绍Paxos以及它在ZK Server ...
Zookeeper的Client直接与用户打交道,是我们使用Zookeeper的interface.了解ZK Client的结构和工作原理有利于我们合理的使用ZK,并能在使用中更早的发现问题.本文将在研究源码的技术上讲述ZK Client的工作原理及内部工作机制. 在看完ZK Client的大致架构 ...
原计划在介绍完ZK Client之后就着手ZK Server的介绍,但是发现ZK Server所包含的内容实在太多,并不是简简单单一篇Blog就能搞定的.于是决定从基础搞起比较好. 那么ZK Server最基础的东西是什么呢?我想应该是Paxos了.所以本文会介绍Paxos以及它在ZK Server ...
原文地址: /2010/08/zookeeper%E5%85%A8%E8%A7%A3%E6%9E%90%E2%80%94%E2%80%94paxos%E7%9A%84%E7%81%B5%E9%AD%82/?spm=0.0.0.0.h43YQe原计划在介绍完 ...
Zookeeper的Client直接与用户打交道,是我们使用Zookeeper的interface.了解ZK Client的结构和工作原理有利于我们合理的使用ZK,并能在使用中更早的发现问题.本文将在研究源码的技术上讲述ZK Client的工作原理及内部工作机制. 在看完ZK Client的大致架构 ...
update:EPSG对该投影的编号设定为EPSG:3857,对应的WKT也发生了变化,下文不再修改,相对来说格式都是那样,可以到http://www.epsg-registry.org 网站输入SRID进行查询.
Google Maps和Virtual Earth等 ...
Unity3D移动平台动态读取外部文件全解析
c#语言规范 阅读目录 前言: 假如我想在editor里动态读取文件 移动平台的资源路径问题 移动平台读取外部文件的方法 补充: 回到目录PC端和移动端的区别,那么下一个问题自然而然的就是移动端的资源路径(要讨论一下Resources.Streami ...
iOS Storyboard全解析
来源:/blog/1493956 Storyboard)是一个能够节省你很多设计手机App界面时间的新特性,下面,为了简明的说明Storyboard的效果,我贴上本教程所完成的Storyboard的截图:
现 在, ...
[mysqld] port = 3306 serverid = 1 socket = /tmp/mysql.sock skip-locking # 避免MySQL的外部锁定,减少出错几率增强稳定性. skip-net ...
首先要定义四个变量: int pageSize: //每页显示多少条记录 int pageNow: //希望显示第几页 int pageCount: //一共有多少页 int rowCount: //一共有多少条记录 ...
提出一种方法——ELBlocker,用于自动检测出Blockin ...
转自:/html/18/t-618.html?action-uchimage
函数cvDrawContours用于在图像上绘制外部和内部轮廓.当thickness & ...
项目界面设计的时候,发现在设置button的enabled=false后,原本设计的字体颜色跟预设的不一样,查了一些资料后,在网上看到这样一段代码: [System.Runtime.InteropServices.D ...
一.访问另一个物体 1.代码中定义一个public的物体 例如:var target:Tr ...
阅读下文前,建议对OBD有初步了解,可阅读下面两篇: 玩转车联网1初识OBD和行车助手 玩 ...
总体设计 1.本程序要实现五子棋的游戏功能,必须先有一个棋盘,所以,通过继承JPanel, ...
可以比对apk签名的fingerprint.假定安装了JDK,如果想查HelloWorld.apk所使用的签名的fingerprint,可以这样做:1. 查找apk里的rsa文件(Windows)& jar t ...
世界上并没有成为高手的捷径,但一些基本原则是可以遵循的. 1. 扎实的基础.数据结构.离散数学.编译原理,这些是所有计算机科学的基础,如果不掌握他们,很难写出高水平的程序.据我的观察,学计算机专业的人比学其他专业的人 ...广告剩余8秒
文档加载中
ZooKeeper在复杂事件处理系统中的应用,zookeeper 事件通知,zookeeper 事件,zooke..
扫扫二维码，随身浏览文档
手机或平板扫扫即可继续访问
ZooKeeper在复杂事件处理系统中的应用
举报该文档为侵权文档。
举报该文档含有违规或不良信息。
反馈该文档无法正常浏览。
举报该文档为重复文档。
推荐理由：
将文档分享至：
分享完整地址
文档地址：
粘贴到BBS或博客
flash地址：
支持嵌入FLASH地址的网站使用
html代码：
&embed src='/DocinViewer--144.swf' width='100%' height='600' type=application/x-shockwave-flash ALLOWFULLSCREEN='true' ALLOWSCRIPTACCESS='always'&&/embed&
450px*300px480px*400px650px*490px
支持嵌入HTML代码的网站使用
您的内容已经提交成功
您所提交的内容需要审核后才能发布，请您等待！
3秒自动关闭窗口zookeeper学习笔记_图文_百度文库
两大类热门资源免费畅读
续费一年阅读会员，立省24元！
评价文档：
zookeeper学习笔记
上传于||文档简介
&&z​o​o​k​e​e​p​e​r​学​习​笔​记
大小：613.00KB
登录百度文库，专享文档复制特权，财富值每天免费拿！
你可能喜欢&&&您需要以后才能回答，未注册用户请先。