序
本文主要研究一下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); // Qualifiers public static final String GOSSIP_REQ = "sc/gossip/req"; // Injected private final Member localMember; private final Transport transport; private final GossipConfig config; // Local State private 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; // Disposables private final Disposable.Composite actionsDisposables = Disposables.composite(); // Subject private final FluxProcessor<Message, Message> subject = DirectProcessor.<Message>create().serialize(); private final FluxSink<Message> sink = subject.sink(); // Scheduled private 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); // Subscribe actionsDisposables.addAll( Arrays.asList( membershipProcessor // .publishOn(scheduler) .subscribe(this::onMemberEvent, this::onError), transport .listen() .publishOn(scheduler) .filter(this::isGossipReq) .subscribe(this::onGossipReq, this::onError))); } @Override public void start() { actionsDisposables.add( scheduler.schedulePeriodically( this::doSpreadGossip, config.getGossipInterval(), config.getGossipInterval(), TimeUnit.MILLISECONDS)); } @Override public void stop() { // Stop accepting gossip requests and spreading gossips actionsDisposables.dispose(); // Stop publishing events sink.complete(); } @Override public Mono<String> spread(Message message) { return Mono.fromCallable(() -> message) .subscribeOn(scheduler) .flatMap(msg -> Mono.create(sink -> futures.put(createAndPutGossip(msg), sink))); } @Override public 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 gossip gossipState = 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 period long period = currentPeriod++; // Check any gossips exists if (gossips.isEmpty()) { return; // nothing to spread } try { // Spread gossips to randomly selected member(s) selectGossipMembers().forEach(member -> spreadGossipsTo(period, member)); // Sweep gossips sweepGossips(period); } catch (Exception ex) { LOGGER.warn("Exception at doSpreadGossip[{}]: {}", period, ex.getMessage(), ex); } } private void spreadGossipsTo(long period, Member member) { // Select gossips to send List<Gossip> gossips = selectGossipsToSend(period, member); if (gossips.isEmpty()) { return; // nothing to spread } // Send gossip request Address 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 all return remoteMembers; } else { // select random members // Shuffle members initially and once reached top bound if (remoteMembersIndex < 0 || remoteMembersIndex + gossipFanout > remoteMembers.size()) { Collections.shuffle(remoteMembers); remoteMembersIndex = 0; } // Select members List<Member> selectedMembers = gossipFanout == 1 ? Collections.singletonList(remoteMembers.get(remoteMembersIndex)) : remoteMembers.subList(remoteMembersIndex, remoteMembersIndex + gossipFanout); // Increment index and return result remoteMembersIndex += 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 sweep int 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 selected if (gossipsToRemove.isEmpty()) { return; // nothing to sweep } // Sweep gossips LOGGER.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