ZooKeeper如何保证事务原子性?

并发笔记

发布于 2022-11-21 12:16:32

1K00

代码可运行

文章被收录于专栏：并发笔记并发笔记

运行总次数：0

代码可运行

先解答疑惑，题主对ZAB理解是正确的。为了便于描述，本文将事务理解为具有ACID的一组操作，一个ZooKeeper请求（例如：create）称之为提案。

ZAB协议是共识算法的一种，共识算法仅能保证单个提案在集群中达成共识，如果是多个提案要保证事务的话，需要在上层再做一次封装。ZAB被称为原子广播协议，也是做了这一层封装，即：multi命令。

multi命令让多个提案，要么同时成功，要么同时失败，所以要知道ZooKeeper怎么处理事务的，只需要关注multi命令的实现即可。

ZooKeeper对提案的协商，是以责任链的形式处理，下图是协商提案的责任链路，大家可以参考。

不难发现，客户端的请求，先到达PrepRequestProcessor，那么在PrepRequestProcessor一定可以找到对multi命令的特殊操作。

以下代码为PrepRequestProcessor#pRequestHelper，我省略掉了try-catch和其他无关代码，在处理multi请求时，ZooKeeper会先遍历multiRequest，把每个元素当做一个单独的提案调用pRequest2Txn()方法来协商，当某个提案协商发生异常时，ZooKeeper会调用rollbackPendingChanges()回滚正在执行中的提案。

// PrepRequestProcessor
private void pRequestHelper(Request request) throws RequestProcessorException {
	switch (request.type) {
	case 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>();
		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);

					/* Rollback change records from failed multi-op */
					rollbackPendingChanges(zxid, pendingChanges);
				}
			}

			try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
				BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
				txn.serialize(boa, "request");
				ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
				txns.add(new Txn(type, bb.array()));
			}
		}
		request.setTxn(new MultiTxn(txns));
		if (digestEnabled) {
			setTxnDigest(request);
		}
		break;
}