zookeeper原始碼決議--請求處理--PrepRequestProcessor-有解無憂

PrepRequestProcessor

1.從待處理請求佇列取出一個請求
2.如果是requestOfDeath
	退出
3.置空請求物件的事務頭，事務體，
4.請求未被過濾下，執行pRequestHelper
5.將請求交給下一級處理

pRequestHelper

對寫請求設定
對請求進行合法性檢查
按請求內容預先更新節點狀態資訊
為請求物件設定事務頭，事務體，事務摘要

對請求處理中遭遇錯誤的，采用拋出例外方式
對錯誤的寫請求，其事務頭，事務體包含錯誤及錯誤資訊，
對錯誤的請求，請求本身包含拋出例外資訊，

如果請求型別為OpCode.createContainer/OpCode.create/OpCode.create2
動態創建CreateRequest物件
//基于原始請求物件，
//對請求進行合法性檢查，
//預先對按請求進行節點更新&資料訪問，
//設定好請求物件的事務頭，事務體，事務摘要
pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
如果請求型別為OpCode.createTTL
	CreateTTLRequest createTtlRequest = new CreateTTLRequest();
    pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest, true);
如果請求型別為OpCode.deleteContainer/OpCode.delete
    DeleteRequest deleteRequest = new DeleteRequest();
    pRequest2Txn(request.type, zks.getNextZxid(), request, deleteRequest, true);
如果請求型別為OpCode.setData
    SetDataRequest setDataRequest = new SetDataRequest();
    pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest, true);
如果請求型別為OpCode.reconfig
    ReconfigRequest reconfigRequest = new ReconfigRequest();
    ByteBufferInputStream.byteBuffer2Record(request.request, reconfigRequest);
    pRequest2Txn(request.type, zks.getNextZxid(), request, reconfigRequest, true);
如果請求型別為OpCode.setACL
    SetACLRequest setAclRequest = new SetACLRequest();
    pRequest2Txn(request.type, zks.getNextZxid(), request, setAclRequest, true);
如果請求型別為OpCode.check
    CheckVersionRequest checkRequest = new CheckVersionRequest();
    pRequest2Txn(request.type, zks.getNextZxid(), request, checkRequest, true);
如果請求型別為OpCode.multi
    // 動態構造多操作請求，請求物體
    MultiOperationRecord multiRequest = new MultiOperationRecord();
    // 反向序列化得到資料物體
    try 
    {
              ByteBufferInputStream.byteBuffer2Record(request.request, multiRequest);
    } 
    catch (IOException e) 
    {
          	// 拋出例外時，仍然設定事務頭
              request.setHdr(
              	new TxnHeader(
            			request.sessionId, request.cxid, 
            			zks.getNextZxid(), Time.currentWallTime(), OpCode.multi));
              // 再次拋出
              throw e;
    }
          
    // 事務容器
    List<Txn> txns = new ArrayList<Txn>();
    // 獲取下一zxid
    long zxid = zks.getNextZxid();
    KeeperException ke = null;
    // 獲取每個子操作&最新狀態資訊，子操作父節點&最新狀態資訊
    Map<String, ChangeRecord> pendingChanges = getPendingChanges(multiRequest);
    // 設定請求的事務頭
    request.setHdr(
          	new TxnHeader(request.sessionId, request.cxid, zxid, Time.currentWallTime(), request.type));
    // 對每個子操作逐個處理
    for (Op op : multiRequest) 
    {
          	// 子操作的，請求物體
              Record subrequest = op.toRequestRecord();
              int type;
              Record txn;
              // 表示之前某個子操作遭遇例外
              if (ke != null) 
              {
              	 // 設定類別
                  type = OpCode.error;
                  // 設定事務體
                  txn = new ErrorTxn(Code.RUNTIMEINCONSISTENCY.intValue());
              } 
              else 
              {
                  try 
                  {
                  	//基于原始請求物件，
					//對請求進行合法性檢查，
					//預先對按請求進行節點更新&資料訪問，
					//設定好請求物件的事務頭，事務體，事務摘要
                    pRequest2Txn(op.getType(), zxid, request, subrequest, false);
                    // 子操作類別
                    type = op.getType();
                    // 獲得事務體
                    txn = request.getTxn();
                  } 
                  // 如果子操作在于例外
                  catch (KeeperException e) 
                  {
                  	// 例外記錄
                    ke = e;
                    // 類別記錄
                    type = OpCode.error;
                    // 用于記錄失敗的事務體
                    txn = new ErrorTxn(e.code().intValue());
                    if (e.code().intValue() > Code.APIERROR.intValue()) 
                    {
                          LOG.info(
                          	"Got user-level KeeperException when processing {} aborting" 
                          	+ " remaining multi ops. Error Path:{} Error:{}", 
                          	+ request.toString(), e.getPath(), e.getMessage());
                    }
			
			 		// 將例外資訊設定到請求物件
                    request.setException(e);
                    // 一旦一個子操作遭遇例外，執行回滾
                    // 回滾需要保證復合操作每個子操作對ChangeRecord的影響撤銷，
                    // 把ChangeRecord恢復到執行復合操作前
                    rollbackPendingChanges(zxid, pendingChanges);
                  }
              }

              try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) 
              {
                  BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
                  // 事務體物件序列化
                  txn.serialize(boa, "request");
                  ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
                  // 存盤到Txn物件集合
                  txns.add(new Txn(type, bb.array()));
              }
          }

 		  // 復合請求的事務體，也包含多個Txn分別對應每個子操作
          request.setTxn(new MultiTxn(txns));
          if (digestEnabled) 
          {
          	 // 設定事務摘要 
              setTxnDigest(request);
          }
 	}
