序
本文主要研究一下 nacos RaftCore 的 signalPublish
RaftCore
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/persistent/raft/RaftCore.java
@Component
public class RaftCore {
public static final String API_VOTE = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/vote";
public static final String API_BEAT = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/beat";
public static final String API_PUB = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
public static final String API_DEL = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
public static final String API_GET = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
public static final String API_ON_PUB = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum/commit";
public static final String API_ON_DEL = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum/commit";
public static final String API_GET_PEER = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/peer";
private ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {Thread t = new Thread(r);
t.setDaemon(true);
t.setName("com.alibaba.nacos.naming.raft.notifier");
return t;
}
});
public static final Lock OPERATE_LOCK = new ReentrantLock();
public static final int PUBLISH_TERM_INCREASE_COUNT = 100;
private volatile Map<String, List<RecordListener>> listeners = new ConcurrentHashMap<>();
private volatile ConcurrentMap<String, Datum> datums = new ConcurrentHashMap<>();
@Autowired
private RaftPeerSet peers;
@Autowired
private SwitchDomain switchDomain;
@Autowired
private GlobalConfig globalConfig;
@Autowired
private RaftProxy raftProxy;
@Autowired
private RaftStore raftStore;
public volatile Notifier notifier = new Notifier();
private boolean initialized = false;
@PostConstruct
public void init() throws Exception {Loggers.RAFT.info("initializing Raft sub-system");
executor.submit(notifier);
long start = System.currentTimeMillis();
raftStore.loadDatums(notifier, datums);
setTerm(NumberUtils.toLong(raftStore.loadMeta().getProperty("term"), 0L));
Loggers.RAFT.info("cache loaded, datum count: {}, current term: {}", datums.size(), peers.getTerm());
while (true) {if (notifier.tasks.size() <= 0) {break;}
Thread.sleep(1000L);
}
initialized = true;
Loggers.RAFT.info("finish to load data from disk, cost: {} ms.", (System.currentTimeMillis() - start));
GlobalExecutor.registerMasterElection(new MasterElection());
GlobalExecutor.registerHeartbeat(new HeartBeat());
Loggers.RAFT.info("timer started: leader timeout ms: {}, heart-beat timeout ms: {}",
GlobalExecutor.LEADER_TIMEOUT_MS, GlobalExecutor.HEARTBEAT_INTERVAL_MS);
}
public Map<String, List<RecordListener>> getListeners() {return listeners;}
public void signalPublish(String key, Record value) throws Exception {if (!isLeader()) {JSONObject params = new JSONObject();
params.put("key", key);
params.put("value", value);
Map<String, String> parameters = new HashMap<>(1);
parameters.put("key", key);
raftProxy.proxyPostLarge(getLeader().ip, API_PUB, params.toJSONString(), parameters);
return;
}
try {OPERATE_LOCK.lock();
long start = System.currentTimeMillis();
final Datum datum = new Datum();
datum.key = key;
datum.value = value;
if (getDatum(key) == null) {datum.timestamp.set(1L);
} else {datum.timestamp.set(getDatum(key).timestamp.incrementAndGet());
}
JSONObject json = new JSONObject();
json.put("datum", datum);
json.put("source", peers.local());
onPublish(datum, peers.local());
final String content = JSON.toJSONString(json);
final CountDownLatch latch = new CountDownLatch(peers.majorityCount());
for (final String server : peers.allServersIncludeMyself()) {if (isLeader(server)) {latch.countDown();
continue;
}
final String url = buildURL(server, API_ON_PUB);
HttpClient.asyncHttpPostLarge(url, Arrays.asList("key=" + key), content, new AsyncCompletionHandler<Integer>() {
@Override
public Integer onCompleted(Response response) throws Exception {if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {Loggers.RAFT.warn("[RAFT] failed to publish data to peer, datumId={}, peer={}, http code={}",
datum.key, server, response.getStatusCode());
return 1;
}
latch.countDown();
return 0;
}
@Override
public STATE onContentWriteCompleted() {return STATE.CONTINUE;}
});
}
if (!latch.await(UtilsAndCommons.RAFT_PUBLISH_TIMEOUT, TimeUnit.MILLISECONDS)) {
// only majority servers return success can we consider this update success
Loggers.RAFT.error("data publish failed, caused failed to notify majority, key={}", key);
throw new IllegalStateException("data publish failed, caused failed to notify majority, key=" + key);
}
long end = System.currentTimeMillis();
Loggers.RAFT.info("signalPublish cost {} ms, key: {}", (end - start), key);
} finally {OPERATE_LOCK.unlock();
}
}
//......
}
- signalPublish 方法判断当前节点是否是 leader,如果不是则转发 publish 到 leader 节点的
/v1/ns/raft/datum
接口 - 如果是 leader 则构造 datum 以及 peers.majorityCount()大小的 CountDownLatch,然后遍历 peers.allServersIncludeMyself(),对于 leader 节点直接 latch.countDown(),对于非 leader 节点则发送异步请求,请求
/v1/ns/raft/datum/commit
接口,在 onCompleted 的时候,如果请求成功执行 latch.countDown() - 最后对于 CountDownLatch 未能在 RAFT_PUBLISH_TIMEOUT 返回的,抛出 IllegalStateException
RaftController
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/controllers/RaftController.java
@RestController
@RequestMapping({UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft",
UtilsAndCommons.NACOS_SERVER_CONTEXT + UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft"})
public class RaftController {
@Autowired
private RaftConsistencyServiceImpl raftConsistencyService;
@Autowired
private ServiceManager serviceManager;
@Autowired
private RaftCore raftCore;
//......
@NeedAuth
@RequestMapping(value = "/datum/commit", method = RequestMethod.POST)
public String onPublish(HttpServletRequest request, HttpServletResponse response) throws Exception {response.setHeader("Content-Type", "application/json; charset=" + getAcceptEncoding(request));
response.setHeader("Cache-Control", "no-cache");
response.setHeader("Content-Encode", "gzip");
String entity = IOUtils.toString(request.getInputStream(), "UTF-8");
String value = URLDecoder.decode(entity, "UTF-8");
JSONObject jsonObject = JSON.parseObject(value);
String key = "key";
RaftPeer source = JSON.parseObject(jsonObject.getString("source"), RaftPeer.class);
JSONObject datumJson = jsonObject.getJSONObject("datum");
Datum datum = null;
if (KeyBuilder.matchInstanceListKey(datumJson.getString(key))) {datum = JSON.parseObject(jsonObject.getString("datum"), new TypeReference<Datum<Instances>>() {});
} else if (KeyBuilder.matchSwitchKey(datumJson.getString(key))) {datum = JSON.parseObject(jsonObject.getString("datum"), new TypeReference<Datum<SwitchDomain>>() {});
} else if (KeyBuilder.matchServiceMetaKey(datumJson.getString(key))) {datum = JSON.parseObject(jsonObject.getString("datum"), new TypeReference<Datum<Service>>() {});
}
raftConsistencyService.onPut(datum, source);
return "ok";
}
//......
}
- onPublish 方法主要是执行 raftConsistencyService.onPut(datum, source)
RaftConsistencyServiceImpl
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/persistent/raft/RaftConsistencyServiceImpl.java
@Service
public class RaftConsistencyServiceImpl implements PersistentConsistencyService {
@Autowired
private RaftCore raftCore;
@Autowired
private RaftPeerSet peers;
@Autowired
private SwitchDomain switchDomain;
//......
public void onPut(Datum datum, RaftPeer source) throws NacosException {
try {raftCore.onPublish(datum, source);
} catch (Exception e) {Loggers.RAFT.error("Raft onPut failed.", e);
throw new NacosException(NacosException.SERVER_ERROR, "Raft onPut failed, datum:" + datum + ", source:" + source, e);
}
}
//......
}
- onPut 方法执行的是 raftCore.onPublish(datum, source)
RaftCore
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/persistent/raft/RaftCore.java
@Component
public class RaftCore {
public static final String API_VOTE = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/vote";
public static final String API_BEAT = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/beat";
public static final String API_PUB = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
public static final String API_DEL = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
public static final String API_GET = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
public static final String API_ON_PUB = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum/commit";
public static final String API_ON_DEL = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum/commit";
public static final String API_GET_PEER = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/peer";
private ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {Thread t = new Thread(r);
t.setDaemon(true);
t.setName("com.alibaba.nacos.naming.raft.notifier");
return t;
}
});
public static final Lock OPERATE_LOCK = new ReentrantLock();
public static final int PUBLISH_TERM_INCREASE_COUNT = 100;
private volatile Map<String, List<RecordListener>> listeners = new ConcurrentHashMap<>();
private volatile ConcurrentMap<String, Datum> datums = new ConcurrentHashMap<>();
@Autowired
private RaftPeerSet peers;
@Autowired
private SwitchDomain switchDomain;
@Autowired
private GlobalConfig globalConfig;
@Autowired
private RaftProxy raftProxy;
@Autowired
private RaftStore raftStore;
public volatile Notifier notifier = new Notifier();
private boolean initialized = false;
//......
public void onPublish(Datum datum, RaftPeer source) throws Exception {RaftPeer local = peers.local();
if (datum.value == null) {Loggers.RAFT.warn("received empty datum");
throw new IllegalStateException("received empty datum");
}
if (!peers.isLeader(source.ip)) {Loggers.RAFT.warn("peer {} tried to publish data but wasn't leader, leader: {}",
JSON.toJSONString(source), JSON.toJSONString(getLeader()));
throw new IllegalStateException("peer(" + source.ip + ") tried to publish" +
"data but wasn't leader");
}
if (source.term.get() < local.term.get()) {Loggers.RAFT.warn("out of date publish, pub-term: {}, cur-term: {}",
JSON.toJSONString(source), JSON.toJSONString(local));
throw new IllegalStateException("out of date publish, pub-term:"
+ source.term.get() + ", cur-term:" + local.term.get());
}
local.resetLeaderDue();
// if data should be persistent, usually this is always true:
if (KeyBuilder.matchPersistentKey(datum.key)) {raftStore.write(datum);
}
datums.put(datum.key, datum);
if (isLeader()) {local.term.addAndGet(PUBLISH_TERM_INCREASE_COUNT);
} else {if (local.term.get() + PUBLISH_TERM_INCREASE_COUNT > source.term.get()) {
//set leader term:
getLeader().term.set(source.term.get());
local.term.set(getLeader().term.get());
} else {local.term.addAndGet(PUBLISH_TERM_INCREASE_COUNT);
}
}
raftStore.updateTerm(local.term.get());
notifier.addTask(datum.key, ApplyAction.CHANGE);
Loggers.RAFT.info("data added/updated, key={}, term={}", datum.key, local.term);
}
//......
}
- onPublish 方法首先判断请求的节点是否是 leader,不是则抛出 IllegalStateException;对于 source.term 小于 local.term 的抛出 IllegalStateException
- 之后执行 local.resetLeaderDue(),以及 raftStore.write(datum),datums.put(datum.key, datum);对于 leader 节点执行 local.term.addAndGet(PUBLISH_TERM_INCREASE_COUNT),非 leader 节点则更新 leader term 以及 local.term
- 最后执行 raftStore.updateTerm(local.term.get())以及 notifier.addTask(datum.key, ApplyAction.CHANGE)
小结
- signalPublish 方法判断当前节点是否是 leader,如果不是则转发 publish 到 leader 节点的
/v1/ns/raft/datum
接口 - 如果是 leader 则构造 datum 以及 peers.majorityCount()大小的 CountDownLatch,然后遍历 peers.allServersIncludeMyself(),对于 leader 节点直接 latch.countDown(),对于非 leader 节点则发送异步请求,请求
/v1/ns/raft/datum/commit
接口,在 onCompleted 的时候,如果请求成功执行 latch.countDown() - 最后对于 CountDownLatch 未能在 RAFT_PUBLISH_TIMEOUT 返回的,抛出 IllegalStateException
doc
- RaftCore