序
本文主要研究一下 scalecube-cluster 的 MembershipProtocol
MembershipProtocol
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocol.java
/**
* Cluster Membership Protocol component responsible for managing information about existing members
* of the cluster.
*/
public interface MembershipProtocol {
/**
* Starts running cluster membership protocol. After started it begins to receive and send cluster
* membership messages
*/
Mono<Void> start();
/** Stops running cluster membership protocol and releases occupied resources. */
void stop();
/** Listen changes in cluster membership. */
Flux<MembershipEvent> listen();
/**
* Returns list of all members of the joined cluster. This will include all cluster members
* including local member.
*
* @return all members in the cluster (including local one)
*/
Collection<Member> members();
/**
* Returns list of all cluster members of the joined cluster excluding local member.
*
* @return all members in the cluster (excluding local one)
*/
Collection<Member> otherMembers();
/**
* Returns local cluster member which corresponds to this cluster instance.
*
* @return local member
*/
Member member();
/**
* Returns cluster member with given id or null if no member with such id exists at joined
* cluster.
*
* @return member by id
*/
Optional<Member> member(String id);
/**
* Returns cluster member by given address or null if no member with such address exists at joined
* cluster.
*
* @return member by address
*/
Optional<Member> member(Address address);
}- MembershipProtocol 接口定义了 start、stop、listen、members、otherMembers、member 方法
MembershipProtocolImpl
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
public final class MembershipProtocolImpl implements MembershipProtocol {private static final Logger LOGGER = LoggerFactory.getLogger(MembershipProtocolImpl.class);
private enum MembershipUpdateReason {
FAILURE_DETECTOR_EVENT,
MEMBERSHIP_GOSSIP,
SYNC,
INITIAL_SYNC,
SUSPICION_TIMEOUT
}
// Qualifiers
public static final String SYNC = "sc/membership/sync";
public static final String SYNC_ACK = "sc/membership/syncAck";
public static final String MEMBERSHIP_GOSSIP = "sc/membership/gossip";
private final Member localMember;
// Injected
private final Transport transport;
private final MembershipConfig config;
private final List<Address> seedMembers;
private final FailureDetector failureDetector;
private final GossipProtocol gossipProtocol;
private final MetadataStore metadataStore;
private final CorrelationIdGenerator cidGenerator;
// State
private final Map<String, MembershipRecord> membershipTable = new HashMap<>();
private final Map<String, Member> members = new HashMap<>();
// Subject
private final FluxProcessor<MembershipEvent, MembershipEvent> subject =
DirectProcessor.<MembershipEvent>create().serialize();
private final FluxSink<MembershipEvent> sink = subject.sink();
// Disposables
private final Disposable.Composite actionsDisposables = Disposables.composite();
// Scheduled
private final Scheduler scheduler;
private final Map<String, Disposable> suspicionTimeoutTasks = new HashMap<>();
/**
* Creates new instantiates of cluster membership protocol with given transport and config.
*
* @param localMember local cluster member
* @param transport cluster transport
* @param failureDetector failure detector
* @param gossipProtocol gossip protocol
* @param metadataStore metadata store
* @param config membership config parameters
* @param scheduler scheduler
* @param cidGenerator correlation id generator
*/
public MembershipProtocolImpl(
Member localMember,
Transport transport,
FailureDetector failureDetector,
GossipProtocol gossipProtocol,
MetadataStore metadataStore,
MembershipConfig config,
Scheduler scheduler,
CorrelationIdGenerator cidGenerator) {this.transport = Objects.requireNonNull(transport);
this.config = Objects.requireNonNull(config);
this.failureDetector = Objects.requireNonNull(failureDetector);
this.gossipProtocol = Objects.requireNonNull(gossipProtocol);
this.metadataStore = Objects.requireNonNull(metadataStore);
this.localMember = Objects.requireNonNull(localMember);
this.scheduler = Objects.requireNonNull(scheduler);
this.cidGenerator = Objects.requireNonNull(cidGenerator);
// Prepare seeds
seedMembers = cleanUpSeedMembers(config.getSeedMembers());
// Init membership table with local member record
membershipTable.put(localMember.id(), new MembershipRecord(localMember, ALIVE, 0));
// fill in the table of members with local member
members.put(localMember.id(), localMember);
actionsDisposables.addAll(
Arrays.asList(
// Listen to incoming SYNC and SYNC ACK requests from other members
transport
.listen() //
.publishOn(scheduler)
.subscribe(this::onMessage, this::onError),
// Listen to events from failure detector
failureDetector
.listen()
.publishOn(scheduler)
.subscribe(this::onFailureDetectorEvent, this::onError),
// Listen to membership gossips
gossipProtocol
.listen()
.publishOn(scheduler)
.subscribe(this::onMembershipGossip, this::onError)));
}
@Override
public Flux<MembershipEvent> listen() {return subject.onBackpressureBuffer();
}
@Override
public Mono<Void> start() {
// Make initial sync with all seed members
return Mono.create(
sink -> {
// In case no members at the moment just schedule periodic sync
if (seedMembers.isEmpty()) {schedulePeriodicSync();
sink.success();
return;
}
// If seed addresses are specified in config - send initial sync to those nodes
LOGGER.debug("Making initial Sync to all seed members: {}", seedMembers);
//noinspection unchecked
Mono<Message>[] syncs =
seedMembers
.stream()
.map(
address -> {String cid = cidGenerator.nextCid();
return transport
.requestResponse(prepareSyncDataMsg(SYNC, cid), address)
.filter(this::checkSyncGroup);
})
.toArray(Mono[]::new);
// Process initial SyncAck
Flux.mergeDelayError(syncs.length, syncs)
.take(1)
.timeout(Duration.ofMillis(config.getSyncTimeout()), scheduler)
.publishOn(scheduler)
.flatMap(message -> onSyncAck(message, true))
.doFinally(
s -> {schedulePeriodicSync();
sink.success();})
.subscribe(
null,
ex -> LOGGER.info("Exception on initial SyncAck, cause: {}", ex.toString()));
});
}
@Override
public void stop() {
// Stop accepting requests, events and sending sync
actionsDisposables.dispose();
// Cancel remove members tasks
for (String memberId : suspicionTimeoutTasks.keySet()) {Disposable future = suspicionTimeoutTasks.get(memberId);
if (future != null && !future.isDisposed()) {future.dispose();
}
}
suspicionTimeoutTasks.clear();
// Stop publishing events
sink.complete();}
@Override
public Collection<Member> members() {return new ArrayList<>(members.values());
}
@Override
public Collection<Member> otherMembers() {return new ArrayList<>(members.values())
.stream()
.filter(member -> !member.equals(localMember))
.collect(Collectors.toList());
}
@Override
public Member member() {return localMember;}
@Override
public Optional<Member> member(String id) {return Optional.ofNullable(members.get(id));
}
@Override
public Optional<Member> member(Address address) {return new ArrayList<>(members.values())
.stream()
.filter(member -> member.address().equals(address))
.findFirst();}
//......
}- MembershipProtocolImpl 实现了 MembershipProtocol 接口；它定义了 MembershipUpdateReason 枚举(
FAILURE_DETECTOR_EVENT、MEMBERSHIP_GOSSIP、SYNC、INITIAL_SYNC、SUSPICION_TIMEOUT) - MembershipProtocolImpl 的构造器监听了 transport.listen()触发 onMessage 方法；监听了 failureDetector.listen()触发 onFailureDetectorEvent 方法；监听了 gossipProtocol.listen()触发 onMembershipGossip 方法
- MembershipProtocolImpl 的 start 方法在 seedMembers.isEmpty()的时候会执行 schedulePeriodicSync 方法即每隔 syncInterval 执行 doSync 方法；当 seedMembers 不为空时则遍历 seedMembers 通过 transport.requestResponse 发送 SYNC，执行成功则触发 onSyncAck；stop 方法则挨个销毁 suspicionTimeoutTasks 的 future
onMessage
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
public final class MembershipProtocolImpl implements MembershipProtocol {
//......
private void onMessage(Message message) {if (checkSyncGroup(message)) {if (SYNC.equals(message.qualifier())) {onSync(message).subscribe(null, this::onError);
}
if (SYNC_ACK.equals(message.qualifier())) {if (message.correlationId() == null) { // filter out initial sync
onSyncAck(message, false).subscribe(null, this::onError);
}
}
}
}
private boolean checkSyncGroup(Message message) {if (message.data() instanceof SyncData) {SyncData syncData = message.data();
return config.getSyncGroup().equals(syncData.getSyncGroup());
}
return false;
}
/** Merges incoming SYNC data, merges it and sending back merged data with SYNC_ACK. */
private Mono<Void> onSync(Message syncMsg) {
return Mono.defer(() -> {LOGGER.debug("Received Sync: {}", syncMsg);
return syncMembership(syncMsg.data(), false)
.doOnSuccess(
avoid -> {Message message = prepareSyncDataMsg(SYNC_ACK, syncMsg.correlationId());
Address address = syncMsg.sender();
transport
.send(address, message)
.subscribe(
null,
ex ->
LOGGER.debug("Failed to send SyncAck: {} to {}, cause: {}",
message,
address,
ex.toString()));
});
});
}
private Mono<Void> onSyncAck(Message syncAckMsg, boolean onStart) {
return Mono.defer(() -> {LOGGER.debug("Received SyncAck: {}", syncAckMsg);
return syncMembership(syncAckMsg.data(), onStart);
});
}
private Mono<Void> syncMembership(SyncData syncData, boolean onStart) {
return Mono.defer(() -> {
MembershipUpdateReason reason =
onStart ? MembershipUpdateReason.INITIAL_SYNC : MembershipUpdateReason.SYNC;
return Mono.whenDelayError(
syncData
.getMembership()
.stream()
.filter(r1 -> !r1.equals(membershipTable.get(r1.id())))
.map(r1 -> updateMembership(r1, reason))
.toArray(Mono[]::new));
});
}
//......
}- onMessage 方法首先通过 checkSyncGroup 检查一下是不是该 syncGroup 的消息；之后根据 message.qualifier()是 SYNC 则执行 onSync，是 SYNC_ACK 则执行 onSyncAck
- onSync 方法则执行 syncMembership，成功时向 sender 返回 SYNC_ACK 信息；onSyncAck 方法也是调用 syncMembership，只不过没有再向 sender 返回信息
- syncMembership 方法则根据 syncData 的 membership 来挨个执行 updateMembership 方法
onFailureDetectorEvent
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
public final class MembershipProtocolImpl implements MembershipProtocol {
//......
/** Merges FD updates and processes them. */
private void onFailureDetectorEvent(FailureDetectorEvent fdEvent) {MembershipRecord r0 = membershipTable.get(fdEvent.member().id());
if (r0 == null) { // member already removed
return;
}
if (r0.status() == fdEvent.status()) { // status not changed
return;
}
LOGGER.debug("Received status change on failure detector event: {}", fdEvent);
if (fdEvent.status() == ALIVE) {
// TODO: Consider to make more elegant solution
// Alive won't override SUSPECT so issue instead extra sync with member to force it spread
// alive with inc + 1
Message syncMsg = prepareSyncDataMsg(SYNC, null);
Address address = fdEvent.member().address();
transport
.send(address, syncMsg)
.subscribe(
null,
ex ->
LOGGER.debug("Failed to send {} to {}, cause: {}", syncMsg, address, ex.toString()));
} else {
MembershipRecord record =
new MembershipRecord(r0.member(), fdEvent.status(), r0.incarnation());
updateMembership(record, MembershipUpdateReason.FAILURE_DETECTOR_EVENT)
.subscribe(null, this::onError);
}
}
//......
}- onFailureDetectorEvent 方法根据 FailureDetectorEvent 判断该 MembershipRecord 的状态是否有变化，如果变为 ALIVE 则往 fdEvent.member().address()发送 SYNC 信息；否则使用 MembershipUpdateReason.FAILURE_DETECTOR_EVENT 来 updateMembership
onMembershipGossip
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
public final class MembershipProtocolImpl implements MembershipProtocol {
//......
/** Merges received membership gossip (not spreading gossip further). */
private void onMembershipGossip(Message message) {if (MEMBERSHIP_GOSSIP.equals(message.qualifier())) {MembershipRecord record = message.data();
LOGGER.debug("Received membership gossip: {}", record);
updateMembership(record, MembershipUpdateReason.MEMBERSHIP_GOSSIP)
.subscribe(null, this::onError);
}
}
//......
}- onMembershipGossip 方法则针对 message.qualifier()为 MEMBERSHIP_GOSSIP 的消息使用 MembershipUpdateReason.MEMBERSHIP_GOSSIP 来 updateMembership
updateMembership
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
public final class MembershipProtocolImpl implements MembershipProtocol {
//......
/**
* Try to update membership table with the given record.
*
* @param r1 new membership record which compares with existing r0 record
* @param reason indicating the reason for updating membership table
*/
private Mono<Void> updateMembership(MembershipRecord r1, MembershipUpdateReason reason) {
return Mono.defer(() -> {Objects.requireNonNull(r1, "Membership record can't be null");
// Get current record
MembershipRecord r0 = membershipTable.get(r1.id());
// Check if new record r1 overrides existing membership record r0
if (!r1.isOverrides(r0)) {return Mono.empty();
}
// If received updated for local member then increase incarnation and spread Alive gossip
if (r1.member().id().equals(localMember.id())) {int currentIncarnation = Math.max(r0.incarnation(), r1.incarnation());
MembershipRecord r2 =
new MembershipRecord(localMember, r0.status(), currentIncarnation + 1);
membershipTable.put(localMember.id(), r2);
LOGGER.debug("Local membership record r0: {}, but received r1: {},"
+ "spread with increased incarnation r2: {}",
r0,
r1,
r2);
spreadMembershipGossip(r2)
.subscribe(
null,
ex -> {// on-op});
return Mono.empty();}
// Update membership
if (r1.isDead()) {membershipTable.remove(r1.id());
} else {membershipTable.put(r1.id(), r1);
}
// Schedule/cancel suspicion timeout task
if (r1.isSuspect()) {scheduleSuspicionTimeoutTask(r1);
} else {cancelSuspicionTimeoutTask(r1.id());
}
// Emit membership and regardless of result spread gossip
return emitMembershipEvent(r0, r1)
.doFinally(
s -> {// Spread gossip (unless already gossiped)
if (reason != MembershipUpdateReason.MEMBERSHIP_GOSSIP
&& reason != MembershipUpdateReason.INITIAL_SYNC) {spreadMembershipGossip(r1)
.subscribe(
null,
ex -> {// no-op});
}
});
});
}
private Mono<Void> spreadMembershipGossip(MembershipRecord record) {
return Mono.defer(() -> {Message msg = Message.withData(record).qualifier(MEMBERSHIP_GOSSIP).build();
LOGGER.debug("Spead membreship: {} with gossip", msg);
return gossipProtocol
.spread(msg)
.doOnError(
ex ->
LOGGER.debug("Failed to spread membership: {} with gossip, cause: {}",
msg,
ex.toString()))
.then();});
}
private void scheduleSuspicionTimeoutTask(MembershipRecord record) {
long suspicionTimeout =
ClusterMath.suspicionTimeout(config.getSuspicionMult(), membershipTable.size(), config.getPingInterval());
suspicionTimeoutTasks.computeIfAbsent(record.id(),
id ->
scheduler.schedule(() -> onSuspicionTimeout(id), suspicionTimeout, TimeUnit.MILLISECONDS));
}
private void onSuspicionTimeout(String memberId) {suspicionTimeoutTasks.remove(memberId);
MembershipRecord record = membershipTable.get(memberId);
if (record != null) {LOGGER.debug("Declare SUSPECTED member {} as DEAD by timeout", record);
MembershipRecord deadRecord =
new MembershipRecord(record.member(), DEAD, record.incarnation());
updateMembership(deadRecord, MembershipUpdateReason.SUSPICION_TIMEOUT)
.subscribe(null, this::onError);
}
}
private void cancelSuspicionTimeoutTask(String memberId) {Disposable future = suspicionTimeoutTasks.remove(memberId);
if (future != null && !future.isDisposed()) {future.dispose();
}
}
private Mono<Void> emitMembershipEvent(MembershipRecord r0, MembershipRecord r1) {
return Mono.defer(() -> {final Member member = r1.member();
if (r1.isDead()) {members.remove(member.id());
// removed
return Mono.fromRunnable(() -> {Map<String, String> metadata = metadataStore.removeMetadata(member);
sink.next(MembershipEvent.createRemoved(member, metadata));
});
}
if (r0 == null && r1.isAlive()) {members.put(member.id(), member);
// added
return metadataStore
.fetchMetadata(member)
.doOnSuccess(
metadata -> {metadataStore.updateMetadata(member, metadata);
sink.next(MembershipEvent.createAdded(member, metadata));
})
.onErrorResume(TimeoutException.class, e -> Mono.empty())
.then();}
if (r0 != null && r0.incarnation() < r1.incarnation()) {
// updated
return metadataStore
.fetchMetadata(member)
.doOnSuccess(
metadata1 -> {
Map<String, String> metadata0 =
metadataStore.updateMetadata(member, metadata1);
sink.next(MembershipEvent.createUpdated(member, metadata0, metadata1));
})
.onErrorResume(TimeoutException.class, e -> Mono.empty())
.then();}
return Mono.empty();});
}
//......
}- updateMembership 会对比传入的 MembershipRecord 与本地的 localMember，如果是需要更新 localMember 则执行 spreadMembershipGossip，之后根据 MembershipRecord 的状态做不同处理，比如 isDead 则从 membershipTable 移除，比如 isSuspect 则执行 scheduleSuspicionTimeoutTask，否则执行 cancelSuspicionTimeoutTask，最后执行 emitMembershipEvent 及 spreadMembershipGossip
- scheduleSuspicionTimeoutTask 方法计算 suspicionTimeout 然后注册一个 SuspicionTimeout 的延时任务如果 suspicionTimeoutTasks 没有该 record.id()的 task 的话；onSuspicionTimeout 首先将该 task 从 suspicionTimeoutTasks 移除，然后使用 MembershipUpdateReason.SUSPICION_TIMEOUT 来 updateMembership；cancelSuspicionTimeoutTask 方法也是将该 task 从 suspicionTimeoutTasks 移除，并 dispose 该 future
- emitMembershipEvent 方法这里主要是更新 member 在 metadataStore 的 Metadata，如果是 isDead 则执行 metadataStore.removeMetadata(member)，其他的则看情况执行 metadataStore.updateMetadata(member, metadata)
schedulePeriodicSync
scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
public final class MembershipProtocolImpl implements MembershipProtocol {
//......
private void schedulePeriodicSync() {int syncInterval = config.getSyncInterval();
actionsDisposables.add(
scheduler.schedulePeriodically(this::doSync, syncInterval, syncInterval, TimeUnit.MILLISECONDS));
}
private void doSync() {Optional<Address> addressOptional = selectSyncAddress();
if (!addressOptional.isPresent()) {return;}
Address address = addressOptional.get();
Message message = prepareSyncDataMsg(SYNC, null);
LOGGER.debug("Send Sync: {} to {}", message, address);
transport
.send(address, message)
.subscribe(
null,
ex ->
LOGGER.debug("Failed to send Sync: {} to {}, cause: {}", message, address, ex.toString()));
}
private Optional<Address> selectSyncAddress() {
List<Address> addresses =
Stream.concat(seedMembers.stream(), otherMembers().stream().map(Member::address))
.collect(Collectors.collectingAndThen(Collectors.toSet(), ArrayList::new));
Collections.shuffle(addresses);
if (addresses.isEmpty()) {return Optional.empty();
} else {int i = ThreadLocalRandom.current().nextInt(addresses.size());
return Optional.of(addresses.get(i));
}
}
private Message prepareSyncDataMsg(String qualifier, String cid) {List<MembershipRecord> membershipRecords = new ArrayList<>(membershipTable.values());
SyncData syncData = new SyncData(membershipRecords, config.getSyncGroup());
return Message.withData(syncData)
.qualifier(qualifier)
.correlationId(cid)
.sender(localMember.address())
.build();}
//......
}- schedulePeriodically 会注册 doSync 任务每隔 syncInterval 执行；doSync 方法首先调用 selectSyncAddress 随机选择一个 member 来作为发送 SYNC 的目标，之后通过 prepareSyncDataMsg 构造 sync 消息，然后通过 transport.send 发送
小结
- MembershipProtocol 接口定义了 start、stop、listen、members、otherMembers、member 方法；MembershipProtocolImpl 实现了 MembershipProtocol 接口；它定义了 MembershipUpdateReason 枚举(
FAILURE_DETECTOR_EVENT、MEMBERSHIP_GOSSIP、SYNC、INITIAL_SYNC、SUSPICION_TIMEOUT) - MembershipProtocolImpl 的构造器监听了 transport.listen()触发 onMessage 方法；监听了 failureDetector.listen()触发 onFailureDetectorEvent 方法；监听了 gossipProtocol.listen()触发 onMembershipGossip 方法；MembershipProtocolImpl 的 start 方法在 seedMembers.isEmpty()的时候会执行 schedulePeriodicSync 方法即每隔 syncInterval 执行 doSync 方法；当 seedMembers 不为空时则遍历 seedMembers 通过 transport.requestResponse 发送 SYNC，执行成功则触发 onSyncAck；stop 方法则挨个销毁 suspicionTimeoutTasks 的 future
- onMessage 方法首先通过 checkSyncGroup 检查一下是不是该 syncGroup 的消息；之后根据 message.qualifier()是 SYNC 则执行 onSync，是 SYNC_ACK 则执行 onSyncAck；onSync 方法则执行 syncMembership，成功时向 sender 返回 SYNC_ACK 信息；onSyncAck 方法也是调用 syncMembership，只不过没有再向 sender 返回信息；syncMembership 方法则根据 syncData 的 membership 来挨个执行 updateMembership 方法
- onFailureDetectorEvent 方法根据 FailureDetectorEvent 判断该 MembershipRecord 的状态是否有变化，如果变为 ALIVE 则往 fdEvent.member().address()发送 SYNC 信息；否则使用 MembershipUpdateReason.FAILURE_DETECTOR_EVENT 来 updateMembership；onMembershipGossip 方法则针对 message.qualifier()为 MEMBERSHIP_GOSSIP 的消息使用 MembershipUpdateReason.MEMBERSHIP_GOSSIP 来 updateMembership
- updateMembership 会对比传入的 MembershipRecord 与本地的 localMember，如果是需要更新 localMember 则执行 spreadMembershipGossip，之后根据 MembershipRecord 的状态做不同处理，比如 isDead 则从 membershipTable 移除，比如 isSuspect 则执行 scheduleSuspicionTimeoutTask，否则执行 cancelSuspicionTimeoutTask，最后执行 emitMembershipEvent 及 spreadMembershipGossip
- schedulePeriodically 会注册 doSync 任务每隔 syncInterval 执行；doSync 方法首先调用 selectSyncAddress 随机选择一个 member 来作为发送 SYNC 的目标，之后通过 prepareSyncDataMsg 构造 sync 消息，然后通过 transport.send 发送
MembershipProtocolImpl 的 start 方法会注册 doSync 任务 (
每隔 syncInterval 执行)，该任务会发送 SYNC 消息给随机选择出来的 member，来 sync 全量的 membershipRecords；onMessage 方法接收到 SYNC 消息时执行 syncMembership 并在成功时返回 SYNC_ACK，接收到 SYNC_ACK 时也是执行 syncMembership；onFailureDetectorEvent 及 onMembershipGossip 方法都会触发 updateMembership 方法来更新 membershipTable 必要是进行 spreadMembershipGossip
doc
- MembershipProtocol