如果請求型別為OpCode.createSession/OpCode.closeSession
   	// 本地會話不需追蹤？
    if (!request.isLocalSession()) 
    {
              pRequest2Txn(request.type, zks.getNextZxid(), request, null, true);
    }
如果請求型別為OpCode.sync/OpCode.exists/OpCode.getData/OpCode.getACL
    OpCode.getChildren/OpCode.getAllChildrenNumber/OpCode.getChildren2/OpCode.ping
    OpCode.setWatches/OpCode.setWatches2/OpCode.checkWatches/OpCode.removeWatches
    OpCode.getEphemerals/OpCode.multiRead/OpCode.addWatch/OpCode.addWatch
  	// 會話檢查
  	// 只讀請求不需要
  	zks.sessionTracker.checkSession(request.sessionId, request.getOwner());

pRequest2TxnCreate(type, request, record, deserialize);

// 做新建節點前的合法性，權限，配額檢查
// 一切合法下，更新父節點，新建節點狀態資訊，
// 設定好本請求物件的事務頭，事務體，事務摘要資訊

if (deserialize) 
{
	// 從位元組流得到資料物件
	ByteBufferInputStream.byteBuffer2Record(request.request, record);
}

// 從資料物件提取資訊
if (type == OpCode.createTTL) 
{
    CreateTTLRequest createTtlRequest = (CreateTTLRequest) record;
    flags = createTtlRequest.getFlags();
    path = createTtlRequest.getPath();
    acl = createTtlRequest.getAcl();
    data = createTtlRequest.getData();
    ttl = createTtlRequest.getTtl();
} 
else 
{
    CreateRequest createRequest = (CreateRequest) record;
    flags = createRequest.getFlags();
    path = createRequest.getPath();
    acl = createRequest.getAcl();
    data = createRequest.getData();
    ttl = -1;
}

CreateMode createMode = CreateMode.fromFlag(flags);
// 合法性驗證，不合法通過拋出特定型別包含特定資訊例外來層層回退
validateCreateRequest(path, createMode, request, ttl);
// 父路徑
String parentPath = validatePathForCreate(path, request.sessionId);
// 權限
List<ACL> listACL = fixupACL(path, request.authInfo, acl);
// 父親路徑最新狀態
ChangeRecord parentRecord = getRecordForPath(parentPath);
// 權限檢查
zks.checkACL(
	request.cnxn, parentRecord.acl, 
	ZooDefs.Perms.CREATE, request.authInfo, path, listACL);
// 父親路徑下遭遇改變次數
int parentCVersion = parentRecord.stat.getCversion();
// 新節點完整路徑
if (createMode.isSequential()) 
{
    path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
}

// 驗證路徑合法性
validatePath(path, request.sessionId);
try 
{
	// 如果要創建的路徑已經存在了
    if (getRecordForPath(path) != null) 
    {
    	// 拋出特定型別，包含特定資訊的例外
        throw new KeeperException.NodeExistsException(path);
    }
} 
catch (KeeperException.NoNodeException e) 
{
}
   
