聊聊scalecubecluster的GossipProtocol

序

本文主要研究一下 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