TransportSingleShardAction的主要作用是处理来自客户端的单个分片请求。TransportSingleShardAction会负责将请求发送到正确的节点，以便执行相关操作。
以下是TransportSingleShardAction的主要作用：
确定请求要访问的分片：当客户端向集群发送请求时，TransportSingleShardAction会解析请求并确定应该访问哪个分片。它会检查请求中的索引、分片号等信息，然后将请求路由到正确的节点上。
处理请求并返回结果：一旦请求到达正确的节点，TransportSingleShardAction将处理请求并返回结果。它会从节点中检索数据、执行搜索、更新索引等操作，并将结果返回给客户端。
处理故障情况：如果在处理请求时发生故障，TransportSingleShardAction会处理异常并返回错误消息给客户端。它还能够检测到节点故障并尝试将请求路由到备用节点上。
总的来说，TransportSingleShardAction负责将请求路由到正确的分片上并执行相关操作.
TransportSingleShardAction 能处理的请求, 都是针对单个shard的请求SingleShardRequest
TransportSingleShardAction<Request extends SingleShardRequest<Request>, Response extends ActionResponse> 从这里也可以看出: Request extends SingleShardRequest<Request> 这类范型在这里的作用, 其实想象表达子类都必须是 SingleShardRequest
- Get
- MultiGet
- TermVectors
- Analyze
- GetField
- Explain
- .....
公共初始化依赖:
- ClusterService
- IndexNameExpressionResolver
- ActionFilters
- ...
提供给子类的抽象方法:
abstract Response shardOperation(Request request, ShardId shardId) throws IOException;abstract boolean resolveIndex(Request request);void resolveRequest(ClusterState state, InternalRequest request)-
abstract ShardsIterator shards(ClusterState state, InternalRequest request);计算可以执行操作的候选shards, 如果返回null, 可能是本地执行了 String getExecutor(Request request, ShardId shardId)
整体的执行流程封装在了内部类:AsyncSingleAction, 在start方法中:
1. 通过各自子类的 shards 方法, 获得要执行的 shards, 不同的子类实现不同: 比如, 查某个逻辑shard的, 这里返回的可能是insync replica 组的所有副本的shard
this.shardIt = shards(clusterState, internalRequest);
...................
...................
2. 判断是否有需要执行的shards, 如果没有的话, 说明可能是不需要shard就能执行的操作
if (shardIt == null) {
executeLocally
}else{
perform // 递归操作各个shard
}
................
.................
3. 获取shard的node, 发送内部请请求
ShardRouting shardRouting = shardIt.nextOrNull();
if (shardRouting == null) {
listener.onFailure(failure);
return;
}
DiscoveryNode node = nodes.get(shardRouting.currentNodeId());
if (node == null) { // onFailure 会递归调用 perform , 每个shard候选都尝试一遍
onFailure(shardRouting, new NoShardAvailableActionException(shardRouting.shardId()));
} else {
transportService.sendRequest(); 给shard所在的node 发送内部执行请求
}
TransportGetAction 分析举例
重写方法
1. 定位候选shards
OperationRouting#getShards 会依据自适应副本策略, 选择最合适的shards, 把它们放在shards最前面. 因为它们的陈功率最高, 速度可能最快.
@Override
protected ShardIterator shards(ClusterState state, InternalRequest request) {
return clusterService.operationRouting() // operationRouting里面保存了关于自适应选择副本的设置
.getShards(
clusterService.state(),
request.concreteIndex(),
request.request().id(),
request.request().routing(),
request.request().preference()
);
}
2. 根据集群的一些状态信息解析Get请求到内部请求
3. 选择线程池
protected String getExecutor(GetRequest request, ShardId shardId) {
final ClusterState clusterState = clusterService.state();
if (clusterState.metadata().getIndexSafe(shardId.getIndex()).isSystem()) {
return ThreadPool.Names.SYSTEM_READ;
} else if (indicesService.indexServiceSafe(shardId.getIndex()).getIndexSettings().isSearchThrottled()) {
return ThreadPool.Names.SEARCH_THROTTLED;
} else {
return super.getExecutor(request, shardId);
}
}
4. 重写异步shard的执行
根据 realtime 参数的值, 可以看到有两个执行逻辑:
a. realtime : 直接调用父类的 shardOperation, 其实就是会执行子类的shardOperation
b. realtime参数不为true : 会等待 shardSearchActive, 即如果有等待刷新落盘的的translog, 那等他刷完之后, 再执行shardOperation
这里涉及到一些关于Get的实时性如何实现的细节(从translog读, 或者refresh下), 在不同版本里Get操作的实时性实现是不同的, 并且有点版本, 还有问题, 比如update的时候的原子性问题, 具体见下一章节
asyncShardOperation
@Override
protected void asyncShardOperation(GetRequest request, ShardId shardId, ActionListener<GetResponse> listener) throws IOException {
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
IndexShard indexShard = indexService.getShard(shardId.id());
if (request.realtime()) { // we are not tied to a refresh cycle here anyway
super.asyncShardOperation(request, shardId, listener);
} else {
indexShard.awaitShardSearchActive(b -> {
try {
super.asyncShardOperation(request, shardId, listener);
} catch (Exception ex) {
listener.onFailure(ex);
}
});
}
}
shardOperation
根据 if 语句里的逻辑可知: Get 操作只有在refresh=true&realtime=false的情况下, 会主动触发一次refresh
@Override
protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
IndexShard indexShard = indexService.getShard(shardId.id());
if (request.refresh() && !request.realtime()) { // 如果设置了refresh, 先执行下. 不影响 realtime
indexShard.refresh("refresh_flag_get");
}
GetResult result = indexShard.getService() // 调用
.get(
request.type(),
request.id(),
request.storedFields(),
request.realtime(),
request.version(),
request.versionType(),
request.fetchSourceContext()
);
return new GetResponse(result);
}
Get的realtime 表现和背后实现
可以看一些pr
https://github.com/elastic/elasticsearch/pull/20102
https://github.com/elastic/elasticsearch/pull/48843
https://github.com/elastic/elasticsearch/pull/64504
早期从translog读, 后来从lucene读(5.x开始), 或者是searcher读, 现在7.x 从translog
参考: https://blog.csdn.net/qq_42848795/article/details/109711741
get的实时是不受refresh频率影响的. 只要realtime为true ,一定可以获取最新的. 哪怕还没有refresh(并且如果refresh参数不设置为true, 不会主动触发refresh).
https://www.elastic.co/guide/en/elasticsearch/reference/7.10/docs-get.html
Realtime
By default, the get API is realtime, and is not affected by the refresh rate of the index (when data will become visible for search). In case where stored fields are requested (see stored_fields parameter) and the document has been updated but is not yet refreshed, the get API will have to parse and analyze the source to extract the stored fields. In order to disable realtime GET, the realtime parameter can be set to false.
调用链:
ShardService org.opensearch.index.get.ShardGetService#innerGet 得到org.opensearch.index.get.GetResult
IndexShard#get -> Engine#get 得到Engine.GetResult
可以看到, 在构造 Engine.Get 的第二个参数 readFromTranslog 如果 realtime为true ,则readFromTrasnlog 也为true. 尽管在InternalEngine#get 方法中, 既有从translog读取, 也有从Saercher读的逻辑(如果在versionmap中没有这个doc, 则说明大概率是已经写入了, 可见了).
get = indexShard.get(
new Engine.Get(realtime, realtime, type, id, uidTerm).version(version)
.versionType(versionType)
.setIfSeqNo(ifSeqNo)
.setIfPrimaryTerm(ifPrimaryTerm)
);