boolean ephemeralParent = EphemeralType.get(
	parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
// 這類父路徑不可包含子節點 
if (ephemeralParent) 
{
    throw new KeeperException.NoChildrenForEphemeralsException(path);
}
   
// 更新父節點遭遇變化次數
int newCversion = parentRecord.stat.getCversion() + 1;
// 檢查配額限制，配額限制超出時，通過拋出特定型別，包含特定資訊的例外來層層退出
zks.checkQuota(path, null, data, OpCode.create);
// 設定事務體
if (type == OpCode.createContainer) 
{
    request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
} 
else if (type == OpCode.createTTL) 
{
    request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
} 
else 
{
    request.setTxn(
    	new CreateTxn(path, data, listACL, createMode.isEphemeral(), newCversion));
}

TxnHeader hdr = request.getHdr();
long ephemeralOwner = 0;
// 創建模式--創建容器節點
if (createMode.isContainer()) 
{
      ephemeralOwner = EphemeralType.CONTAINER_EPHEMERAL_OWNER;
} 
// 創建模式--創建自帶生命期節點
else if (createMode.isTTL()) 
{
      ephemeralOwner = EphemeralType.TTL.toEphemeralOwner(ttl);
} 
// 創建模式--創建臨時節點【會話關閉，會話所屬臨時節點自動銷毀】
else if (createMode.isEphemeral()) 
{
      ephemeralOwner = request.sessionId;
}

// 創建節點狀態
StatPersisted s 
	= DataTree.createStat(hdr.getZxid(), hdr.getTime(), ephemeralOwner);
// 基于原有父節點的parentRecord得到新的
parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
parentRecord.childCount++;
// 記錄了此節點下孩子累計新增孩子數量-變化為節點孩子變化
parentRecord.stat.setCversion(newCversion);
// 記錄了最后修改此節點狀態的事務id
parentRecord.stat.setPzxid(request.getHdr().getZxid());
// 更新整體和節點自身摘要資訊
parentRecord.precalculatedDigest 
	= precalculateDigest(DigestOpCode.UPDATE, parentPath, parentRecord.data, parentRecord.stat);
addChangeRecord(parentRecord);
// 一個節點可能包含多個關聯項
zks.outstandingChanges.add(c);
// 一個節點，只有一個路徑，也就只記錄一個關聯項，反映此節點最新狀態
zks.outstandingChangesForPath.put(c.path, c);
// 反映新建節點
ChangeRecord nodeRecord 
	= new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL);
nodeRecord.data = data;
nodeRecord.precalculatedDigest = precalculateDigest(DigestOpCode.ADD, path, nodeRecord.data, s);
setTxnDigest(request, nodeRecord.precalculatedDigest);
request.setTxnDigest(new TxnDigest(digestCalculator.getDigestVersion(), preCalculatedDigest.treeDigest));
addChangeRecord(nodeRecord);

pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);

基于原始請求物件，
對請求進行合法性檢查，
預先對按請求進行節點更新&資料訪問，
設定好請求物件的事務頭，事務體，事務摘要

