Consistent Hashing Ring
基本上只要做Cluster，都會使用到一致性Hash環(huán)，具體作用此處就不細講诅病，我們只了解HiveMQ怎么用它，怎么實現(xiàn)它，這樣實現(xiàn)能夠帶來什么好處贤笆。
HiveMQ沒有Master/Slave,它只由JGroup View(詳情請查閱JGroup)第一個node作為Coordinator蝇棉，這樣就可以達到一個node也可以做集群(雖然這樣的集群沒有什么卵用)。
HiveMQ采用兩個一致性Hash環(huán),來解決腦裂問題芥永，以及腦裂后merge的問題篡殷。
每個node 500個虛擬節(jié)點，來增加node變化帶來的動蕩問題埋涧。
Primary環(huán):排除joining的node,即只添加RUNNING狀態(tài)的node板辽。
Minority環(huán):包含joining的node,即添加JOINING、RUNNING棘催、MERGING狀態(tài)的node劲弦。
它的hash算法由net.openhft.hashing.LongHashFunction.xx_r39()提供
ConsistentHashingRing源碼
相對來說比較簡單，我就不一行一行寫注釋了醇坝，網(wǎng)上針對一致性hash環(huán)實現(xiàn)各種版本到處都是邑跪，詳細講解也到處都是。
@Singleton
public class ConsistentHashingRing {
private static final Logger LOGGER = LoggerFactory.getLogger(ConsistentHashingRing.class);
private final String name;
public static final int NODE_BUCKET_COUNT = 500;
private final LongHashFunction hashFunction;
private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(true);
@VisibleForTesting
final NavigableMap<Long, String> buckets;
@VisibleForTesting
final ConcurrentHashMap<String, String> bucketNodes = new ConcurrentHashMap<>();
final Set<String> nodes = Sets.newConcurrentHashSet();
public ConsistentHashingRing(String name, LongHashFunction hashFunction) {
this.name = name;
this.buckets = new ConcurrentSkipListMap();
this.hashFunction = hashFunction;
}
public void add(@NotNull String node) {
Preconditions.checkNotNull(node, "Name must not be null");
LOGGER.trace("Add node {} to the {}.", node, this.name);
Lock lock = this.readWriteLock.writeLock();
lock.lock();
try {
for (int bucketIndex = 0; bucketIndex < NODE_BUCKET_COUNT; bucketIndex++) {
long bucketHash = this.hashFunction.hashChars(node + bucketIndex);
if (this.buckets.containsKey(bucketHash)) {
if (this.buckets.get(bucketHash).compareTo(node + 1) > 0) {
this.buckets.put(bucketHash, node + bucketIndex);
this.nodes.add(node);
this.bucketNodes.put(node + bucketIndex, node);
}
} else {
this.buckets.put(bucketHash, node + bucketIndex);
this.nodes.add(node);
this.bucketNodes.put(node + bucketIndex, node);
}
}
} finally {
lock.unlock();
}
}
public void remove(@NotNull String node) {
Preconditions.checkNotNull(node, "Name must not be null");
LOGGER.trace("Remove node {} from the {}.", node, this.name);
Lock lock = this.readWriteLock.writeLock();
lock.lock();
try {
for (int bucketIndex = 0; bucketIndex < NODE_BUCKET_COUNT; bucketIndex++) {
long bucketHash = this.hashFunction.hashChars(node + bucketIndex);
this.buckets.remove(bucketHash);
this.bucketNodes.remove(node + bucketIndex);
}
this.nodes.remove(node);
} finally {
lock.unlock();
}
}
public Set<String> getReplicaNodes(@NotNull String key, int replicateCount) {
Preconditions.checkNotNull(key, "key must not be null");
int nodeCount = this.nodes.size();
if (replicateCount > nodeCount - 1) {
LOGGER.trace("There are not enough buckets in the consistent hash ring for {} replicas.", replicateCount);
replicateCount = nodeCount - 1;
}
String bucket = getBucket(key);
long bucketHash = this.hashFunction.hashChars(bucket);
Lock lock = this.readWriteLock.readLock();
lock.lock();
Set<String> buckets = new HashSet<>();
try {
for (Map.Entry<Long, String> entry = this.buckets.higherEntry(bucketHash);
buckets.size() < replicateCount;
entry = this.buckets.higherEntry(entry.getKey())) {
if (entry == null) {
entry = this.buckets.firstEntry();
}
if (!this.bucketNodes.get(entry.getValue()).equals(this.bucketNodes.get(bucket))) {
buckets.add(this.bucketNodes.get(entry.getValue()));
}
}
return buckets;
} finally {
lock.unlock();
}
}
public Set<String> getNodes() {
ImmutableSet.Builder<String> builder = ImmutableSet.builder();
Lock lock = this.readWriteLock.readLock();
lock.lock();
try {
return builder.addAll(this.nodes).build();
} finally {
lock.unlock();
}
}
public String getBucket(@NotNull String key) {
Preconditions.checkNotNull(key, "key must not be null");
if (this.buckets.isEmpty()) {
throw new IllegalStateException("Consistent hash ring is empty.");
}
long keyHash = this.hashFunction.hashChars(key);
Lock lock = this.readWriteLock.readLock();
lock.lock();
try {
Map.Entry<Long, String> entry = this.buckets.ceilingEntry(keyHash);
if (entry != null) {
return entry.getValue();
}
return this.buckets.ceilingEntry(Long.MIN_VALUE).getValue();
} finally {
lock.unlock();
}
}
public String getNode(@NotNull String key) {
Preconditions.checkNotNull(key, "key must not be null");
if (this.buckets.isEmpty()) {
throw new IllegalStateException("Consistent hash ring is empty.");
}
long keyHash = this.hashFunction.hashChars(key);
Lock lock = this.readWriteLock.readLock();
lock.lock();
try {
Map.Entry<Long, String> entry = this.buckets.ceilingEntry(keyHash);
if (entry != null) {
return this.bucketNodes.get(entry.getValue());
}
return this.bucketNodes.get(this.buckets.ceilingEntry(Long.MIN_VALUE).getValue());
} finally {
lock.unlock();
}
}
}
Node Lifecycle
其實了解了上面HiveMQ Cluster的基礎之后呼猪，再來看node的生命周期画畅，就是一件簡單的事情了。
廢話少說郑叠，我們直接上狀態(tài)變化圖夜赵。
這里寫圖片描述
各種狀態(tài)簡介
UNKNOWN
當JGroup通知新的node連接,但在本地不存在,則該node狀態(tài)標記為UNKNOWN
NOT_JOINED
當node連接上JGroup后,若它不是唯一的node,則它將自己主動標記為NOT_JOINED
JOINING
當node將自己的狀態(tài)更新至Cluster完成后,它將自己主動標記為JOINING
MERGE_MINORITY
當腦裂后與Coordinator在同組的其他node都將被標記為MERGE_MINORITY;或者加入Primary Group失敗后它將自己主動標記為MERGE_MINORITY
MERGING
MERGE_MINORITY會一直去嘗試主動將自己標記為MERGING
RUNNING
當MERGING成功后,node將會進行Replicate操作,當Replicate操作完成,就主動將自己標記為RUNNING
SHUTTING_DOWN/SHUTDOWN_FINISHED/DEAD
這三種狀態(tài)在源碼中未被使用,但HiveMQ還這樣定義，或許是保留吧乡革，反正博主未搞懂，不過不重要摊腋，不懂就算了沸版，_。
