序

本文主要研究一下scalecube-cluster的GossipProtocol

GossipProtocol

scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocol.java

/*** Gossip Protocol component responsible for spreading information (gossips) over the cluster* members using infection-style dissemination algorithms. It provides reliable cross-cluster* broadcast.*/
public interface GossipProtocol {/** Starts running gossip protocol. After started it begins to receive and send gossip messages */void start();/** Stops running gossip protocol and releases occupied resources. */void stop();/*** Spreads given message between cluster members.** @return future result with gossip id once gossip fully spread.*/Mono<String> spread(Message message);/** Listens for gossips from other cluster members. */Flux<Message> listen();
}

• GossipProtocol接口定义了start、stop、spread、listen方法

GossipProtocolImpl

scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java

public final class GossipProtocolImpl implements GossipProtocol {private static final Logger LOGGER = LoggerFactory.getLogger(GossipProtocolImpl.class);// Qualifierspublic static final String GOSSIP_REQ = "sc/gossip/req";// Injectedprivate final Member localMember;private final Transport transport;private final GossipConfig config;// Local Stateprivate long currentPeriod = 0;private long gossipCounter = 0;private Map<String, GossipState> gossips = new HashMap<>();private Map<String, MonoSink<String>> futures = new HashMap<>();private List<Member> remoteMembers = new ArrayList<>();private int remoteMembersIndex = -1;// Disposablesprivate final Disposable.Composite actionsDisposables = Disposables.composite();// Subjectprivate final FluxProcessor<Message, Message> subject =DirectProcessor.<Message>create().serialize();private final FluxSink<Message> sink = subject.sink();// Scheduledprivate final Scheduler scheduler;/*** Creates new instance of gossip protocol with given memberId, transport and settings.** @param localMember local cluster member* @param transport cluster transport* @param membershipProcessor membership event processor* @param config gossip protocol settings* @param scheduler scheduler*/public GossipProtocolImpl(Member localMember,Transport transport,Flux<MembershipEvent> membershipProcessor,GossipConfig config,Scheduler scheduler) {this.transport = Objects.requireNonNull(transport);this.config = Objects.requireNonNull(config);this.localMember = Objects.requireNonNull(localMember);this.scheduler = Objects.requireNonNull(scheduler);// SubscribeactionsDisposables.addAll(Arrays.asList(membershipProcessor //.publishOn(scheduler).subscribe(this::onMemberEvent, this::onError),transport.listen().publishOn(scheduler).filter(this::isGossipReq).subscribe(this::onGossipReq, this::onError)));}@Overridepublic void start() {actionsDisposables.add(scheduler.schedulePeriodically(this::doSpreadGossip,config.getGossipInterval(),config.getGossipInterval(),TimeUnit.MILLISECONDS));}@Overridepublic void stop() {// Stop accepting gossip requests and spreading gossipsactionsDisposables.dispose();// Stop publishing eventssink.complete();}@Overridepublic Mono<String> spread(Message message) {return Mono.fromCallable(() -> message).subscribeOn(scheduler).flatMap(msg -> Mono.create(sink -> futures.put(createAndPutGossip(msg), sink)));}@Overridepublic Flux<Message> listen() {return subject.onBackpressureBuffer();}private void onMemberEvent(MembershipEvent event) {Member member = event.member();if (event.isRemoved()) {remoteMembers.remove(member);}if (event.isAdded()) {remoteMembers.add(member);}}private void onGossipReq(Message message) {long period = this.currentPeriod;GossipRequest gossipRequest = message.data();for (Gossip gossip : gossipRequest.gossips()) {GossipState gossipState = gossips.get(gossip.gossipId());if (gossipState == null) { // new gossipgossipState = new GossipState(gossip, period);gossips.put(gossip.gossipId(), gossipState);sink.next(gossip.message());}gossipState.addToInfected(gossipRequest.from());}}private boolean isGossipReq(Message message) {return GOSSIP_REQ.equals(message.qualifier());}private String createAndPutGossip(Message message) {long period = this.currentPeriod;Gossip gossip = new Gossip(generateGossipId(), message);GossipState gossipState = new GossipState(gossip, period);gossips.put(gossip.gossipId(), gossipState);return gossip.gossipId();}//......}

• GossipProtocolImpl实现了GossipProtocol接口，它维护了名为gossips的gossipId与GossipState的map，以及remoteMembers列表
• 它的构造器订阅了membershipProcessor，触发onMemberEvent方法，该方法根据MembershipEvent来对remoteMembers进行添加或移除member；订阅了transport.listen()，过滤出GossipReq，触发onGossipReq方法，该方法合并GossipRequest的gossips到本地的gossips，对于新的gossip的message则发送到sink，并维护该gossip的gossipState，将请求的memberId添加到infected中；spread方法则将message放入到本地的gossips中
• start方法每隔gossipInterval执行doSpreadGossip方法；spread方法则通过createAndPutGossip创建Gossip并放入gossips中

doSpreadGossip

scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java

public final class GossipProtocolImpl implements GossipProtocol {//......private List<Member> remoteMembers = new ArrayList<>();private int remoteMembersIndex = -1;private void doSpreadGossip() {// Increment periodlong period = currentPeriod++;// Check any gossips existsif (gossips.isEmpty()) {return; // nothing to spread}try {// Spread gossips to randomly selected member(s)selectGossipMembers().forEach(member -> spreadGossipsTo(period, member));// Sweep gossipssweepGossips(period);} catch (Exception ex) {LOGGER.warn("Exception at doSpreadGossip[{}]: {}", period, ex.getMessage(), ex);}}private void spreadGossipsTo(long period, Member member) {// Select gossips to sendList<Gossip> gossips = selectGossipsToSend(period, member);if (gossips.isEmpty()) {return; // nothing to spread}// Send gossip requestAddress address = member.address();gossips.stream().map(this::buildGossipRequestMessage).forEach(message ->transport.send(address, message).subscribe(null,ex ->LOGGER.debug("Failed to send GossipReq[{}]: {} to {}, cause: {}",period,message,address,ex.toString())));}private List<Gossip> selectGossipsToSend(long period, Member member) {int periodsToSpread =ClusterMath.gossipPeriodsToSpread(config.getGossipRepeatMult(), remoteMembers.size() + 1);return gossips.values().stream().filter(gossipState -> gossipState.infectionPeriod() + periodsToSpread >= period) // max rounds.filter(gossipState -> !gossipState.isInfected(member.id())) // already infected.map(GossipState::gossip).collect(Collectors.toList());}private List<Member> selectGossipMembers() {int gossipFanout = config.getGossipFanout();if (remoteMembers.size() < gossipFanout) { // select allreturn remoteMembers;} else { // select random members// Shuffle members initially and once reached top boundif (remoteMembersIndex < 0 || remoteMembersIndex + gossipFanout > remoteMembers.size()) {Collections.shuffle(remoteMembers);remoteMembersIndex = 0;}// Select membersList<Member> selectedMembers =gossipFanout == 1? Collections.singletonList(remoteMembers.get(remoteMembersIndex)): remoteMembers.subList(remoteMembersIndex, remoteMembersIndex + gossipFanout);// Increment index and return resultremoteMembersIndex += gossipFanout;return selectedMembers;}}private Message buildGossipRequestMessage(Gossip gossip) {GossipRequest gossipRequest = new GossipRequest(gossip, localMember.id());return Message.withData(gossipRequest).qualifier(GOSSIP_REQ).sender(localMember.address()).build();}private void sweepGossips(long period) {// Select gossips to sweepint periodsToSweep =ClusterMath.gossipPeriodsToSweep(config.getGossipRepeatMult(), remoteMembers.size() + 1);Set<GossipState> gossipsToRemove =gossips.values().stream().filter(gossipState -> period > gossipState.infectionPeriod() + periodsToSweep).collect(Collectors.toSet());// Check if anything selectedif (gossipsToRemove.isEmpty()) {return; // nothing to sweep}// Sweep gossipsLOGGER.debug("Sweep gossips[{}]: {}", period, gossipsToRemove);for (GossipState gossipState : gossipsToRemove) {gossips.remove(gossipState.gossip().gossipId());MonoSink<String> sink = futures.remove(gossipState.gossip().gossipId());if (sink != null) {sink.success(gossipState.gossip().gossipId());}}}//......}

• doSpreadGossip方法首先递增currentPeriod，然后执行selectGossipMembers，遍历该member执行spreadGossipsTo，最后执行sweepGossips
• selectGossipMembers方法会根据gossipFanout配置随机选择gossipFanout个member，这里维护了remoteMembersIndex，具体是对remoteMembers进行subList，当remoteMembersIndex小于0或remoteMembersIndex + gossipFanout > remoteMembers.size()时会Collections.shuffle(remoteMembers)并重置remoteMembersIndex为0，之后对remoteMembersIndex加上gossipFanout
• spreadGossipsTo方法首先执行selectGossipsToSend获取要发送的gossips，然后通过buildGossipRequestMessage构造GOSSIP_REQ消息，最后通过transport.send方法发送
• sweepGossips方法则选取periodsToSweep，然后从gossips移除period > gossipState.infectionPeriod() + periodsToSweep的gossipState

小结

• GossipProtocol接口定义了start、stop、spread、listen方法；GossipProtocolImpl实现了GossipProtocol接口，它维护了名为gossips的gossipId与GossipState的map，以及remoteMembers列表
• GossipProtocolImpl的构造器订阅了membershipProcessor，触发onMemberEvent方法，该方法根据MembershipEvent来对remoteMembers进行添加或移除member；订阅了transport.listen()，过滤出GossipReq，触发onGossipReq方法，该方法合并GossipRequest的gossips到本地的gossips，对于新的gossip的message则发送到sink，并维护该gossip的gossipState，将请求的memberId添加到infected中；spread方法则将message放入到本地的gossips中
• GossipProtocolImpl的start方法每隔gossipInterval执行doSpreadGossip方法；spread方法则通过createAndPutGossip创建Gossip并放入gossips中；doSpreadGossip方法首先递增currentPeriod，然后执行selectGossipMembers，遍历该member执行spreadGossipsTo，最后执行sweepGossips

这里GossipProtocolImpl注册了onMemberEvent及onGossipReq，其中onMemberEvent用于监听MembershipEvent，并根据该event来维护remoteMembers列表；onGossipReq则是监听其他member的doSpreadGossip方法发送过来的GossipReq消息，合并该消息的gossips到本地的gossips；而doSpreadGossip方法则是每隔gossipInterval执行，根据gossipFanout配置随机选择gossipFanout个member，然后针对每个member选择要发送的gossips进行spread(onGossipReq及spread方法会更改gossips，而每隔gossipInterval触发的doSpreadGossip则从gossips选择待spread的消息进行发送)

doc

• Gossip
• GossipProtocol
• GossipProtocolImpl