如果請求物件事務頭為null
	request.setHdr(
		new TxnHeader(request.sessionId, request.cxid, zxid, Time.currentWallTime(), type));
	如果類別為OpCode.create/OpCode.create2/OpCode.createTTL/OpCode.createContainer
		// 做新建節點前的合法性，權限，配額檢查
		// 一切合法下，更新父節點，新建節點狀態資訊，
		// 設定好本請求物件的事務頭，事務體，事務摘要資訊
		pRequest2TxnCreate(type, request, record, deserialize);
	如果類別為OpCode.deleteContainer
		String path = new String(request.request.array(), UTF_8);
        // 獲取父路徑，及驗證路徑
        String parentPath = getParentPathAndValidate(path);
        // 獲取洗掉路徑最新資訊
        ChangeRecord nodeRecord = getRecordForPath(path);
        // 如果洗掉節點包含子節點，不可洗掉
        if (nodeRecord.childCount > 0) 
        {
            throw new KeeperException.NotEmptyException(path);
        }
        
        // 要洗掉節點類別非法
        if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner()) 
        	== EphemeralType.NORMAL) 
        {
            throw new KeeperException.BadVersionException(path);
        }
        
        // 獲取父節點最新狀態
        ChangeRecord parentRecord = getRecordForPath(parentPath);
        // 設定改請求的事務體
        request.setTxn(new DeleteTxn(path));
        // 基于parentRecord得到新的
        parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
        parentRecord.childCount--;
        parentRecord.stat.setPzxid(request.getHdr().getZxid());
        parentRecord.precalculatedDigest 
        	= precalculateDigest(DigestOpCode.UPDATE, 
        		parentPath, parentRecord.data, parentRecord.stat);
        // 完成對父節點最新狀態的更新
        addChangeRecord(parentRecord);
        // 對洗掉節點，設定改節點ChangeRecord
        // 通過改ChangeRecord可辨別節點已被洗掉
        nodeRecord = new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null);
        // 更新摘要資訊
        nodeRecord.precalculatedDigest = precalculateDigest(DigestOpCode.REMOVE, path);
        // 設定事務摘要
        setTxnDigest(request, nodeRecord.precalculatedDigest);
        // 完成對洗掉節點的狀態更新
        addChangeRecord(nodeRecord);
   	如果類別為OpCode.delete
   		// 會話檢查
   		zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
        DeleteRequest deleteRequest = (DeleteRequest) record;
        // 反向序列化得到資料物體
        if (deserialize) 
        {
            ByteBufferInputStream.byteBuffer2Record(request.request, deleteRequest);
        }
        
        // 要洗掉路徑
        String path = deleteRequest.getPath();
        // 要洗掉路徑父路徑
        String parentPath = getParentPathAndValidate(path);
        // 父路徑最新狀態
        ChangeRecord parentRecord = getRecordForPath(parentPath);
        // 權限檢查
        zks.checkACL(
        	request.cnxn, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo, path, null);
        // 要洗掉路徑自身
        // 路徑不存在時，通過設定例外型別，例外資訊，層層退出
        ChangeRecord nodeRecord = getRecordForPath(path);
        // 版本檢查
        checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path);
        // 有孩子的節點不可被洗掉
        if (nodeRecord.childCount > 0) 
        {
            throw new KeeperException.NotEmptyException(path);
        }
        // 設定事務體
        request.setTxn(new DeleteTxn(path));
        // 基于parentRecord得到新的
        parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
        // 更新父節點最新狀態資訊
        // 狀態--孩子數量
        parentRecord.childCount--;
        // 狀態--最后對其修改的事務的id
        parentRecord.stat.setPzxid(request.getHdr().getZxid());
        // 更新摘要
        parentRecord.precalculatedDigest 
        	= precalculateDigest(DigestOpCode.UPDATE, 
        	parentPath, parentRecord.data, parentRecord.stat);
        addChangeRecord(parentRecord);
        // 為要洗掉節點，動態構造ChangeRecord
        nodeRecord = new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null);
        // 更新摘要
        nodeRecord.precalculatedDigest = precalculateDigest(DigestOpCode.REMOVE, path);
        // 設定事務摘要
        setTxnDigest(request, nodeRecord.precalculatedDigest);
        addChangeRecord(nodeRecord);
 	如果類別為OpCode.setData
 		// 檢查會話
 		zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
        // 事務體
        SetDataRequest setDataRequest = (SetDataRequest) record;
        // 反向序列化得到資料物件
        if (deserialize) 
        {
            ByteBufferInputStream.byteBuffer2Record(request.request, setDataRequest);
        }
        // 路徑
        path = setDataRequest.getPath();
        // 驗證路徑合法性
        validatePath(path, request.sessionId);
        // 獲取此路徑最新資訊
        nodeRecord = getRecordForPath(path);
        // 權限檢查
        zks.checkACL(
        	request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, path, null);
        // 配額檢查
        zks.checkQuota(path, nodeRecord.data, setDataRequest.getData(), OpCode.setData);
        // 版本檢查
        int newVersion = 
        	checkAndIncVersion(nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);
        // 設定事務體
        request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
        // 基于nodeRecode得到新的
        nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
        // 新的相比舊的，版本變了
        nodeRecord.stat.setVersion(newVersion);
        // 修改時間變了
        nodeRecord.stat.setMtime(request.getHdr().getTime());
        // 最后修改的事務id變了
        nodeRecord.stat.setMzxid(zxid);
        // 節點資料內容變了
        nodeRecord.data = setDataRequest.getData();
        // 更新摘要
        nodeRecord.precalculatedDigest 
        	= precalculateDigest(DigestOpCode.UPDATE, path, nodeRecord.data, nodeRecord.stat);
        // 設定事務摘要
        setTxnDigest(request, nodeRecord.precalculatedDigest);
        addChangeRecord(nodeRecord);
    如果類別為OpCode.setACL
    	// 檢查會話
    	zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
        // 事務體
        SetACLRequest setAclRequest = (SetACLRequest) record;
        // 反向序列化得到資料物件
        if (deserialize) 
        {
            ByteBufferInputStream.byteBuffer2Record(request.request, setAclRequest);
        }
        // 路徑
        path = setAclRequest.getPath();
        // 驗證路徑
        validatePath(path, request.sessionId);
        // 權限串列
        List<ACL> listACL = fixupACL(path, request.authInfo, setAclRequest.getAcl());
        // 路徑最新狀態
        nodeRecord = getRecordForPath(path);
        // 權限檢查
        zks.checkACL(
        	request.cnxn, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo, path, listACL);
        // 版本檢查
        newVersion 
        	= checkAndIncVersion(nodeRecord.stat.getAversion(), setAclRequest.getVersion(), path);
        // 設定事務體
        request.setTxn(new SetACLTxn(path, listACL, newVersion));
        // 基于nodeRecord得到新的
        nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
        // 設定新的版本
        nodeRecord.stat.setAversion(newVersion);
        // 更新摘要資訊
        nodeRecord.precalculatedDigest 
        	= precalculateDigest(DigestOpCode.UPDATE, path, nodeRecord.data, nodeRecord.stat);
        // 設定事務摘要
        setTxnDigest(request, nodeRecord.precalculatedDigest);
        addChangeRecord(nodeRecord);
   	如果型別為OpCode.createSession
   		request.request.rewind();
   		// 獲取超時時間？
        int to = request.request.getInt();
        // 設定事務體
        request.setTxn(new CreateSessionTxn(to));
        request.request.rewind();
        // 會話追蹤
        zks.sessionTracker.trackSession(request.sessionId, to);
        // 會話擁有者，為請求擁有者
        zks.setOwner(request.sessionId, request.getOwner());
    如果型別為OpCode.closeSession
    	// 當前時間
    	long startTime = Time.currentElapsedTime();
        synchronized (zks.outstandingChanges) 
        {
        	// 獲取屬于此會話的所有臨時節點
            Set<String> es = zks.getZKDatabase().getEphemerals(request.sessionId);
            // 對所有排隊中ChangeRecord
            for (ChangeRecord c : zks.outstandingChanges) 
            {
            	// 表示待實施請求中，已經包含洗掉某路徑請求
                if (c.stat == null) 
                {
                	// 將此會話下已經規劃為洗掉的路徑從這里的臨時節點中去掉，避免重復洗掉
                    es.remove(c.path);
                } 
                // 表示待實施請求中，包含某個路徑屬于此臨時會話
                else if (c.stat.getEphemeralOwner() == request.sessionId) 
                {
                	// 將改路徑加入這里的臨時節點，放置漏掉屬于此會話的待實施的臨時節點，
                    es.add(c.path);
                }
            }
			
			// 對此會話的每個臨時節點，對其父節點進行更新
            for (String path2Delete : es) 
            {
                if (digestEnabled) 
                {
                	// 獲取洗掉節點，父節點&驗證
                    parentPath = getParentPathAndValidate(path2Delete);
                    // 獲取父節點最新狀態
                    parentRecord = getRecordForPath(parentPath);
                    // 基于parentRecord得到新的
                    parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
                    // 更新修改父節點孩子的事務id
                    parentRecord.stat.setPzxid(request.getHdr().getZxid());
                    // 更新摘要
                    parentRecord.precalculatedDigest 
                    	= precalculateDigest(DigestOpCode.UPDATE, 
                    		parentPath, parentRecord.data, parentRecord.stat);
                    addChangeRecord(parentRecord);
                }
                
                // 動態構造要洗掉節點的nodeRecord
                nodeRecord = 
                	new ChangeRecord(request.getHdr().getZxid(), path2Delete, null, 0, null);
                // 更新摘要
                nodeRecord.precalculatedDigest 
                	= precalculateDigest(DigestOpCode.REMOVE, path2Delete);
                addChangeRecord(nodeRecord);
            }

            if (ZooKeeperServer.isCloseSessionTxnEnabled()) 
            {
            	// 設定事務體
                request.setTxn(new CloseSessionTxn(new ArrayList<String>(es)));
            }
            
            // 設定會話狀態為關閉中
            zks.sessionTracker.setSessionClosing(request.sessionId);
        }

        ServerMetrics.getMetrics()
        	.CLOSE_SESSION_PREP_TIME.add(Time.currentElapsedTime() - startTime);
	如果型別為OpCode.check
		// 會話檢查
		zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
        CheckVersionRequest checkVersionRequest = (CheckVersionRequest) record;
        // 反向序列化得到資料物體
        if (deserialize) 
        {
            ByteBufferInputStream.byteBuffer2Record(request.request, checkVersionRequest);
        }
        // 獲取路徑
        path = checkVersionRequest.getPath();
        // 驗證路徑
        validatePath(path, request.sessionId);
        // 獲取路徑最新資訊
        nodeRecord = getRecordForPath(path);
        // 權限檢查
        zks.checkACL(
        	request.cnxn, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo, path, null);
        // 設定事務體
        request.setTxn(
        	new CheckVersionTxn(path, 
        		checkAndIncVersion(nodeRecord.stat.getVersion(), 
        		checkVersionRequest.getVersion(), null));

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/419127.html

標籤：其他

上一篇：TiDB 在國信證券海量資料高并發場景中的實踐

下一篇：基于Flink實時數倉——DWS 層-地區主題表（8）