import akka.cluster.Member; //导入依赖的package包/类
public void waitForMembersUp(final String... otherMembers) {
    Set<String> otherMembersSet = Sets.newHashSet(otherMembers);
    Stopwatch sw = Stopwatch.createStarted();
    while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
        CurrentClusterState state = Cluster.get(getSystem()).state();
        for (Member m: state.getMembers()) {
            if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.getRoles().iterator().next())
                    && otherMembersSet.isEmpty()) {
                return;
            }
        }

        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
    }

    fail("Member(s) " + otherMembersSet + " are not Up");
}
 

import akka.cluster.Member; //导入依赖的package包/类
static void waitForMembersUp(final ActorSystem node, final UniqueAddress... addresses) {
    Set<UniqueAddress> otherMembersSet = Sets.newHashSet(addresses);
    Stopwatch sw = Stopwatch.createStarted();
    while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
        CurrentClusterState state = Cluster.get(node).state();
        for (Member m : state.getMembers()) {
            if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.uniqueAddress())
                    && otherMembersSet.isEmpty()) {
                return;
            }
        }

        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
    }

    fail("Member(s) " + otherMembersSet + " are not Up");
}
 

import akka.cluster.Member; //导入依赖的package包/类
void register(Member member) {
    if (member.hasRole(Constants.CLUSTER_BACKEND)) {
        logger.info("Remote port:{} , nodes register.", member.address().port().get());
        getContext().watch(sender());

        String portStr = member.address().port().get().toString();
        clusterNodeRouters.put(Integer.valueOf(portStr.substring(portStr.length() - 2)), sender());
    }
}
 

import akka.cluster.Member; //导入依赖的package包/类
void remove(Member member) {
    if (member.hasRole(Constants.CLUSTER_BACKEND)) {
        logger.info("Remote port:{} , nodes remove.", member.address().port().get());
        getContext().unwatch(sender());
        clusterNodeRouters.inverse().remove(getSender());
    }
}
 

import akka.cluster.Member; //导入依赖的package包/类
@Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(ApiRequest.class, this::onRequest)
                .match(ApiRequestForward.class, this::onForward)
                .match(MemberOfflineEvent.class, this::onOffline)
                .match(MemberOnlineEvent.class, this::onOnline)
                .match(BackendFindEvent.class, e -> getSender().tell(findFrontendSocket, getSelf()))

                .match(ClusterEvent.CurrentClusterState.class, state -> {
                    for (Member member : state.getMembers()) {
                        if (member.status().equals(MemberStatus.up())) {
                            register(member);
                        } else if (member.status().equals(MemberStatus.removed())) {
                            remove(member);
                        }
                    }
                })
                .match(ClusterEvent.MemberUp.class, mUp -> register(mUp.member()))
                .match(ClusterEvent.MemberRemoved.class, mRem -> remove(mRem.member()))
                .match(Terminated.class, t -> {
                    logger.info("Frontend {} left.", getSender());
//                    getContext().unwatch(getSender());
//                    ClusterChildNodeSystem.INSTANCE.removeRouterFronted();
                })
                .build();
    }
 

import akka.cluster.Member; //导入依赖的package包/类
void register(Member member) {
    if (member.hasRole(Constants.CLUSTER_FRONTEND)) {
        logger.info("Frontend port:{} , nodes register.", member.address().port().get());
        String frontendPath = member.address() + Constants.CLUSTER_FRONTEND_PATH;
        findFrontendSocket = BackendFindFrontend.newBuilder().setFrontendAddress(frontendPath).build();
    }
}
 

import akka.cluster.Member; //导入依赖的package包/类
/**
 * Remove member from local copy of member list. If member down is self, then stop the actor
 *
 * @param member who went down
 */
private void receiveMemberRemoveOrUnreachable(final Member member) {
    LOG.debug("Received memberDown or Unreachable: {}", member);

    //if its self, then stop itself
    if (selfAddress.equals(member.address())) {
        getContext().stop(getSelf());
        return;
    }

    removePeer(member.address());
    LOG.debug("Removed member [{}], Active member list [{}]", member.address(), clusterMembers);
}
 

import akka.cluster.Member; //导入依赖的package包/类
/**
 * Add member to the local copy of member list if it doesn't already.
 *
 * @param member the member to add
 */
private void receiveMemberUpOrReachable(final Member member) {
    LOG.debug("Received memberUp or reachable: {}", member);

    //ignore up notification for self
    if (selfAddress.equals(member.address())) {
        return;
    }

    addPeer(member.address());
    LOG.debug("Added member [{}], Active member list [{}]", member.address(), clusterMembers);
}
 

