solrCloud 索引更新逻辑学习笔记
2017-09-14 10:39
393 查看
ZkCoreNodeProps 封装了一个node的相关信息，包括base_url，core_name,state,node_name,core_url,isLeader
SolrCmdDistributor
solr分布式更新的一个重要实现工具类，因为它本身的只负责分布式的请求处理，并没有很多的业务逻辑。
[java] view
plain copy
staticAdjustableSemaphoresemaphore = new AdjustableSemaphore(8);
限制同时并发的请求最多数。从构造函数看可以跟结点数相关，但最大是16.
[java] view
plain copy
public SolrCmdDistributor(intnumHosts, ThreadPoolExecutorexecutor) {
int maxPermits = Math.max(16, numHosts * 16);
// limits how many tasks can actually execute at once
if (maxPermits != semaphore.getMaxPermits()) {
semaphore.setMaxPermits(maxPermits);
}
completionService = new ExecutorCompletionService<Request>(executor);
pending = new HashSet<Future<Request>>();
}
[java] view
plain copy
privatefinalMap<Node,List<AddRequest>> adds = new HashMap<Node,List<AddRequest>>();
privatefinalMap<Node,List<DeleteRequest>> deletes = new HashMap<Node,List<DeleteRequest>>();
这两个字段主要是实现用于缓存更新请求
执行缓存的请求，调用下面方法
[java] view
plain copy
publicvoid finish() {
// piggyback on any outstanding adds or deletes if possible.
flushAdds(1);
flushDeletes(1);
checkResponses(true);
}
提交请求
[java] view
plain copy
void submit(UpdateRequestExt ureq, Node node) {
Request sreq = new Request();
sreq.node = node;
sreq.ureq = ureq;
submit(sreq);
}
然后是检查响应结果，调用
void checkResponses(booleanblock) 作为检查上一次提交的请求响应。当请求需要重试的时候，会默认重试最大次数10次
将最终结果返回到响应结果里，有异常也会记录下来。
分布式增加更新
[java] view
plain copy
publicvoid distribAdd(AddUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params) throws IOException {
//执行前都会去掉之前还会检查可能没响应完的请求，不等待，直接删除旧的请求。
checkResponses(false);
// 确保所有删除的请求被执行
flushDeletes(1);
//克隆更新请求重用
AddUpdateCommand clone = new AddUpdateCommand(null);
clone.solrDoc = cmd.solrDoc;
clone.commitWithin = cmd.commitWithin;
clone.overwrite = cmd.overwrite;
clone.setVersion(cmd.getVersion());
AddRequest addRequest = new AddRequest();
addRequest.cmd = clone;
addRequest.params = params;
//增加对每个节点的请求到缓存adds里
for (Nodenode : nodes) {
List<AddRequest> alist = adds.get(node);
if (alist == null) {
alist = new ArrayList<AddRequest>(2);
adds.put(node, alist);
}
alist.add(addRequest);
}
//执行缓存adds的请求
flushAdds(maxBufferedAddsPerServer);
}
其它的doDelete,addCommit的请求逻辑的处理都相差不多
DistributedUpdateProcessor
这个是solrCloud主要的一个更新处理链，使用cloud模式的时候必要的一个处理链，负责分布式更新的逻辑处理
一个重要的hash算法，作为更新记录具体分配到哪个shard的算法
这算法应该会在后期重构并设计为插件方式 ，可被用户自定议的hash算法替换。
[java] view
plain copy
private int hash(AddUpdateCommandcmd) {
String hashableId = cmd.getHashableId();
return Hash.murmurhash3_x86_32(hashableId, 0, hashableId.length(), 0);
}
private int hash(DeleteUpdateCommandcmd) {
return Hash.murmurhash3_x86_32(cmd.getId(), 0, cmd.getId().length(), 0);
}
其中cmd.getHashableId()方法返回的主要是文档的主键的值
通过hash值定位更新到哪个shard
[java] view
plain copy
private String getShard(int hash, String collection, ClusterState clusterState) {
return clusterState.getShard(hash, collection);
}
通过取到collection对应的RangeInfo，计算该hash值座落在哪个Range，就可以计算到相应的shard
[java] view
plain copy
public String getShard(int hash, String collection) {
RangeInfo rangInfo = getRanges(collection);
int cnt = 0;
for (Range range : rangInfo.ranges) {
if (range.includes(hash)) {
return rangInfo.shardList.get(cnt);
}
cnt++;
}
throw new IllegalStateException("The HashPartitioner failed");
}
HashPartitioner
做为切分为多个范围的Range，主要实现方法如下：
[java] view
plain copy
public List<Range> partitionRange(int partitions, int min, int max) {
assert max >= min;
if (partitions == 0) return Collections.EMPTY_LIST;
long range = (long)max - (long)min;
long srange = Math.max(1, range / partitions);
List<Range> ranges = new ArrayList<Range>(partitions);
long start = min;
long end = start;
while (end < max) {
end = start + srange;
// make last range always end exactly on MAX_VALUE
if (ranges.size() == partitions - 1) {
end = max;
}
ranges.add(new Range((int)start, (int)end));
start = end + 1L;
}
return ranges;
}
指定了某个范围[min,max]切分为多个partitions的Ranges;切分的范围是按平均的切分。
Range类封装了主键hash值范围【min,max】
RangeInfo封装一个collection下所有shard信息对应的Range,主要实现方法如下：
[java] view
plain copy
private RangeInfo addRangeInfo(String collection) {
List<Range> ranges;
RangeInfo rangeInfo= new RangeInfo();
Map<String,Slice> slices = getSlices(collection);
if (slices == null) {
throw new SolrException(ErrorCode.BAD_REQUEST, "Can not find collection "
+ collection + " in " + this);
}
Set<String> shards = slices.keySet();
ArrayList<String> shardList = new ArrayList<String>(shards.size());
shardList.addAll(shards);
Collections.sort(shardList);
ranges = hp.partitionRange(shards.size(), Integer.MIN_VALUE, Integer.MAX_VALUE);
rangeInfo.ranges = ranges;
rangeInfo.shardList = shardList;
rangeInfos.put(collection, rangeInfo);
return rangeInfo;
}
从上面方法的实现可以看到，会先将所有shard的名称排序，然后根据shard的大小切分相应的多个的范围 ，每一个shard在排序完的位置 有对应的范围Range，两者的信息存放在RangeInfo.
不用担心，上面按整数最小值 ，最大值的平均切分的范围会导致分配不匀的情况，
可能你会担心如果我的主键值是整数，那主键的hash值会不会跟他的值所对应呢，这样的话，会让hash出来的数据先填满小的shard，其它shard不够匀称。其实设计者本身使用的hash算法是针对任何类型，取的主键值也是以字节数组去做hash。这个可以自己使用它的hash算法去校验。
再来看一下DistributedUpdateProcessor
先看add请求，请求的来源有多种：
privateList<Node> setupRequest(inthash)
此方法就是为了判断上面请求来源而决定分发的结点
请求来自leader转发：FROMLEADER，那么就只需要写到本地ulog,不需要转发给leader，也不需要转发给其它replicas
请求不是来自leader，但自己就是leader，那么就需要将请求写到本地，顺便分发给其他的replicas.
请求不是来自leader，但自己又不是leader,也就是该更新请求是最原始的更新请求，那么需要将请求写到本地ulog，顺便转发给leader,再由leader分发
所以为了不让更新请求不会转发来转发去。提交索引的时候，只提交给所有leader是最佳选择。
也就是能预先知道该数据 是要到哪个leader，这个solrj好像有实现。solrcloudserver，分对更新的数据预先做分发请求。
先来讲一下增加的更新逻辑
[java] view
plain copy
@Override
public void processAdd(AddUpdateCommand cmd) throws IOException {
int hash = 0;
if (zkEnabled) {//cloud模式下
zkCheck();//检查zk连接状态
hash = hash(cmd);//取得更新请求hash值，再决定hash到哪一个shard
//判断更新请求来源，决定需要转发的nodes
nodes = setupRequest(hash);
} else {
isLeader = getNonZkLeaderAssumption(req);
}
boolean dropCmd = false;
if (!forwardToLeader) {//不需要转发，即是请求来源是由leader转发来的，或者本人就是leader
dropCmd = versionAdd(cmd);//决定该请求是否丢弃，丢弃原因：1）只需要更新到ulog，也就是请求是来源leader转发来的，不需写到索引。2）ulog已有该更新记录且版本比当前的版本更新，则可以丢弃。
}
// 可以丢弃该请求，不需要写到lucene索引
if (dropCmd) {
// TODO: do we need to add anything to the response?
return;
ModifiableSolrParams params = null;
if (nodes != null) {
params = new ModifiableSolrParams(filterParams(req.getParams()));
}
//需要转发该请求，转发leader,或者由leader分发
params.set(DISTRIB_UPDATE_PARAM,
(isLeader ?
DistribPhase.FROMLEADER.toString() :
DistribPhase.TOLEADER.toString()));
if (isLeader) {
params.set("distrib.from", ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
}
params.set("distrib.from", ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
//转发请求到nodes
cmdDistrib.distribAdd(cmd, nodes, params);
}
//增加更新的响应内容
if (returnVersions && rsp != null && idField != null) {
if (addsResponse == null) {
addsResponse = new NamedList<String>();
rsp.add("adds",addsResponse);
}
if (scratch == null) scratch = new CharsRef();
idField.getType().indexedToReadable(cmd.getIndexedId(), scratch);
addsResponse.add(scratch.toString(), cmd.getVersion());
}
}
SolrCmdDistributor
solr分布式更新的一个重要实现工具类，因为它本身的只负责分布式的请求处理，并没有很多的业务逻辑。
[java] view
plain copy
staticAdjustableSemaphoresemaphore = new AdjustableSemaphore(8);
限制同时并发的请求最多数。从构造函数看可以跟结点数相关，但最大是16.
[java] view
plain copy
public SolrCmdDistributor(intnumHosts, ThreadPoolExecutorexecutor) {
int maxPermits = Math.max(16, numHosts * 16);
// limits how many tasks can actually execute at once
if (maxPermits != semaphore.getMaxPermits()) {
semaphore.setMaxPermits(maxPermits);
}
completionService = new ExecutorCompletionService<Request>(executor);
pending = new HashSet<Future<Request>>();
}
[java] view
plain copy
privatefinalMap<Node,List<AddRequest>> adds = new HashMap<Node,List<AddRequest>>();
privatefinalMap<Node,List<DeleteRequest>> deletes = new HashMap<Node,List<DeleteRequest>>();
这两个字段主要是实现用于缓存更新请求
执行缓存的请求，调用下面方法
[java] view
plain copy
publicvoid finish() {
// piggyback on any outstanding adds or deletes if possible.
flushAdds(1);
flushDeletes(1);
checkResponses(true);
}
提交请求
[java] view
plain copy
void submit(UpdateRequestExt ureq, Node node) {
Request sreq = new Request();
sreq.node = node;
sreq.ureq = ureq;
submit(sreq);
}
然后是检查响应结果，调用
void checkResponses(booleanblock) 作为检查上一次提交的请求响应。当请求需要重试的时候，会默认重试最大次数10次
将最终结果返回到响应结果里，有异常也会记录下来。
分布式增加更新
[java] view
plain copy
publicvoid distribAdd(AddUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params) throws IOException {
//执行前都会去掉之前还会检查可能没响应完的请求，不等待，直接删除旧的请求。
checkResponses(false);
// 确保所有删除的请求被执行
flushDeletes(1);
//克隆更新请求重用
AddUpdateCommand clone = new AddUpdateCommand(null);
clone.solrDoc = cmd.solrDoc;
clone.commitWithin = cmd.commitWithin;
clone.overwrite = cmd.overwrite;
clone.setVersion(cmd.getVersion());
AddRequest addRequest = new AddRequest();
addRequest.cmd = clone;
addRequest.params = params;
//增加对每个节点的请求到缓存adds里
for (Nodenode : nodes) {
List<AddRequest> alist = adds.get(node);
if (alist == null) {
alist = new ArrayList<AddRequest>(2);
adds.put(node, alist);
}
alist.add(addRequest);
}
//执行缓存adds的请求
flushAdds(maxBufferedAddsPerServer);
}
其它的doDelete,addCommit的请求逻辑的处理都相差不多
DistributedUpdateProcessor
这个是solrCloud主要的一个更新处理链，使用cloud模式的时候必要的一个处理链，负责分布式更新的逻辑处理
一个重要的hash算法，作为更新记录具体分配到哪个shard的算法
这算法应该会在后期重构并设计为插件方式 ，可被用户自定议的hash算法替换。
[java] view
plain copy
private int hash(AddUpdateCommandcmd) {
String hashableId = cmd.getHashableId();
return Hash.murmurhash3_x86_32(hashableId, 0, hashableId.length(), 0);
}
private int hash(DeleteUpdateCommandcmd) {
return Hash.murmurhash3_x86_32(cmd.getId(), 0, cmd.getId().length(), 0);
}
其中cmd.getHashableId()方法返回的主要是文档的主键的值
通过hash值定位更新到哪个shard
[java] view
plain copy
private String getShard(int hash, String collection, ClusterState clusterState) {
return clusterState.getShard(hash, collection);
}
通过取到collection对应的RangeInfo，计算该hash值座落在哪个Range，就可以计算到相应的shard
[java] view
plain copy
public String getShard(int hash, String collection) {
RangeInfo rangInfo = getRanges(collection);
int cnt = 0;
for (Range range : rangInfo.ranges) {
if (range.includes(hash)) {
return rangInfo.shardList.get(cnt);
}
cnt++;
}
throw new IllegalStateException("The HashPartitioner failed");
}
HashPartitioner
做为切分为多个范围的Range，主要实现方法如下：
[java] view
plain copy
public List<Range> partitionRange(int partitions, int min, int max) {
assert max >= min;
if (partitions == 0) return Collections.EMPTY_LIST;
long range = (long)max - (long)min;
long srange = Math.max(1, range / partitions);
List<Range> ranges = new ArrayList<Range>(partitions);
long start = min;
long end = start;
while (end < max) {
end = start + srange;
// make last range always end exactly on MAX_VALUE
if (ranges.size() == partitions - 1) {
end = max;
}
ranges.add(new Range((int)start, (int)end));
start = end + 1L;
}
return ranges;
}
指定了某个范围[min,max]切分为多个partitions的Ranges;切分的范围是按平均的切分。
Range类封装了主键hash值范围【min,max】
RangeInfo封装一个collection下所有shard信息对应的Range,主要实现方法如下：
[java] view
plain copy
private RangeInfo addRangeInfo(String collection) {
List<Range> ranges;
RangeInfo rangeInfo= new RangeInfo();
Map<String,Slice> slices = getSlices(collection);
if (slices == null) {
throw new SolrException(ErrorCode.BAD_REQUEST, "Can not find collection "
+ collection + " in " + this);
}
Set<String> shards = slices.keySet();
ArrayList<String> shardList = new ArrayList<String>(shards.size());
shardList.addAll(shards);
Collections.sort(shardList);
ranges = hp.partitionRange(shards.size(), Integer.MIN_VALUE, Integer.MAX_VALUE);
rangeInfo.ranges = ranges;
rangeInfo.shardList = shardList;
rangeInfos.put(collection, rangeInfo);
return rangeInfo;
}
从上面方法的实现可以看到，会先将所有shard的名称排序，然后根据shard的大小切分相应的多个的范围 ，每一个shard在排序完的位置 有对应的范围Range，两者的信息存放在RangeInfo.
不用担心，上面按整数最小值 ，最大值的平均切分的范围会导致分配不匀的情况，
可能你会担心如果我的主键值是整数，那主键的hash值会不会跟他的值所对应呢，这样的话，会让hash出来的数据先填满小的shard，其它shard不够匀称。其实设计者本身使用的hash算法是针对任何类型，取的主键值也是以字节数组去做hash。这个可以自己使用它的hash算法去校验。
再来看一下DistributedUpdateProcessor
先看add请求，请求的来源有多种：
privateList<Node> setupRequest(inthash)
此方法就是为了判断上面请求来源而决定分发的结点
请求来自leader转发：FROMLEADER，那么就只需要写到本地ulog,不需要转发给leader，也不需要转发给其它replicas
请求不是来自leader，但自己就是leader，那么就需要将请求写到本地，顺便分发给其他的replicas.
请求不是来自leader，但自己又不是leader,也就是该更新请求是最原始的更新请求，那么需要将请求写到本地ulog，顺便转发给leader,再由leader分发
所以为了不让更新请求不会转发来转发去。提交索引的时候，只提交给所有leader是最佳选择。
也就是能预先知道该数据 是要到哪个leader，这个solrj好像有实现。solrcloudserver，分对更新的数据预先做分发请求。
先来讲一下增加的更新逻辑
[java] view
plain copy
@Override
public void processAdd(AddUpdateCommand cmd) throws IOException {
int hash = 0;
if (zkEnabled) {//cloud模式下
zkCheck();//检查zk连接状态
hash = hash(cmd);//取得更新请求hash值，再决定hash到哪一个shard
//判断更新请求来源，决定需要转发的nodes
nodes = setupRequest(hash);
} else {
isLeader = getNonZkLeaderAssumption(req);
}
boolean dropCmd = false;
if (!forwardToLeader) {//不需要转发，即是请求来源是由leader转发来的，或者本人就是leader
dropCmd = versionAdd(cmd);//决定该请求是否丢弃，丢弃原因：1）只需要更新到ulog，也就是请求是来源leader转发来的，不需写到索引。2）ulog已有该更新记录且版本比当前的版本更新，则可以丢弃。
}
// 可以丢弃该请求，不需要写到lucene索引
if (dropCmd) {
// TODO: do we need to add anything to the response?
return;
ModifiableSolrParams params = null;
if (nodes != null) {
params = new ModifiableSolrParams(filterParams(req.getParams()));
}
//需要转发该请求，转发leader,或者由leader分发
params.set(DISTRIB_UPDATE_PARAM,
(isLeader ?
DistribPhase.FROMLEADER.toString() :
DistribPhase.TOLEADER.toString()));
if (isLeader) {
params.set("distrib.from", ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
}
params.set("distrib.from", ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
//转发请求到nodes
cmdDistrib.distribAdd(cmd, nodes, params);
}
//增加更新的响应内容
if (returnVersions && rsp != null && idField != null) {
if (addsResponse == null) {
addsResponse = new NamedList<String>();
rsp.add("adds",addsResponse);
}
if (scratch == null) scratch = new CharsRef();
idField.getType().indexedToReadable(cmd.getIndexedId(), scratch);
addsResponse.add(scratch.toString(), cmd.getVersion());
}
}
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- solrCloud 索引更新逻辑学习笔记
- solr学习文档之增量更新索引
- 2.myql数据导入到solr，并建立solr索引（学习笔记）
- Solr学习笔记三--更新文档搜索
- Solr学习笔记之3、Solr dataimport - 从SQLServer导入数据建立索引
- Solr学习笔记四--solrj的使用（更新文档）
- MATLAB学习笔记之-----数组、矩阵、向量索引和逻辑索引
- solr笔记--solr3.2以后支持document和json两种对象来更新索引
- solrCloud Update Request Handling 更新索引流程
- solr笔记--solr3.2以后支持document和json两种对象来更新索引
- solr学习第九课程----SolrCloud索引分布介绍
- Python学习笔记（2014-01-13 23:53:58更新）
- solr学习笔记
- spring cloud eureka 学习笔记
- Red Hat Enterprise Linux 5学习笔记（持续更新...）
- Swift学习笔记之逻辑分支与循环体
- 我的 CSDN 博客目录索引（主要记录了我学习视频、书籍的笔记，持续更新中）
- 《零基础入门学习Python》学习过程笔记【012列表的常用函数，逻辑关系，+，*，in，列表推导式】
- Oracle 学习笔记10 —— 逻辑控制2（循环）
- python学习笔记04（游戏点球大战、list索引与切片）