SOFAJRaft源码之StateMachine(十五)

public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
	// currVal为业务数据来着的
	final long currVal = this.value.get();
	
	executor.submit(() -> {
		final CounterSnapshotFile snapshot = new CounterSnapshotFile(writer.getPath() + File.separator + "data");
		// *************************************************************************
		// 自己实现数据的写出(CounterSnapshotFile)
		// *************************************************************************
		if (snapshot.save(currVal)) {
			
			// *************************************************************************
			// 添加元数据信息,在前面剖析过源码来着,这里不详解了.
			// 我现在感觉元数据仅仅只是一个标记而已,先不管了,后再剖析KV的时候再看它的实现是咋回事
			// *************************************************************************
			if (writer.addFile("data")) {
				done.run(Status.OK());
			} else {
				done.run(new Status(RaftError.EIO, "Fail to add file to writer"));
			}
		} else {
			done.run(new Status(RaftError.EIO, "Fail to save counter snapshot %s", snapshot.getPath()));
		}
	});
}

public boolean save(final long value) {
	try {
		// 将数据转换成字符串,写出到指定的目录
		FileUtils.writeStringToFile(new File(path), String.valueOf(value));
		return true;
	} catch (IOException e) {
		LOG.error("Fail to save snapshot", e);
		return false;
	}
}

public boolean onSnapshotLoad(final SnapshotReader reader) {
	if (isLeader()) { // leader 不允许加载快照
		LOG.warn("Leader is not supposed to load snapshot");
		return false;
	}
	
	// 验证下元数据是否存在
	if (reader.getFileMeta("data") == null) {
		LOG.error("Fail to find data file in {}", reader.getPath());
		return false;
	}
	

	final CounterSnapshotFile snapshot = new CounterSnapshotFile(reader.getPath() + File.separator + "data");
	try {
		// ********************************************************************************************
		//  委托给:CounterSnapshotFile加载数据
		// ********************************************************************************************
		this.value.set(snapshot.load());
		return true;
	} catch (final IOException e) {
		LOG.error("Fail to load snapshot from {}", snapshot.getPath());
		return false;
	}

}

public long load() throws IOException {
	// 从磁盘加载数据,转换成long类型
	final String s = FileUtils.readFileToString(new File(path));
	if (!StringUtils.isBlank(s)) {
		return Long.parseLong(s);
	}
	throw new IOException("Fail to load snapshot from " + path + ",content: " + s);
}

// ***************************************************************************
// onApply是由FSMCaller.onCommitted触发回调的
// ***************************************************************************

public void onApply(final Iterator iter) {
	while (iter.hasNext()) {
		long current = 0;
		CounterOperation counterOperation = null;

		CounterClosure closure = null;

		// 从Iterator里获得数据.
		if (iter.done() != null) {
			// 从Closure中获取数据
			// This task is applied by this node, get value from closure to avoid additional parsing.
			closure = (CounterClosure) iter.done();
			counterOperation = closure.getCounterOperation();
		} else {
			// Have to parse FetchAddRequest from this user log.
			final ByteBuffer data = iter.getData();
			try {
				// 反序列化操作
				counterOperation = SerializerManager.getSerializer(SerializerManager.Hessian2).deserialize(
					data.array(), CounterOperation.class.getName());
			} catch (final CodecException e) {
				LOG.error("Fail to decode IncrementAndGetRequest", e);
			}
			// follower ignore read operation
			if (counterOperation != null && counterOperation.isReadOp()) {
				iter.next();
				continue;
			}
		}

		if (counterOperation != null) {
			switch (counterOperation.getOp()) {
				case GET: // 获取操作
					current = this.value.get();
					LOG.info("Get value={} at logIndex={}", current, iter.getIndex());
					break;
				case INCREMENT: // 自增操作
					final long delta = counterOperation.getDelta();
					final long prev = this.value.get();
					current = this.value.addAndGet(delta);
					LOG.info("Added value={} by delta={} at logIndex={}", prev, delta, iter.getIndex());
					break;
			}

			if (closure != null) {
				closure.success(current);
				closure.run(Status.OK());
			}
		}

		iter.next();
	}
}