todo
OpCode.createSession&zks.finishSessionInit(request.cnxn, true);
FinalRequestProcessor
LOG.debug("Processing request:: {}", request);
// 日志記錄
if (LOG.isTraceEnabled())
{
long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
if (request.type == OpCode.ping)
{
traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
}
ZooTrace.logRequest(LOG, traceMask, 'E', request, "");
}
ProcessTxnResult rc = null;
// 請求未被過濾--對其處理
if (!request.isThrottled())
{
rc = applyRequest(request);
}
// 需要進行回復的請求物件會包含cnxn
if (request.cnxn == null)
{
return;
}
// 獲取請求中的ServerCcxn以便進行回復
ServerCnxn cnxn = request.cnxn;
// 獲得LastProcessedZxid
long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
String lastOp = "NA";
// 更新zk服務物件處理中請求個數
zks.decInProcess();
// zk服務物件執行請求完成處理
zks.requestFinished(request);
// 大請求長度
int largeRequestLength = request.getLargeRequestSize();
if (largeRequestLength != -1)
{
// 更新currentLargeRequestBytes
currentLargeRequestBytes.addAndGet(-largeRequestLength);
}
Code err = Code.OK;
Record rsp = null;
String path = null;
int responseSize = 0;
try
{
// 如果請求的事務頭不是null
// 且 請求的事務頭的型別為錯誤
// 表示請求處理出錯了
if (request.getHdr() != null && request.getHdr().getType() == OpCode.error)
{
AuditHelper.addAuditLog(request, rc, true);
// 獲取請求中的例外物件
if (request.getException() != null)
{
// 再次拋出
throw request.getException();
}
else
{
// 構造例外物件并拋出
throw KeeperException.
create(KeeperException.Code.get(((ErrorTxn) request.getTxn()).getErr()));
}
}
// 獲取請求中的例外物件
KeeperException ke = request.getException();
// 如果是會話移動例外
if (ke instanceof SessionMovedException)
{
// 再次拋出
throw ke;
}
// 如果請求存在例外物件，且不是復合請求
if (ke != null && request.type != OpCode.multi)
{
// 再次拋出
throw ke;
}
LOG.debug("{}", request);
// 如果請求已經過期，
if (request.isStale())
{
// 統計
ServerMetrics.getMetrics().STALE_REPLIES.add(1);
}
// 如果請求被過濾
if (request.isThrottled())
{
// 拋出例外
throw KeeperException.create(Code.THROTTLEDOP);
}
AuditHelper.addAuditLog(request, rc);
// 請求型別
switch (request.type)
{
// 請求為ping
case OpCode.ping:
{
lastOp = "PING";
updateStats(request, lastOp, lastZxid);
// 發送回復
responseSize = cnxn.sendResponse(
new ReplyHeader(ClientCnxn.PING_XID, lastZxid, 0), null, "response");
return;
}
// 請求為createSession
case OpCode.createSession:
{
lastOp = "SESS";
updateStats(request, lastOp, lastZxid);
zks.finishSessionInit(request.cnxn, true);
return;
}
case OpCode.multi:
{
lastOp = "MULT";
rsp = new MultiResponse();
for (ProcessTxnResult subTxnResult : rc.multiResult)
{
OpResult subResult;
switch (subTxnResult.type)
{
case OpCode.check:
subResult = new CheckResult();
break;
case OpCode.create:
subResult = new CreateResult(subTxnResult.path);
break;
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer:
subResult = new CreateResult(subTxnResult.path, subTxnResult.stat);
break;
case OpCode.delete:
case OpCode.deleteContainer:
subResult = new DeleteResult();
break;
case OpCode.setData:
subResult = new SetDataResult(subTxnResult.stat);
break;
case OpCode.error:
subResult = new ErrorResult(subTxnResult.err);
if (subTxnResult.err == Code.SESSIONMOVED.intValue())
{
throw new SessionMovedException();
}
break;
default:
throw new IOException("Invalid type of op");
}
((MultiResponse) rsp).add(subResult);
}
break;
}
case OpCode.multiRead:
{
lastOp = "MLTR";
MultiOperationRecord multiReadRecord = new MultiOperationRecord();
ByteBufferInputStream.byteBuffer2Record(request.request, multiReadRecord);
rsp = new MultiResponse();
OpResult subResult;
for (Op readOp : multiReadRecord)
{
try
{
Record rec;
switch (readOp.getType())
{
case OpCode.getChildren:
rec = handleGetChildrenRequest(readOp.toRequestRecord(), cnxn, request.authInfo);
subResult = new GetChildrenResult(((GetChildrenResponse) rec).getChildren());
break;
case OpCode.getData:
rec = handleGetDataRequest(readOp.toRequestRecord(), cnxn, request.authInfo);
GetDataResponse gdr = (GetDataResponse) rec;
subResult = new GetDataResult(gdr.getData(), gdr.getStat());
break;
default:
throw new IOException("Invalid type of readOp");
}
}
catch (KeeperException e)
{
subResult = new ErrorResult(e.code().intValue());
}
((MultiResponse) rsp).add(subResult);
}
break;
}
case OpCode.create:
{
lastOp = "CREA";
rsp = new CreateResponse(rc.path);
err = Code.get(rc.err);
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer:
{
lastOp = "CREA";
rsp = new Create2Response(rc.path, rc.stat);
err = Code.get(rc.err);
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
case OpCode.delete:
case OpCode.deleteContainer:
{
lastOp = "DELE";
err = Code.get(rc.err);
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
case OpCode.setData:
{
lastOp = "SETD";
rsp = new SetDataResponse(rc.stat);
err = Code.get(rc.err);
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
case OpCode.reconfig:
{
lastOp = "RECO";
rsp = new GetDataResponse(((QuorumZooKeeperServer) zks).self.getQuorumVerifier().toString().getBytes(UTF_8), rc.stat);
err = Code.get(rc.err);
break;
}
case OpCode.setACL:
{
lastOp = "SETA";
rsp = new SetACLResponse(rc.stat);
err = Code.get(rc.err);
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
case OpCode.closeSession:
{
lastOp = "CLOS";
err = Code.get(rc.err);
break;
}
case OpCode.sync:
{
lastOp = "SYNC";
SyncRequest syncRequest = new SyncRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, syncRequest);
rsp = new SyncResponse(syncRequest.getPath());
requestPathMetricsCollector.registerRequest(request.type, syncRequest.getPath());
break;
}
case OpCode.check:
{
lastOp = "CHEC";
rsp = new SetDataResponse(rc.stat);
err = Code.get(rc.err);
break;
}
case OpCode.exists:
{
lastOp = "EXIS";
ExistsRequest existsRequest = new ExistsRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, existsRequest);
path = existsRequest.getPath();
if (path.indexOf('\0') != -1)
{
throw new KeeperException.BadArgumentsException();
}
Stat stat = zks.getZKDatabase().statNode(path, existsRequest.getWatch() ? cnxn : null);
rsp = new ExistsResponse(stat);
requestPathMetricsCollector.registerRequest(request.type, path);
break;
}
case OpCode.getData:
{
lastOp = "GETD";
GetDataRequest getDataRequest = new GetDataRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, getDataRequest);
path = getDataRequest.getPath();
rsp = handleGetDataRequest(getDataRequest, cnxn, request.authInfo);
requestPathMetricsCollector.registerRequest(request.type, path);
break;
}
case OpCode.setWatches:
{
lastOp = "SETW";
SetWatches setWatches = new SetWatches();
request.request.rewind();
ByteBufferInputStream.byteBuffer2Record(request.request, setWatches);
long relativeZxid = setWatches.getRelativeZxid();
zks.getZKDatabase().setWatches(relativeZxid, setWatches.getDataWatches(), setWatches.getExistWatches(), setWatches.getChildWatches(), Collections.emptyList(), Collections.emptyList(), cnxn);
break;
}
case OpCode.setWatches2:
{
lastOp = "STW2";
SetWatches2 setWatches = new SetWatches2();
request.request.rewind();
ByteBufferInputStream.byteBuffer2Record(request.request, setWatches);
long relativeZxid = setWatches.getRelativeZxid();
zks.getZKDatabase().setWatches(relativeZxid, setWatches.getDataWatches(), setWatches.getExistWatches(), setWatches.getChildWatches(), setWatches.getPersistentWatches(), setWatches.getPersistentRecursiveWatches(), cnxn);
break;
}
case OpCode.addWatch:
{
lastOp = "ADDW";
AddWatchRequest addWatcherRequest = new AddWatchRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, addWatcherRequest);
zks.getZKDatabase().addWatch(addWatcherRequest.getPath(), cnxn, addWatcherRequest.getMode());
rsp = new ErrorResponse(0);
break;
}
case OpCode.getACL:
{
lastOp = "GETA";
GetACLRequest getACLRequest = new GetACLRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, getACLRequest);
path = getACLRequest.getPath();
DataNode n = zks.getZKDatabase().getNode(path);
if (n == null)
{
throw new KeeperException.NoNodeException();
}
zks.checkACL(request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ | ZooDefs.Perms.ADMIN, request.authInfo, path, null);
Stat stat = new Stat();
List<ACL> acl = zks.getZKDatabase().getACL(path, stat);
requestPathMetricsCollector.registerRequest(request.type, getACLRequest.getPath());
try
{
zks.checkACL(request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.ADMIN, request.authInfo, path, null);
rsp = new GetACLResponse(acl, stat);
}
catch (KeeperException.NoAuthException e)
{
List<ACL> acl1 = new ArrayList<ACL>(acl.size());
for (ACL a : acl)
{
if ("digest".equals(a.getId().getScheme()))
{
Id id = a.getId();
Id id1 = new Id(id.getScheme(), id.getId().replaceAll(":.*", ":x"));
acl1.add(new ACL(a.getPerms(), id1));
}
else
{
acl1.add(a);
}
}
rsp = new GetACLResponse(acl1, stat);
}
break;
}
case OpCode.getChildren:
{
lastOp = "GETC";
GetChildrenRequest getChildrenRequest = new GetChildrenRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, getChildrenRequest);
path = getChildrenRequest.getPath();
rsp = handleGetChildrenRequest(getChildrenRequest, cnxn, request.authInfo);
requestPathMetricsCollector.registerRequest(request.type, path);
break;
}
case OpCode.getAllChildrenNumber:
{
lastOp = "GETACN";
GetAllChildrenNumberRequest getAllChildrenNumberRequest = new GetAllChildrenNumberRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, getAllChildrenNumberRequest);
path = getAllChildrenNumberRequest.getPath();
DataNode n = zks.getZKDatabase().getNode(path);
if (n == null)
{
throw new KeeperException.NoNodeException();
}
zks.checkACL(request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, request.authInfo, path, null);
int number = zks.getZKDatabase().getAllChildrenNumber(path);
rsp = new GetAllChildrenNumberResponse(number);
break;
}
case OpCode.getChildren2:
{
lastOp = "GETC";
GetChildren2Request getChildren2Request = new GetChildren2Request();
ByteBufferInputStream.byteBuffer2Record(request.request, getChildren2Request);
Stat stat = new Stat();
path = getChildren2Request.getPath();
DataNode n = zks.getZKDatabase().getNode(path);
if (n == null)
{
throw new KeeperException.NoNodeException();
}
zks.checkACL(request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, request.authInfo, path, null);
List<String> children = zks.getZKDatabase().getChildren(path, stat, getChildren2Request.getWatch() ? cnxn : null);
rsp = new GetChildren2Response(children, stat);
requestPathMetricsCollector.registerRequest(request.type, path);
break;
}
case OpCode.checkWatches:
{
lastOp = "CHKW";
CheckWatchesRequest checkWatches = new CheckWatchesRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, checkWatches);
WatcherType type = WatcherType.fromInt(checkWatches.getType());
path = checkWatches.getPath();
boolean containsWatcher = zks.getZKDatabase().containsWatcher(path, type, cnxn);
if (!containsWatcher)
{
String msg = String.format(Locale.ENGLISH, "%s (type: %s)", path, type);
throw new KeeperException.NoWatcherException(msg);
}
requestPathMetricsCollector.registerRequest(request.type, checkWatches.getPath());
break;
}
case OpCode.removeWatches:
{
lastOp = "REMW";
RemoveWatchesRequest removeWatches = new RemoveWatchesRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, removeWatches);
WatcherType type = WatcherType.fromInt(removeWatches.getType());
path = removeWatches.getPath();
boolean removed = zks.getZKDatabase().removeWatch(path, type, cnxn);
if (!removed)
{
String msg = String.format(Locale.ENGLISH, "%s (type: %s)", path, type);
throw new KeeperException.NoWatcherException(msg);
}
requestPathMetricsCollector.registerRequest(request.type, removeWatches.getPath());
break;
}
case OpCode.whoAmI:
{
lastOp = "HOMI";
rsp = new WhoAmIResponse(AuthUtil.getClientInfos(request.authInfo));
break;
}
case OpCode.getEphemerals:
{
lastOp = "GETE";
GetEphemeralsRequest getEphemerals = new GetEphemeralsRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, getEphemerals);
String prefixPath = getEphemerals.getPrefixPath();
Set<String> allEphems = zks.getZKDatabase().getDataTree().getEphemerals(request.sessionId);
List<String> ephemerals = new ArrayList<>();
if (prefixPath == null || prefixPath.trim().isEmpty() || "/".equals(prefixPath.trim()))
{
ephemerals.addAll(allEphems);
}
else
{
for (String p : allEphems)
{
if (p.startsWith(prefixPath))
{
ephemerals.add(p);
}
}
}
rsp = new GetEphemeralsResponse(ephemerals);
break;
}
}
}
catch (SessionMovedException e)
{
cnxn.sendCloseSession();
return;
}
catch (KeeperException e)
{
err = e.code();
}
catch (Exception e)
{
LOG.error("Failed to process {}", request, e);
StringBuilder sb = new StringBuilder();
ByteBuffer bb = request.request;
bb.rewind();
while (bb.hasRemaining())
{
sb.append(Integer.toHexString(bb.get() & 0xff));
}
LOG.error("Dumping request buffer: 0x{}", sb.toString());
err = Code.MARSHALLINGERROR;
}
ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());
updateStats(request, lastOp, lastZxid);
try
{
if (path == null || rsp == null)
{
responseSize = cnxn.sendResponse(hdr, rsp, "response");
}
else
{
int opCode = request.type;
Stat stat = null;
switch (opCode)
{
case OpCode.getData :
{
GetDataResponse getDataResponse = (GetDataResponse) rsp;
stat = getDataResponse.getStat();
responseSize = cnxn.sendResponse(hdr, rsp, "response", path, stat, opCode);
break;
}
case OpCode.getChildren2 :
{
GetChildren2Response getChildren2Response = (GetChildren2Response) rsp;
stat = getChildren2Response.getStat();
responseSize = cnxn.sendResponse(hdr, rsp, "response", path, stat, opCode);
break;
}
default:
responseSize = cnxn.sendResponse(hdr, rsp, "response");
}
}
if (request.type == OpCode.closeSession)
{
cnxn.sendCloseSession();
}
}
catch (IOException e)
{
LOG.error("FIXMSG", e);
}
finally
{
ServerMetrics.getMetrics().RESPONSE_BYTES.add(responseSize);
}
applyRequest
// 對請求進行處理
ProcessTxnResult rc = zks.processTxn(request);
// 關倍訓話
if (request.type == OpCode.closeSession && connClosedByClient(request))
{
if (closeSession(zks.serverCnxnFactory, request.sessionId)
|| closeSession(zks.secureServerCnxnFactory, request.sessionId))
{
return rc;
}
}
// 請求是寫請求
if (request.getHdr() != null)
{
// 統計
long propagationLatency = Time.currentWallTime() - request.getHdr().getTime();
if (propagationLatency >= 0)
{
ServerMetrics.getMetrics().PROPAGATION_LATENCY.add(propagationLatency);
}
}
return rc;
finishSessionInit
try
{
if (valid)
{
if (serverCnxnFactory != null && serverCnxnFactory.cnxns.contains(cnxn))
{
serverCnxnFactory.registerConnection(cnxn);
}
else if (secureServerCnxnFactory != null && secureServerCnxnFactory.cnxns.contains(cnxn))
{
secureServerCnxnFactory.registerConnection(cnxn);
}
}
}
catch (Exception e)
{
LOG.warn("Failed to register with JMX", e);
}
try
{
ConnectResponse rsp = new ConnectResponse(
0,
valid ? cnxn.getSessionTimeout() : 0,
valid ? cnxn.getSessionId() : 0,
valid ? generatePasswd(cnxn.getSessionId()) : new byte[16]);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
bos.writeInt(-1, "len");
rsp.serialize(bos, "connect");
if (!cnxn.isOldClient)
{
bos.writeBool(this instanceof ReadOnlyZooKeeperServer, "readOnly");
}
baos.close();
ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
bb.putInt(bb.remaining() - 4).rewind();
cnxn.sendBuffer(bb);
if (valid)
{
LOG.debug("Established session 0x{} with negotiated timeout {} for client {}",
Long.toHexString(cnxn.getSessionId()), cnxn.getSessionTimeout(),
cnxn.getRemoteSocketAddress());
// 開啟連接上資料接收
cnxn.enableRecv();
}
else
{
LOG.info("Invalid session 0x{} for client {}, probably expired",
Long.toHexString(cnxn.getSessionId()), cnxn.getRemoteSocketAddress());
// 發送會話終止標記
cnxn.sendBuffer(ServerCnxnFactory.closeConn);
}
}
catch (Exception e)
{
LOG.warn("Exception while establishing session, closing", e);
cnxn.close(ServerCnxn.DisconnectReason.IO_EXCEPTION_IN_SESSION_INIT);
}
processTxn
// 獲取請求的事務頭
TxnHeader hdr = request.getHdr();
// 針對創建會話&關倍訓話進行處理
processTxnForSessionEvents(request, hdr, request.getTxn());
// 寫請求
final boolean writeRequest = (hdr != null);
// 集群請求
final boolean quorumRequest = request.isQuorum();
// 不是寫請求
// 且不是集群請求
if (!writeRequest && !quorumRequest)
{
// 直接回傳處理結果
return new ProcessTxnResult();
}
synchronized (outstandingChanges)
{
ProcessTxnResult rc = processTxnInDB(hdr, request.getTxn(), request.getTxnDigest());
if (hdr == null)
{
return new ProcessTxnResult();
}
else
{
return getZKDatabase().processTxn(hdr, txn, digest);
dataTree.processTxn(hdr, txn, digest);
ProcessTxnResult result = processTxn(header, txn);
this.processTxn(header, txn, false);
compareDigest(header, txn, digest);
return result;
}
// 如果是寫請求
if (writeRequest)
{
// 獲取zxid
long zxid = hdr.getZxid();
// 如果 存在處理中ChangeRecord
// 且 首個ChangeRecord的zxid小于這里請求的zxid
while (!outstandingChanges.isEmpty() && outstandingChanges.peek().zxid <= zxid)
{
// 移除
ChangeRecord cr = outstandingChanges.remove();
ServerMetrics.getMetrics().OUTSTANDING_CHANGES_REMOVED.add(1);
if (cr.zxid < zxid)
{
LOG.warn("Zxid outstanding 0x{} is less than current 0x{}",
Long.toHexString(cr.zxid), Long.toHexString(zxid));
}
// 處理路徑的最新狀態資訊
if (outstandingChangesForPath.get(cr.path) == cr)
{
// path的關聯項從outstandingChangesForPath移除
// 因為path的關聯項已經持久化到資料物體物件中了
outstandingChangesForPath.remove(cr.path);
}
}
}
// 集群請求
if (quorumRequest)
{
getZKDatabase().addCommittedProposal(request);
}
return rc;
}
addCommittedProposal
對最近一定數量的已經完成最后處理的請求物件，
構造關聯提議物件，
將其存盤起來，以備需要時使用
// 寫鎖
WriteLock wl = logLock.writeLock();
try
{
// 鎖定
wl.lock();
// 容器中提議物件數量超過commitLogCount
if (committedLog.size() > commitLogCount)
{
// 從容器移除一個元素
committedLog.remove();
// 獲取首個提議物件對應集群包中的zxid---容器中所有提議中最小的zxid
minCommittedLog = committedLog.peek().packet.getZxid();
}
// committedLog變空了
if (committedLog.isEmpty())
{
// 容器中所有提議中最小的zxid，最大的zxid此時均為處理掉請求的zxid
minCommittedLog = request.zxid;
maxCommittedLog = request.zxid;
}
// 請求物件序列化
byte[] data = SerializeUtils.serializeRequest(request);
// 構造集群包
// 包型別Leader.PROPOSAL
// 包zxid為處理掉物件的zxid
// 包資料為物件序列化所得
QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
// 動態構造新的提議物件
Proposal p = new Proposal();
p.packet = pp;
p.request = request;
// 將提議物件加入committedLog
committedLog.add(p);
// 更新committedLog中所有提議的最大的zxid
maxCommittedLog = p.packet.getZxid();
}
finally
{
wl.unlock();
}
processTxnForSessionEvents(request, hdr, request.getTxn());
// 請求型別
int opCode = (request == null) ? hdr.getType() : request.type;
// 客戶端id
long sessionId = (request == null) ? hdr.getClientId() : request.sessionId;
// 創建會話
if (opCode == OpCode.createSession)
{
// 事務頭非空
// 且 事務體為CreateSessionTxn
if (hdr != null && txn instanceof CreateSessionTxn)
{
CreateSessionTxn cst = (CreateSessionTxn) txn;
// 向會話追蹤提交 會話id--超時設定
sessionTracker.commitSession(sessionId, cst.getTimeOut());
}
// 請求為空
// 且 請求不是本地會話
else if (request == null || !request.isLocalSession())
{
LOG.warn("*****>>>>> Got {} {}", txn.getClass(), txn.toString());
}
}
// 關倍訓話
else if (opCode == OpCode.closeSession)
{
// 向會話追蹤移除會話
sessionTracker.removeSession(sessionId);
}
集群請求判斷
public boolean isQuorum()
{
switch (this.type)
{
// 讀取類
case OpCode.exists:
case OpCode.getACL:
case OpCode.getChildren:
case OpCode.getAllChildrenNumber:
case OpCode.getChildren2:
case OpCode.getData:
case OpCode.getEphemerals:
case OpCode.multiRead:
case OpCode.whoAmI:
return false;
// 寫入類
case OpCode.create:
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer:
case OpCode.error:
case OpCode.delete:
case OpCode.deleteContainer:
case OpCode.setACL:
case OpCode.setData:
case OpCode.check:
case OpCode.multi:
case OpCode.reconfig:
return true;
// 會話創建，管理類在請求非本地會話時，屬于集群
case OpCode.closeSession:
case OpCode.createSession:
return !this.isLocalSession;
default:
return false;
}
}
processTxn
利用事務頭，事務體，實際處理寫請求
// 動態構造ProcessTxnResult物件
ProcessTxnResult rc = new ProcessTxnResult();
try
{
// 事務頭包含客戶端id
rc.clientId = header.getClientId();
// 事務頭包含cxid
rc.cxid = header.getCxid();
// 事務頭包含zxid
rc.zxid = header.getZxid();
// 事務頭包含類別
rc.type = header.getType();
// 事務頭包含錯誤碼
rc.err = 0;
// 事務頭包含multiResult
rc.multiResult = null;
switch (header.getType())
{
// 創建節點
case OpCode.create:
// 事務體
CreateTxn createTxn = (CreateTxn) txn;
// 節點路徑
rc.path = createTxn.getPath();
createNode(
// 路徑
createTxn.getPath(),
// 資料
createTxn.getData(),
// 權限
createTxn.getAcl(),
// 臨時節點包含所屬客戶端id
createTxn.getEphemeral() ? header.getClientId() : 0,
// 父節點累計改變版本
createTxn.getParentCVersion(),
// zxid
header.getZxid(),
// 時間
header.getTime(),
null);
break;
case OpCode.create2:
CreateTxn create2Txn = (CreateTxn) txn;
rc.path = create2Txn.getPath();
Stat stat = new Stat();
createNode(
create2Txn.getPath(),
create2Txn.getData(),
create2Txn.getAcl(),
create2Txn.getEphemeral() ? header.getClientId() : 0,
create2Txn.getParentCVersion(),
header.getZxid(),
header.getTime(), stat);
// 獲取節點狀態
rc.stat = stat;
break;
case OpCode.createTTL:
CreateTTLTxn createTtlTxn = (CreateTTLTxn) txn;
rc.path = createTtlTxn.getPath();
stat = new Stat();
createNode(
createTtlTxn.getPath(),
createTtlTxn.getData(),
createTtlTxn.getAcl(),
EphemeralType.TTL.toEphemeralOwner(createTtlTxn.getTtl()),
createTtlTxn.getParentCVersion(),
header.getZxid(),
header.getTime(),
stat);
// 獲取節點狀態
rc.stat = stat;
break;
case OpCode.createContainer:
CreateContainerTxn createContainerTxn = (CreateContainerTxn) txn;
rc.path = createContainerTxn.getPath();
stat = new Stat();
createNode(
createContainerTxn.getPath(),
createContainerTxn.getData(),
createContainerTxn.getAcl(),
EphemeralType.CONTAINER_EPHEMERAL_OWNER,
createContainerTxn.getParentCVersion(),
header.getZxid(),
header.getTime(),
stat);
rc.stat = stat;
break;
case OpCode.delete:
case OpCode.deleteContainer:
DeleteTxn deleteTxn = (DeleteTxn) txn;
rc.path = deleteTxn.getPath();
deleteNode(deleteTxn.getPath(), header.getZxid());
break;
case OpCode.reconfig:
case OpCode.setData:
SetDataTxn setDataTxn = (SetDataTxn) txn;
rc.path = setDataTxn.getPath();
rc.stat = setData(
setDataTxn.getPath(),
setDataTxn.getData(),
setDataTxn.getVersion(),
header.getZxid(),
header.getTime());
break;
case OpCode.setACL:
SetACLTxn setACLTxn = (SetACLTxn) txn;
rc.path = setACLTxn.getPath();
rc.stat = setACL(
setACLTxn.getPath(),
setACLTxn.getAcl(),
setACLTxn.getVersion());
break;
case OpCode.closeSession:
long sessionId = header.getClientId();
if (txn != null)
{
killSession(
sessionId,
header.getZxid(),
ephemerals.remove(sessionId),
((CloseSessionTxn) txn).getPaths2Delete());
}
else
{
killSession(sessionId, header.getZxid());
}
break;
case OpCode.error:
ErrorTxn errTxn = (ErrorTxn) txn;
// 錯誤碼
rc.err = errTxn.getErr();
break;
case OpCode.check:
// 檢查
CheckVersionTxn checkTxn = (CheckVersionTxn) txn;
rc.path = checkTxn.getPath();
break;
case OpCode.multi:
// 復合操作
// 復合操作的事務體
MultiTxn multiTxn = (MultiTxn) txn;
// 取出復合的多個Txn
List<Txn> txns = multiTxn.getTxns();
// 動態構造用于存盤多個處理結果的容器
rc.multiResult = new ArrayList<ProcessTxnResult>();
boolean failed = false;
// 依次處理每個子事務
for (Txn subtxn : txns)
{
// 一旦有一個子事務之前已經判定為失敗
if (subtxn.getType() == OpCode.error)
{
// 整體失敗，后續不處理
failed = true;
break;
}
}
boolean post_failed = false;
// 對每個子事務依次處理
for (Txn subtxn : txns)
{
// 位元組流
ByteBuffer bb = ByteBuffer.wrap(subtxn.getData());
Record record = null;
// 子事務型別
switch (subtxn.getType())
{
// 創建
case OpCode.create:
// 動態構造
record = new CreateTxn();
break;
case OpCode.createTTL:
record = new CreateTTLTxn();
break;
case OpCode.createContainer:
record = new CreateContainerTxn();
break;
case OpCode.delete:
case OpCode.deleteContainer:
record = new DeleteTxn();
break;
case OpCode.setData:
record = new SetDataTxn();
break;
case OpCode.error:
record = new ErrorTxn();
post_failed = true;
break;
case OpCode.check:
record = new CheckVersionTxn();
break;
default:
throw new IOException("Invalid type of op: " + subtxn.getType());
}
assert (record != null);
// 反向序列化得到請求物體
ByteBufferInputStream.byteBuffer2Record(bb, record);
// 如果某個子事務判斷為失敗
// 且 當前子事務非失敗
if (failed && subtxn.getType() != OpCode.error)
{
// 對此子事務，OK
int ec = post_failed ? Code.RUNTIMEINCONSISTENCY.intValue() : Code.OK.intValue();
// 設定子事務型別為失敗
subtxn.setType(OpCode.error);
// 動態構造ErrorTxn
record = new ErrorTxn(ec);
}
assert !failed || (subtxn.getType() == OpCode.error);
// 動態構造事務頭
TxnHeader subHdr = new TxnHeader(
// 客戶端id
header.getClientId(),
// cxid
header.getCxid(),
// zxid
header.getZxid(),
// 時間
header.getTime(),
// 型別
subtxn.getType());
// 對子事務進行處理---若存在某個子事務為失敗，所有其他子事務的事務體會設定為ErrorTxn
ProcessTxnResult subRc = processTxn(subHdr, record, true);
// 加入處理結果
rc.multiResult.add(subRc);
// 子事務結果為失敗
// 整體結果不為失敗
if (subRc.err != 0 && rc.err == 0)
{
// 設定整體結果為失敗的那個子事務的結果
rc.err = subRc.err;
}
}
break;
}
}
catch (KeeperException e)
{
LOG.debug("Failed: {}:{}", header, txn, e);
rc.err = e.code().intValue();
}
catch (IOException e)
{
LOG.debug("Failed: {}:{}", header, txn, e);
}
// 型別為創建
// 失敗原因為節點不存在
if (header.getType() == OpCode.create && rc.err == Code.NODEEXISTS.intValue())
{
LOG.debug("Adjusting parent cversion for Txn: {} path: {} err: {}", header.getType(), rc.path, rc.err);
int lastSlash = rc.path.lastIndexOf('/');
String parentName = rc.path.substring(0, lastSlash);
CreateTxn cTxn = (CreateTxn) txn;
try
{
setCversionPzxid(parentName, cTxn.getParentCVersion(), header.getZxid());
}
catch (KeeperException.NoNodeException e)
{
LOG.error("Failed to set parent cversion for: {}", parentName, e);
rc.err = e.code().intValue();
}
}
else if (rc.err != Code.OK.intValue())
{
LOG.debug("Ignoring processTxn failure hdr: {} : error: {}", header.getType(), rc.err);
}
if (!isSubTxn)
{
if (rc.zxid > lastProcessedZxid)
{
lastProcessedZxid = rc.zxid;
}
if (digestFromLoadedSnapshot != null)
{
compareSnapshotDigests(rc.zxid);
}
else
{
logZxidDigest(rc.zxid, getTreeDigest());
}
}
return rc;
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/420015.html
標籤：其他
上一篇：Scala編程語言入門（1）