es源码笔记-Routing

ES 建索引时默认是根据文档标识符 _id 将文档均分至多个分片。当搜索数据时，默认查询所有分片结果然后汇总，而并不必须知道数据到底存在哪个分片上。

路由算法就是根据routing和文档id计算目标shardid的过程。
一般情况下，路由计算方式为下面的公式：

shard_num = hash（_routing） % num_primary_shards

默认情况下，_routing值就是文档id。
ES使用随机id和Hash算法来确保文档均匀地分配给分片。当使用自定义id或routing时， id 或 routing 值可能不够随机，造成数据倾斜，部分分片过大。在这种情况下，可以使用index.routing_partition_size 配置来减少倾斜的风险。routing_partition_size越大，数据的分布越均匀。
在设置了index.routing_partition_size的情况下，计算公式为：

shard_num = （hash（_routing） + hash（_id） % routing_partition_size） % num_primary_shards

也就是说，对于同一个routing值，hash(_routing)的结果固定的，hash(_id) % routing_partition_size的结果有 routing_partition_size 个可能的值，两个组合在一起，对于同一个routing值的多个doc，也就能计算出 routing_partition_size 可能的shard了，即一个shard集合。
index.routing_partition_size取值应具有大于1且小于index.number_of_shards的值

计算过程的实现如下

    private static int calculateScaledShardId(IndexMetaData indexMetaData, String effectiveRouting, int partitionOffset) {
        final int hash = Murmur3HashFunction.hash(effectiveRouting) + partitionOffset;

        // we don't use IMD#getNumberOfShards since the index might have been shrunk such that we need to use the size
        // of original index to hash document
        return Math.floorMod(hash, indexMetaData.getRoutingNumShards()) / indexMetaData.getRoutingFactor();
    }

Search时如何根据routing找到指定的分片？

例子

GET /{index}/{type}/_search?routing=beijing

通过发送search请求查询数据。指定了routing是beijing

1、解析流程

当es接收到上面的请求时，交给org.elasticsearch.rest.action.search.RestSearchAction处理，repareRequest方法中将请求体解析为SearchRequest数据结构

public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
        SearchRequest searchRequest = new SearchRequest();      
        IntConsumer setSize = size -> searchRequest.source().size(size);
        request.withContentOrSourceParamParserOrNull(parser ->
            parseSearchRequest(searchRequest, request, parser, setSize));

        return channel -> client.search(searchRequest, new RestStatusToXContentListener<>(channel));
    }

    public static void parseSearchRequest(SearchRequest searchRequest, RestRequest request,
                                          XContentParser requestContentParser,
                                          IntConsumer setSize) throws IOException {      
        String searchType = request.param("search_type");
        parseSearchSource(searchRequest.source(), request, setSize);        searchRequest.requestCache(request.paramAsBoolean("request_cache", null));
        String scroll = request.param("scroll");
        searchRequest.routing(request.param("routing")); //接收到routing参数，并封装到searchRequest中
        searchRequest.preference(request.param("preference"));
        searchRequest.indicesOptions(IndicesOptions.fromRequest(request, searchRequest.indicesOptions()));        searchRequest.setCcsMinimizeRoundtrips(request.paramAsBoolean("ccs_minimize_roundtrips", true));
        checkRestTotalHits(request, searchRequest);
    }

构造目的shard列表

prepareRequest方法构造请求后通过transport模块发送给org.elasticsearch.action.search.TransportSearchAction处理

    private void executeSearch(SearchTask task, SearchTimeProvider timeProvider, SearchRequest searchRequest,
                               OriginalIndices localIndices, String[] concreteIndices, Map<String, Set<String>> routingMap,
                               Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
                               List<SearchShardIterator> remoteShardIterators, BiFunction<String, String, DiscoveryNode> remoteConnections,
                               ClusterState clusterState, ActionListener<SearchResponse> listener, SearchResponse.Clusters clusters) {

        Map<String, Long> nodeSearchCounts = searchTransportService.getPendingSearchRequests();
        GroupShardsIterator<ShardIterator> localShardsIterator = clusterService.operationRouting().searchShards(clusterState,
                concreteIndices, routingMap, searchRequest.preference(), searchService.getResponseCollectorService(), nodeSearchCounts);
        GroupShardsIterator<SearchShardIterator> shardIterators = mergeShardsIterators(localShardsIterator, localIndices,
            searchRequest.getLocalClusterAlias(), remoteShardIterators);

       。。。。
    }

将请求涉及的本集群shard列表和远程集群的shard列表（远程集群用于跨集群访问）合并

其中routing查找指定分片的流程就在org.elasticsearch.cluster.routing.OperationRouting.searchShards(ClusterState, String[], Map<String, Set<String>>, String, ResponseCollectorService, Map<String, Long>)方法中

    public GroupShardsIterator<ShardIterator> searchShards(ClusterState clusterState,
                                                           String[] concreteIndices,
                                                           @Nullable Map<String, Set<String>> routing,
                                                           @Nullable String preference,
                                                           @Nullable ResponseCollectorService collectorService,
                                                           @Nullable Map<String, Long> nodeCounts) {
        final Set<IndexShardRoutingTable> shards = computeTargetedShards(clusterState, concreteIndices, routing);
····
    }

    private Set<IndexShardRoutingTable> computeTargetedShards(ClusterState clusterState, String[] concreteIndices,
                                                              @Nullable Map<String, Set<String>> routing) {
   
        for (String index : concreteIndices) {
           ····
            if (effectiveRouting != null) {
                for (String r : effectiveRouting) {
                 
                        set.add(RoutingTable.shardRoutingTable(indexRouting, calculateScaledShardId(indexMetaData, r, partitionOffset)));
                   
                }
        
        }
        return set;
    }

最终调用calculateScaledShardId方法计算出指定的分片