import akka.cluster.ClusterEvent.MemberUp; //导入依赖的package包/类
@Override
public void onReceive(Object message) {
  if (message instanceof MemberUp) {
    MemberUp mUp = (MemberUp) message;
    log.info("Member is Up: {}", mUp.member());

  } else if (message instanceof UnreachableMember) {
    UnreachableMember mUnreachable = (UnreachableMember) message;
    log.info("Member detected as unreachable: {}", mUnreachable.member());

  } else if (message instanceof MemberRemoved) {
    MemberRemoved mRemoved = (MemberRemoved) message;
    log.info("Member is Removed: {}", mRemoved.member());

  } else if (message instanceof MemberEvent) {
    // ignore

  } else {
    unhandled(message);
  }
}
 

import akka.cluster.ClusterEvent.MemberUp; //导入依赖的package包/类
/**
 * Overridden super class method. Receives callback messages for cluster membership changes.
 * Updates list of current cluster member addresses based on these updates
 * @see akka.actor.UntypedActor#onReceive(java.lang.Object)
 */
public void onReceive(Object message) {
	if (message instanceof MemberUp) {
		MemberUp mUp = (MemberUp) message;
		logger.info("Cluster Member is Up: " + mUp.member());
		memberAddresses.add(mUp.member().address());
	} else if (message instanceof UnreachableMember) {
		UnreachableMember mUnreachable = (UnreachableMember) message;
		logger.info("Cluster Member detected as unreachable " + mUnreachable.member());
	} else if (message instanceof MemberRemoved) {
		MemberRemoved mRemoved = (MemberRemoved) message;
		logger.info("Cluster Member is Removed: " + mRemoved.member());
		memberAddresses.remove(mRemoved.member().address());
	} else if (message instanceof MemberEvent) {
		// ignore
	} else {
		unhandled(message);
	}
}
 

import akka.cluster.ClusterEvent.MemberUp; //导入依赖的package包/类
@Override
protected void handleCommand(final Object message) throws Exception {
    LOG.debug("{} : Received {}", clusterWrapper.getCurrentMemberName(), message);
    if (message instanceof ClusterEvent.MemberUp) {
        memberUp((ClusterEvent.MemberUp) message);
    } else if (message instanceof ClusterEvent.MemberWeaklyUp) {
        memberWeaklyUp((ClusterEvent.MemberWeaklyUp) message);
    } else if (message instanceof ClusterEvent.MemberExited) {
        memberExited((ClusterEvent.MemberExited) message);
    } else if (message instanceof ClusterEvent.MemberRemoved) {
        memberRemoved((ClusterEvent.MemberRemoved) message);
    } else if (message instanceof ClusterEvent.UnreachableMember) {
        memberUnreachable((ClusterEvent.UnreachableMember) message);
    } else if (message instanceof ClusterEvent.ReachableMember) {
        memberReachable((ClusterEvent.ReachableMember) message);
    } else if (message instanceof ProducerCreated) {
        onProducerCreated((ProducerCreated) message);
    } else if (message instanceof NotifyProducerCreated) {
        onNotifyProducerCreated((NotifyProducerCreated) message);
    } else if (message instanceof ProducerRemoved) {
        onProducerRemoved((ProducerRemoved) message);
    } else if (message instanceof NotifyProducerRemoved) {
        onNotifyProducerRemoved((NotifyProducerRemoved) message);
    } else if (message instanceof PrefixShardCreated) {
        onPrefixShardCreated((PrefixShardCreated) message);
    } else if (message instanceof LookupPrefixShard) {
        onLookupPrefixShard((LookupPrefixShard) message);
    } else if (message instanceof PrefixShardRemovalLookup) {
        onPrefixShardRemovalLookup((PrefixShardRemovalLookup) message);
    } else if (message instanceof PrefixShardRemoved) {
        onPrefixShardRemoved((PrefixShardRemoved) message);
    } else if (message instanceof StartConfigShardLookup) {
        onStartConfigShardLookup((StartConfigShardLookup) message);
    }
}
 

import akka.cluster.ClusterEvent.MemberUp; //导入依赖的package包/类
private void memberUp(final MemberUp message) {
    final MemberName memberName = memberToName(message.member());

    LOG.info("{}: Received MemberUp: memberName: {}, address: {}", persistenceId(), memberName,
            message.member().address());

    resolver.addPeerAddress(memberName, message.member().address());
}
 

import akka.cluster.ClusterEvent.MemberUp; //导入依赖的package包/类
@Override
public void preStart() {
    cluster.subscribe(getSelf(), MemberUp.class);
    virtualNode = getSelf();
    nodeName = self().path().name();
    Address address = cluster.selfAddress();
    System.out.println("Virtual Node : " + nodeName + " is up @ " + address.protocol() + " : " + address.hostPort());
    this.bootstraper = getContext().actorSelection(address.protocol() + "://" + address.hostPort() + "/user/bootstraper");
    //this.bootstraper.tell(new Identify(nodeName), virtualNode);
    this.ringManager = new ConsistentHash<>(new HashFunction(), numReplicas, new ArrayList<ActorRef>());
    this.localDB = new Memory();
    this.numOfNodes = 0;
}
 

import akka.cluster.ClusterEvent.MemberUp; //导入依赖的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);
}