Solr In Action 笔记(4) 之 SolrCloud Index 基础
SolrCloud Index流程研究了两天，还是没有完全搞懂，先简单记下基础的知识，过几天再写个深入点的。先补充上前文来不及写的内容。
1. Solr.xml的重要配置
Solr.xml的内容如下：
<solr>
<solrcloud>
<str name="host">${host:}</str>
<int name="hostPort">${jetty.port:8983}</int>
<str name="hostContext">${hostContext:solr}</str>
<int name="zkClientTimeout">${zkClientTimeout:15000}</int>
<bool name="genericCoreNodeNames">${genericCoreNodeNames:true}</bool>
</solrcloud>
<shardHandlerFactory name="shardHandlerFactory"
class="HttpShardHandlerFactory">
<int name="socketTimeout">${socketTimeout:0}</int>
<int name="connTimeout">${connTimeout:0}</int>
</shardHandlerFactory>
</solr>
- host ， host 指的是Solr节点的IP地址，当Solr节点上线时候，它会向Zookeeper进行注册，注册信息如IP地址就会存储在/clusterstate.json中。这里不但可以直接使用host IP地址如192.168.1.0，也可以使用机器的hostname比如bigdata01。
- port ， port 指的时Solr用来监听的端口，默认是8983，同样它会存储在/clusterstate.json中。
- Solr Host Context， 指的是Solr.war部署的环境路径，多数情况下不用修改。
- zookeeper client timeout，上一节讲到过，zookeeper Znode节点变化最大反应时间。
- core node name, 该节点控制Solr core的命名策略，如果genericCoreNodeNames为true，那么Solr会给core取普通的名字比如，core_node1 ；如果设为true，则会给core取容易辨别的名字，比如带上host信息，比如10.0.1.7:8983_solr_logmill
- Leader Vote Wait Period：
该参数并未直接在solr.xml中列出来，SolrCloud的leader和其他replica下线只剩最后一个replica的时候，这个Replica并不会立马选举leader，他会等待一段时间，查看leader是否上线，如果上线了，那么leader仍然还是leader，replica仍然还是replica，如果在这个时间段外leader没有上线，那么replica就变为leader了。这个时间就是Leader Vote Wait Period，它的存在防止了当leader和其他replica下线时候，具有旧的数据的node选为leader。
比如以下一个例子，一个shard有两个node，X为leader，Y为replica，如果X在线，Y下线，那么X仍然可以接受update请求，SolrCloud仍然继续正常运行，只不过leader X不需要再把数据分发给Y了，Y上线后X只需要简单将数据同步给Y就行（Peer sync 策略)。如果X下线，Y在线，那么这个时候因为没有leader接受update请求以及没有leader转发数据，Y是不会接收到update请求的，所以这个时候的SolrCloud的所以建立是无法进行的，所以一旦X挂了SolrCloud就会进行leader选举，但是我们不能立马让Y变为leader，因为Y的数据相比较X来说是旧的数据。如果Y选举为Leader了，那么后续的update他就会接受，过段时间X上线了，由于Y已经是leader了所以X只能是replica，数据的流向变成了Y转发到X，这个时候就发现了奇怪的现象就是X中有部分数据新于Y（Y当选为leader前的数据），Y中有部分数据也新于X（Y当选为leader后的数据），这个时候就需要启动Snapshot replication 策略进行数据复原了，比较麻烦。如果设置了leaderVoteWait 那么X下线后，Y会等待leaderVoteWait时间，这个时间内update操作都是失败的，如果在这时间内X上线了，那么X立马恢复leader状态继续工作，否则就会Y就会变成leader。
要改善这种情况，可以增加shard和replica的数量，较少leader和replica同时挂掉的可能性。
- zkHost，同样没有出现在上面的solr.xml上，它可以在solr.xml的zkHost配置中设置zookeepr集群信息比如192.168.0.1:2181,192.168.0.2:2181表示两个zookeeper组成一个zookeeper集群。
2. SolrCloud的分布式建索引
2.1 Document的Hash
建好的SolrCloud集群每一个shard都会有一个Hash区间，当Document进行update的时候，SolrCloud就会计算这个Document的Hash值，然后根据该值和shard的hash区间来判断这个document应该发往哪个shard，所以首先让我们先来学习下SolrCloud的hash算法。Solr使用document route组件来进行document的分发。目前Solr有两个DocRouter类的子类CompositeIdRouter(Solr默认采用的)类和ImplicitDocRouter类，当然我们也可以通过继承DocRouter来定制化我们的document route组件。
之前我们学习过，当Solr Shard建立时候，Solr会给每一个shard分配32bit的hash值的区间，比如SolrCloud有两个shard分别为A,B，那么A的hash值区间就为 80000000-ffffffff ，B的hash值区间为0-7fffffff 。默认的CompositeIdRouter hash策略会根据document ID计算出唯一的Hash值，并判断该值在那个shard的hash区间内。
SolrCloud对于Hash值的获取提出了以下几个要求：
- hash计算速度必须快，因为hash计算是分布式建索引的第一步，SolrCloud不可能在这一不上花很多时间。
- hash值必须能均匀的分布于每一个shard，如果有一个shard的document数量大于另一个shard，那么在查询的时候前一个shard所花的时间就会大于后一个，SolrCloud的查询是先分后汇总的过程，也就是说最后每一个shard查询完毕才算完毕，所以SolrCloud的查询速度是由最慢的shard的查询速度决定的。我们有理由让SolrCloud做好充分的负载均衡。
基于以上两点，SolrCloud采用了MurmurHash 算法，那么让我们先来看下该算法的代码，说实话这个代码我真没看懂，等下次独立写个章节学习下MurmurHash算法吧。
/** Returns the MurmurHash3_x86_32 hash of the UTF-8 bytes of the String without actually encoding
* the string to a temporary buffer. This is more than 2x faster than hashing the result
* of String.getBytes().
*/
public static int murmurhash3_x86_32(CharSequence data, int offset, int len, int seed) { final int c1 = 0xcc9e2d51;
final int c2 = 0x1b873593; int h1 = seed; int pos = offset;
int end = offset + len;
int k1 = 0;
int k2 = 0;
int shift = 0;
int bits = 0;
int nBytes = 0; // length in UTF8 bytes while (pos < end) {
int code = data.charAt(pos++);
if (code < 0x80) {
k2 = code;
bits = 8; /***
// optimized ascii implementation (currently slower!!! code size?)
if (shift == 24) {
k1 = k1 | (code << 24); k1 *= c1;
k1 = (k1 << 15) | (k1 >>> 17); // ROTL32(k1,15);
k1 *= c2; h1 ^= k1;
h1 = (h1 << 13) | (h1 >>> 19); // ROTL32(h1,13);
h1 = h1*5+0xe6546b64; shift = 0;
nBytes += 4;
k1 = 0;
} else {
k1 |= code << shift;
shift += 8;
}
continue;
***/ }
else if (code < 0x800) {
k2 = (0xC0 | (code >> 6))
| ((0x80 | (code & 0x3F)) << 8);
bits = 16;
}
else if (code < 0xD800 || code > 0xDFFF || pos>=end) {
// we check for pos>=end to encode an unpaired surrogate as 3 bytes.
k2 = (0xE0 | (code >> 12))
| ((0x80 | ((code >> 6) & 0x3F)) << 8)
| ((0x80 | (code & 0x3F)) << 16);
bits = 24;
} else {
// surrogate pair
// int utf32 = pos < end ? (int) data.charAt(pos++) : 0;
int utf32 = (int) data.charAt(pos++);
utf32 = ((code - 0xD7C0) << 10) + (utf32 & 0x3FF);
k2 = (0xff & (0xF0 | (utf32 >> 18)))
| ((0x80 | ((utf32 >> 12) & 0x3F))) << 8
| ((0x80 | ((utf32 >> 6) & 0x3F))) << 16
| (0x80 | (utf32 & 0x3F)) << 24;
bits = 32;
} k1 |= k2 << shift; // int used_bits = 32 - shift; // how many bits of k2 were used in k1.
// int unused_bits = bits - used_bits; // (bits-(32-shift)) == bits+shift-32 == bits-newshift shift += bits;
if (shift >= 32) {
// mix after we have a complete word k1 *= c1;
k1 = (k1 << 15) | (k1 >>> 17); // ROTL32(k1,15);
k1 *= c2; h1 ^= k1;
h1 = (h1 << 13) | (h1 >>> 19); // ROTL32(h1,13);
h1 = h1*5+0xe6546b64; shift -= 32;
// unfortunately, java won't let you shift 32 bits off, so we need to check for 0
if (shift != 0) {
k1 = k2 >>> (bits-shift); // bits used == bits - newshift
} else {
k1 = 0;
}
nBytes += 4;
} } // inner // handle tail
if (shift > 0) {
nBytes += shift >> 3;
k1 *= c1;
k1 = (k1 << 15) | (k1 >>> 17); // ROTL32(k1,15);
k1 *= c2;
h1 ^= k1;
} // finalization
h1 ^= nBytes; // fmix(h1);
h1 ^= h1 >>> 16;
h1 *= 0x85ebca6b;
h1 ^= h1 >>> 13;
h1 *= 0xc2b2ae35;
h1 ^= h1 >>> 16; return h1;
}
最后我们再简单地学习下hash计算的源码吧：
- SolrCloud 利用CompositeIdRouter.sliceHash来计算document的hash
public int sliceHash(String id, SolrInputDocument doc, SolrParams params, DocCollection collection) {
String shardFieldName = getRouteField(collection);
if (shardFieldName != null && doc != null) {
Object o = doc.getFieldValue(shardFieldName);
if (o == null)
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No value for :" + shardFieldName + ". Unable to identify shard");
id = o.toString();
}
if (id.indexOf(SEPARATOR) < 0) {
return Hash.murmurhash3_x86_32(id, 0, id.length(), 0);
}
return new KeyParser(id).getHash();
}
- 根据计算出来的hash值计算应该将document发往哪些节点
public Collection<Slice> getSearchSlicesSingle(String shardKey, SolrParams params, DocCollection collection) {
if (shardKey == null) {
// search across whole collection
// TODO: this may need modification in the future when shard splitting could cause an overlap
return collection.getActiveSlices();
}
String id = shardKey;
if (shardKey.indexOf(SEPARATOR) < 0) {
// shardKey is a simple id, so don't do a range
return Collections.singletonList(hashToSlice(Hash.murmurhash3_x86_32(id, 0, id.length(), 0), collection));
}
Range completeRange = new KeyParser(id).getRange();
List<Slice> targetSlices = new ArrayList<>(1);
for (Slice slice : collection.getActiveSlices()) {
Range range = slice.getRange();
if (range != null && range.overlaps(completeRange)) {
targetSlices.add(slice);
}
}
return targetSlices;
}
- 最后，看下SolrCloud是怎么划分shard的hash值区间的。以下代码需要注意几点，
boolean round = rangeStep >= (1 << bits) * 16 判断shard个数是否小于4096个，如果round为true，肯定小于4096个，也就是每个shard的区间长度大于 (1 << bits) * 16
int mask = 0x0000ffff; end & mask != mask 表示判断shard个数不是2的指数次，如果shard个数是2的指数次那么shard的区间肯定是mask的整数倍，也就是说end & mask后最后的16位全为1即0xffff。
end - roundDown < roundUp - end ；当shard个数不是2的指数次时，end离哪个边界近就设置为哪个边界(这里的边界是0x0000ffff的整数倍)。
- 从上面得知，shard的区间得满足0x0000ffff的整数倍
public List<Range> partitionRange(int partitions, Range range) {
int min = range.min;
int max = range.max;
assert max >= min;
if (partitions == 0) return Collections.EMPTY_LIST;
long rangeSize = (long) max - (long) min;
long rangeStep = Math.max(1, rangeSize / partitions);
List<Range> ranges = new ArrayList<>(partitions);
long start = min;
long end = start;
// keep track of the idealized target to avoid accumulating rounding errors
long targetStart = min;
long targetEnd = targetStart;
// Round to avoid splitting hash domains across ranges if such rounding is not significant.
// With default bits==16, one would need to create more than 4000 shards before this
// becomes false by default.
int mask = 0x0000ffff;
boolean round = rangeStep >= (1 << bits) * 16;
while (end < max) {
targetEnd = targetStart + rangeStep;
end = targetEnd;
if (round && ((end & mask) != mask)) {
// round up or down?
int increment = 1 << bits; // 0x00010000
long roundDown = (end | mask) - increment;
long roundUp = (end | mask) + increment;
if (end - roundDown < roundUp - end && roundDown > start) {
end = roundDown;
} else {
end = roundUp;
}
}
// 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;
targetStart = targetEnd + 1L;
}
return ranges;
}
2.2 ADD Document过程
SolrCloud进行update/add document的过程是采用的索引链的方式，暂时我还没看懂，所以本节先不讲代码，大致学习原理以及过程，下节再开一章讲述下Add document的过程。整个过程我们从Solrj客户端讲起。
- 当SolrJ 发送update请求给CloudSolrServer ，CloudSolrServer会连接至Zookeeper获取当前SolrCloud的集群状态，并会在/clusterstate.json 和/live_nodes 注册watcher，便于监视Zookeeper和SolrCloud，这样做的好处有以下几点：
- CloudSolrServer获取到SolrCloud的状态后，它能跟直接将document发往SolrCloud的leader，降低网络转发消耗。
- 注册watcher有利于建索引时候的负载均衡，比如如果有个节点leader下线了，那么CloudSolrServer会立马得知，那它就会停止往下线leader发送document。
- 路由document至正确的shard。CloudSolrServer 在发送document时候需要知道发往哪个shard，这就是上一小节2.1讲过的内容，但是这里需要注意，单个document的路由非常简单，但是SolrCloud支持批量add，也就是说正常情况下N个document同时进行路由。这个时候SolrCloud就会根据document路由的去向分开存放document即进行分类，然后进行并发发送至相应的shard，这就需要较高的并发能力。
- Leader接受到update请求后，先将update信息存放到本地的update log，同时Leader还会给documrnt分配新的version，对于已存在的document，Leader就会验证分配的新version与已有的version，如果新的版本高就会抛弃旧版本，最后发送至replica。
- 一旦document经过验证以及加入version后，就会并行的被转发至所有上线的replica。SolrCloud并不会关注那些已经下线的replica，因为当他们上线时候会有recovery进程对他们进行恢复。如果转发的replica处于recovering状态，那么这个replica就会把update放入update transaction 日志。
- 当leader接受到所有的replica的反馈成功后，它才会反馈客户端成功。只要shard中又一个replica是active的，Solr就会继续接受update请求。这一策略其实是牺牲了一致性换取了写入的有效性。之前我们讲到leaderVoteWait参数，它表示当只有一个replica时候，这个replica会进入recovering状态并持续一段时间等待leader的重新上线。那么如果在这段时间内leader没有上线，那么他就会转成leader会有一些document丢失。(这里我有点不明白，既然leader挂了，难道update 请求还会发送成功？如果成功是发往哪的？) 当然后续会有方法来避免这个情况，比如使用majority quorum 策略，跟Zookeeper的leader选举策略一样，比如当多数的replica下线了，那么客户端的write就会失败。
- 最后的步骤就是commit了，commit有两种，一种是softcommit，即在内存中生成segment，document是可见的(可查询到)但是没有写入磁盘，断电后数据丢失。另一种是hardcommit，直接将数据写入磁盘且数据可见。前一种消耗较少，后一种消耗较大。
大致讲了下SolrCloud的index流程，不是很细致。最后总结以下几点：
- leader转发的规则
- 请求来自leader转发：FROMLEADER，那么就只需要写到本地ulog,不需要转发给leader，也不需要转发给其它replicas。如果replica处于非actibe状态中，就会讲update请求接受并写入ulog，但不会写入索引。如果发现重复的更新就会丢弃旧版本的更新。
- 请求不是来自leader，但自己就是leader，那么就需要将请求写到本地，顺便分发给其他的replicas.
- 请求不是来自leader，但自己又不是leader,也就是该更新请求是最原始的更新请求，那么需要将请求写到本地ulog，顺便转发给leader,再由leader分发
- 每commit一次，就会重新生成一个ulog更新日志，当服务器挂掉，内存数据丢失，就可以从ulog中恢复
- 建索引时候最好使用CloudSolrServer，直接向leader发送update请求避免网络开销
- 批量add document时候，建议在客户端提前做好document的路由，在SolrCloud内进行document开销较大
2.3 NRT 近实时搜索
SolrCloud支持近实时搜索，所谓的近实时搜索即在较短的时间内使得add的document可见可查，这主要基于softcommit机制(Lucene是没有softcommit的，只有hardcommit)。
当进行SoftCommit时候，Solr会打开新的Searcher从而使得新的document可见，同时Solr还会进行预热缓存以及查询以使得缓存的数据也是可见的。所以必须保证预热缓存以及预热查询的执行时间必须短于commit的频率，否则就会由于打开太多的searcher而造成commit失败。
最后说说在工作中近实时搜索的感受吧，近实时搜索是相对的，对于有些客户1分钟就是近实时了，有些3分钟就是近实时了。而对于Solr来说，softcommit越频繁实时性更高，而softcommit越频繁则Solr的负荷越大（commit越频率越会生成小且多的segment，于是merge出现的更频繁）。目前我们公司的softcommit频率是3分钟，之前设置过1分钟而使得Solr在Index所占资源过多大大影响了查询。所以近实时蛮困扰着我们的，因为客户会不停的要求你更加实时，目前公司采用加入缓存机制来弥补这个实时性。
2.4 Node recovery process
Node recovery process 是SolrCloud容灾能力的重要体现，也是我最近研究的重点之一，目前还没得及深入研究，所以一样先看下概念吧。
SolrCloud支持两种recovery策略，Peer sync 和 Snapshot replication ，分别对应类PeerSync和Snapshot ,他们两者是根据节点下线时丢失的update 请求的数量进行区分的。
- Peer sync， 如果中断的时间较短，recovering node只是丢失少量update请求，那么它可以从leader的update log中获取。这个临界值是100个update请求，如果大于100，就会从leader进行完整的索引快照恢复。
- Snapshot replication， 如果节点下线太久以至于不能从leader那进行同步，它就会使用solr的基于http进行索引的快照恢复
- 当你加入新的replica到shard中，它就会进行一个完整的index Snapshot。