import akka.cluster.Member; //导入依赖的package包/类
@Override
public void onReceive(Object message) throws Throwable
{
  if (message instanceof ClusterEvent.MemberUp) {
    ClusterEvent.MemberUp event = (ClusterEvent.MemberUp) message;
    this.up(event.member());
  } else if (message instanceof ClusterEvent.CurrentClusterState) {
    ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message;
    Iterable<Member> members = state.getMembers();
    for (Member o : members) {
      if (o.status().equals(MemberStatus.up())) {
        this.up(o);
      }
    }
  } else if (message instanceof ClusterEvent.UnreachableMember) {
    ClusterEvent.UnreachableMember mUnreachable = (ClusterEvent.UnreachableMember) message;
    log.info("有个节点不可达: {}", mUnreachable.member());
  } else if (message instanceof ClusterEvent.MemberRemoved) {
    ClusterEvent.MemberRemoved mRemoved = (ClusterEvent.MemberRemoved) message;
    log.info("有节点被删掉: {}", mRemoved.member());
  } else if (message instanceof ClusterEvent.MemberEvent) {
    log.info("节点事件: {}", message);
  } else if (message instanceof Registration) {
    Registration r = (Registration) message;
    if (match(this.slaves, r.getService())) {
      getContext().watch(getSender());
      this.addService(getSender(), r.getService());
      log.info("register success! sender =" + getSender());
      log.info("register success! self =" + this.self());
    } else {
        log.info("register failed! r.service = " + r.getService());
        log.info("register failed! this.service=" + this.service);
    }
  } else if (message instanceof Terminated) {
    Terminated terminated = (Terminated) message;
    this.removeService(terminated.actor());
  } else {
    this.onMessage(message);
  }
}
 

import akka.cluster.Member; //导入依赖的package包/类
/**
 * 向新加入节点注册自己
 *
 * @param member
 */
private void up(Member member)
{
  ActorSelection as = getContext().actorSelection(member.address() + "/user/*");
  if (as != null) {
    as.tell(new Registration(service), self());
  }
}
 

import akka.cluster.Member; //导入依赖的package包/类
void remove(Member member) {
    if (member.hasRole(Constants.CLUSTER_FRONTEND)) {
        logger.info("Frontend port:{} , nodes remove.", member.address().port().get());
        findFrontendSocket = BackendFindFrontend.getDefaultInstance();
    }
}
 

import akka.cluster.Member; //导入依赖的package包/类
@VisibleForTesting
static MemberName memberToName(final Member member) {
    return MemberName.forName(member.roles().iterator().next());
}
 

import akka.cluster.Member; //导入依赖的package包/类
private static MemberName memberToName(final Member member) {
    return MemberName.forName(member.roles().iterator().next());
}
 

import akka.cluster.Member; //导入依赖的package包/类
public static MemberRemoved createMemberRemoved(final String memberName, final String address) {
    UniqueAddress uniqueAddress = new UniqueAddress(AddressFromURIString.parse(address), 55L);
    Member member = new Member(uniqueAddress, 1, MemberStatus.removed(), setOf(memberName));

    return new MemberRemoved(member, MemberStatus.up());
}
 

import akka.cluster.Member; //导入依赖的package包/类
public static MemberUp createMemberUp(final String memberName, final String address) {
    UniqueAddress uniqueAddress = new UniqueAddress(AddressFromURIString.parse(address), 55L);
    Member member = new Member(uniqueAddress, 1, MemberStatus.up(), setOf(memberName));

    return new MemberUp(member);
}
 

import akka.cluster.Member; //导入依赖的package包/类
public static UnreachableMember createUnreachableMember(final String memberName, final String address) {
    UniqueAddress uniqueAddress = new UniqueAddress(AddressFromURIString.parse(address), 55L);
    Member member = new Member(uniqueAddress, 1, MemberStatus.up(), setOf(memberName));

    return new UnreachableMember(member);
}
 

import akka.cluster.Member; //导入依赖的package包/类
public static ReachableMember createReachableMember(final String memberName, final String address) {
    UniqueAddress uniqueAddress = new UniqueAddress(AddressFromURIString.parse(address), 55L);
    Member member = new Member(uniqueAddress, 1, MemberStatus.up(), setOf(memberName));

    return new ReachableMember(member);
}
 

import akka.cluster.Member; //导入依赖的package包/类
private void countDownIfOther(final Member member, final CountDownLatch latch) {
    if (!getCluster().getCurrentMemberName().equals(memberToName(member))) {
        latch.countDown();
    }
}
 

import akka.cluster.Member; //导入依赖的package包/类
/**
 * Process messages about nodes up, down, reachable and unreachable.
 */
private void processAsClusterEvent(Object message) {

    if (message instanceof ClusterEvent.CurrentClusterState) {
        ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message;
        nodes.clear();
        for (Member member : state.getMembers()) {
            if (member.hasRole("io") && member.status().equals( MemberStatus.up())) {
                nodes.add(member.address());
                logger.debug("RequestActor {} received cluster-state member-up for {}", name, member.address());
            }
        }

    } else if (message instanceof ClusterEvent.MemberUp) {
        ClusterEvent.MemberUp mUp = (ClusterEvent.MemberUp) message;
        if (mUp.member().hasRole("io")) {
            nodes.add( mUp.member().address() );
        }
        logger.debug("{} received member-up for {}", name, mUp.member().address());

    } else if (message instanceof ClusterEvent.MemberEvent) {
        ClusterEvent.MemberEvent other = (ClusterEvent.MemberEvent) message;
        nodes.remove(other.member().address());

    } else if (message instanceof ClusterEvent.UnreachableMember) {
        ClusterEvent.UnreachableMember unreachable = (ClusterEvent.UnreachableMember) message;
        nodes.remove(unreachable.member().address());
        logger.debug("{} received un-reachable for {}", name, unreachable.member().address());

    } else if (message instanceof ClusterEvent.ReachableMember) {
        ClusterEvent.ReachableMember reachable = (ClusterEvent.ReachableMember) message;
        if (reachable.member().hasRole("io")) {
            nodes.add( reachable.member().address() );
        }
        logger.debug("{} received reachable for {}", name, reachable.member().address());

    } else {
        logger.error("{}: unhandled message: {}", name, message.toString());
        unhandled(message);
    }
}