集群状态发布
cluster模块封装了在集群层面执行的任务，如集群健康、集群级元信息管理、分片分配给节点、节点管理等。集群任务执行之后可能会产生新的集群状态，如果产生新的集群状态主节点会将集群状态广播给其他节点。
集群状态封装在clusterState中，支持增量同步
提交集群任务的主要时机有以下几种：
- 索引的创建、删除、打开、关闭
- 索引模板、映射、别名的变化
- gateway模块发布选举出来的集群状态
- 快照
- 分片分配
- 集群节点变化等
提交集群任务入口在ClusterService的submitStateUpdateTask方法，第一个参数是事件来源，第二个参数是要执行的具体任务
public <T extends ClusterStateTaskConfig & ClusterStateTaskExecutor<T> & ClusterStateTaskListener>
void submitStateUpdateTask(String source, T updateTask) {
submitStateUpdateTask(source, updateTask, updateTask, updateTask, updateTask);
}
public <T> void submitStateUpdateTask(String source, T task,
ClusterStateTaskConfig config,
ClusterStateTaskExecutor<T> executor,
ClusterStateTaskListener listener) {
submitStateUpdateTasks(source, Collections.singletonMap(task, listener), config, executor);
}
public <T> void submitStateUpdateTasks(final String source,
final Map<T, ClusterStateTaskListener> tasks, final ClusterStateTaskConfig config,
final ClusterStateTaskExecutor<T> executor) {
masterService.submitStateUpdateTasks(source, tasks, config, executor);
}
最有代表性的任务是ClusterStateUpdateTask，它实现了ClusterStateTaskConfig、ClusterStateTaskExecutor
public abstract class ClusterStateUpdateTask
implements ClusterStateTaskConfig, ClusterStateTaskExecutor<ClusterStateUpdateTask>, ClusterStateTaskListener {
ClusterStateTaskConfig包含了任务的配置信息和优先级
TimeValue timeout();
Priority priority();
ClusterStateTaskExecutor主要是定义要执行的任务，最主要的方法就是execute方法
ClusterTasksResult<T> execute(ClusterState currentState, List<T> tasks) throws Exception;
任务执行时会传入当前集群状态，任务运行过程中如果产生新的集群状态就返回新的集群状态，如果没有就返回原来的集群状态
ClusterStateTaskListener主要是提交任务后的回调处理
/**
* A callback called when execute fails.
*/
void onFailure(String source, Exception e);
/**
* called when the task was rejected because the local node is no longer master.
* Used only for tasks submitted to {@link MasterService}.
*/
default void onNoLongerMaster(String source) {
onFailure(source, new NotMasterException("no longer master. source: [" + source + "]"));
}
/**
* Called when the result of the {@link ClusterStateTaskExecutor#execute(ClusterState, List)} have been processed
* properly by all listeners.
*/
default void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
}
MasterService主要负责集群任务管理和运行，只有主节点会提交集群任务到内部队列，并运行队列中的任务
public <T> void submitStateUpdateTasks(final String source,
final Map<T, ClusterStateTaskListener> tasks, final ClusterStateTaskConfig config,
final ClusterStateTaskExecutor<T> executor) {
if (!lifecycle.started()) {
return;
}
final ThreadContext threadContext = threadPool.getThreadContext();
final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(true);
try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
threadContext.markAsSystemContext();
//封装任务
List<Batcher.UpdateTask> safeTasks = tasks.entrySet().stream()
.map(e -> taskBatcher.new UpdateTask(config.priority(), source, e.getKey(), safe(e.getValue(), supplier), executor))
.collect(Collectors.toList());
//提交任务
taskBatcher.submitTasks(safeTasks, config.timeout());
} catch (EsRejectedExecutionException e) {
// ignore cases where we are shutting down..., there is really nothing interesting
// to be done here...
if (!lifecycle.stoppedOrClosed()) {
throw e;
}
}
}
public void submitTasks(List<? extends BatchedTask> tasks, @Nullable TimeValue timeout) throws EsRejectedExecutionException {
if (tasks.isEmpty()) {
return;
}
final BatchedTask firstTask = tasks.get(0);
assert tasks.stream().allMatch(t -> t.batchingKey == firstTask.batchingKey) :
"tasks submitted in a batch should share the same batching key: " + tasks;
// convert to an identity map to check for dups based on task identity
//根据任务标识检查重复数据
final Map<Object, BatchedTask> tasksIdentity = tasks.stream().collect(Collectors.toMap(
BatchedTask::getTask,
Function.identity(),
(a, b) -> { throw new IllegalStateException("cannot add duplicate task: " + a); },
IdentityHashMap::new));
synchronized (tasksPerBatchingKey) {
//添加相同batchingKey的任务，返回已存在batchingKey的任务
LinkedHashSet<BatchedTask> existingTasks = tasksPerBatchingKey.computeIfAbsent(firstTask.batchingKey,
k -> new LinkedHashSet<>(tasks.size()));
//检查是否存在相同batchingKey的任务
for (BatchedTask existing : existingTasks) {
// check that there won't be two tasks with the same identity for the same batching key
BatchedTask duplicateTask = tasksIdentity.get(existing.getTask());
if (duplicateTask != null) {
throw new IllegalStateException("task [" + duplicateTask.describeTasks(
Collections.singletonList(existing)) + "] with source [" + duplicateTask.source + "] is already queued");
}
}
existingTasks.addAll(tasks);
}
//执行任务
if (timeout != null) {
threadExecutor.execute(firstTask, timeout, () -> onTimeoutInternal(tasks, timeout));
} else {
threadExecutor.execute(firstTask);
}
}
这里有去重逻辑，拥有相同ClusterStateTaskExecutor对象实例的任务只会执行一次，然后对于其他相同的实例直接赋值相同的执行结果。区分重复任务的方式时通过定义的任务本身，去重的方式不是将重复的数据删除，而是在执行完任务后赋予重复任务相同的结果。
ClusterStateTaskExecutor相同有两种情况可能是提交的任务本身重复，还有就是之前提交的任务已存在，但是尚未执行此时提交相同的任务就会保存到对应的列表中，只会执行一次
任务会被封装到UpdateTask中
class UpdateTask extends BatchedTask {
final ClusterStateTaskListener listener;
UpdateTask(Priority priority, String source, Object task, ClusterStateTaskListener listener,
ClusterStateTaskExecutor<?> executor) {
super(priority, source, executor, task);
this.listener = listener;
}
@Override
public String describeTasks(List<? extends BatchedTask> tasks) {
return ((ClusterStateTaskExecutor<Object>) batchingKey).describeTasks(
tasks.stream().map(BatchedTask::getTask).collect(Collectors.toList()));
}
}
提交到线程池运行调用run方法
@Override
public void run() {
//运行还没处理的任务
runIfNotProcessed(this);
}
void runIfNotProcessed(BatchedTask updateTask) {
//具有相同batching key的任务只会执行一次
if (updateTask.processed.get() == false) {
final List<BatchedTask> toExecute = new ArrayList<>();
final Map<String, List<BatchedTask>> processTasksBySource = new HashMap<>();
synchronized (tasksPerBatchingKey) {
//获取任务列表
LinkedHashSet<BatchedTask> pending = tasksPerBatchingKey.remove(updateTask.batchingKey);
if (pending != null) {
for (BatchedTask task : pending) {
if (task.processed.getAndSet(true) == false) {
logger.trace("will process {}", task);
//构建要执行的任务列表
toExecute.add(task);
processTasksBySource.computeIfAbsent(task.source, s -> new ArrayList<>()).add(task);
} else {
logger.trace("skipping {}, already processed", task);
}
}
}
}
if (toExecute.isEmpty() == false) {
final String tasksSummary = processTasksBySource.entrySet().stream().map(entry -> {
String tasks = updateTask.describeTasks(entry.getValue());
return tasks.isEmpty() ? entry.getKey() : entry.getKey() + "[" + tasks + "]";
}).reduce((s1, s2) -> s1 + ", " + s2).orElse("");
//执行任务
run(updateTask.batchingKey, toExecute, tasksSummary);
}
}
}
执行任务并发布集群状态的逻辑在MasterService中
@Override
protected void run(Object batchingKey, List<? extends BatchedTask> tasks, String tasksSummary) {
ClusterStateTaskExecutor<Object> taskExecutor = (ClusterStateTaskExecutor<Object>) batchingKey;
List<UpdateTask> updateTasks = (List<UpdateTask>) tasks;
//运行任务，并发布集群状态
runTasks(new TaskInputs(taskExecutor, updateTasks, tasksSummary));
}
private void runTasks(TaskInputs taskInputs) {
final String summary = taskInputs.summary;
if (!lifecycle.started()) {
logger.debug("processing [{}]: ignoring, master service not started", summary);
return;
}
logger.debug("executing cluster state update for [{}]", summary);
//之前集群状态
final ClusterState previousClusterState = state();
//只在主节点执行
if (!previousClusterState.nodes().isLocalNodeElectedMaster() && taskInputs.runOnlyWhenMaster()) {
logger.debug("failing [{}]: local node is no longer master", summary);
taskInputs.onNoLongerMaster();
return;
}
final long computationStartTime = threadPool.relativeTimeInMillis();
//执行task任务生成新的集群状态
final TaskOutputs taskOutputs = calculateTaskOutputs(taskInputs, previousClusterState);
taskOutputs.notifyFailedTasks();
final TimeValue computationTime = getTimeSince(computationStartTime);
logExecutionTime(computationTime, "compute cluster state update", summary);
if (taskOutputs.clusterStateUnchanged()) {
final long notificationStartTime = threadPool.relativeTimeInMillis();
taskOutputs.notifySuccessfulTasksOnUnchangedClusterState();
final TimeValue executionTime = getTimeSince(notificationStartTime);
logExecutionTime(executionTime, "notify listeners on unchanged cluster state", summary);
} else {//集群状态发生改变
final ClusterState newClusterState = taskOutputs.newClusterState;
if (logger.isTraceEnabled()) {
logger.trace("cluster state updated, source [{}]\n{}", summary, newClusterState);
} else {
logger.debug("cluster state updated, version [{}], source [{}]", newClusterState.version(), summary);
}
final long publicationStartTime = threadPool.relativeTimeInMillis();
try {
ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent(summary, newClusterState, previousClusterState);
// new cluster state, notify all listeners
final DiscoveryNodes.Delta nodesDelta = clusterChangedEvent.nodesDelta();
if (nodesDelta.hasChanges() && logger.isInfoEnabled()) {
String nodesDeltaSummary = nodesDelta.shortSummary();
if (nodesDeltaSummary.length() > 0) {
logger.info("{}, term: {}, version: {}, delta: {}",
summary, newClusterState.term(), newClusterState.version(), nodesDeltaSummary);
}
}
logger.debug("publishing cluster state version [{}]", newClusterState.version());
//发布集群状态
publish(clusterChangedEvent, taskOutputs, publicationStartTime);
} catch (Exception e) {
handleException(summary, publicationStartTime, newClusterState, e);
}
}
}
执行方法前判断是不是主节点因为只有主节点可以运行集群任务，根据执行任务前的集群状态执行任务生成新的集群状态
执行任务获取任务执行结果，并生成新的集群状态
private TaskOutputs calculateTaskOutputs(TaskInputs taskInputs, ClusterState previousClusterState) {
//执行提交的任务，并且返回新的集群状态
ClusterTasksResult<Object> clusterTasksResult = executeTasks(taskInputs, previousClusterState);
//根据分配分片结果生成新的集群状态
ClusterState newClusterState = patchVersions(previousClusterState, clusterTask