import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Helper method to broadcast a message to all members in a channel
 * @param msgtype int
 * @param rpc boolean
 * @throws ChannelException
 */
protected void broadcast(int msgtype, boolean rpc) throws ChannelException {
    Member[] members = channel.getMembers();
    // No destination.
    if (members.length == 0 ) return;
    //send out a map membership message, only wait for the first reply
    MapMessage msg = new MapMessage(this.mapContextName, msgtype,
                                    false, null, null, null, channel.getLocalMember(false), null);
    if ( rpc) {
        Response[] resp = rpcChannel.send(members, msg,
                RpcChannel.FIRST_REPLY, (channelSendOptions), rpcTimeout);
        if (resp.length > 0) {
            for (int i = 0; i < resp.length; i++) {
                mapMemberAdded(resp[i].getSource());
                messageReceived(resp[i].getMessage(), resp[i].getSource());
            }
        } else {
            log.warn("broadcast received 0 replies, probably a timeout.");
        }
    } else {
        channel.send(channel.getMembers(),msg,channelSendOptions);
    }
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Helper method to broadcast a message to all members in a channel
 * 
 * @param msgtype
 *            int
 * @param rpc
 *            boolean
 * @throws ChannelException
 */
protected void broadcast(int msgtype, boolean rpc) throws ChannelException {
	Member[] members = channel.getMembers();
	// No destination.
	if (members.length == 0)
		return;
	// send out a map membership message, only wait for the first reply
	MapMessage msg = new MapMessage(this.mapContextName, msgtype, false, null, null, null,
			channel.getLocalMember(false), null);
	if (rpc) {
		Response[] resp = rpcChannel.send(members, msg, RpcChannel.FIRST_REPLY, (channelSendOptions), rpcTimeout);
		if (resp.length > 0) {
			for (int i = 0; i < resp.length; i++) {
				mapMemberAdded(resp[i].getSource());
				messageReceived(resp[i].getMessage(), resp[i].getSource());
			}
		} else {
			log.warn("broadcast received 0 replies, probably a timeout.");
		}
	} else {
		channel.send(channel.getMembers(), msg, channelSendOptions);
	}
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Send the list of members to the <code>member</code>
 *
 * @param member The member to whom the member list has to be sent
 */
public void sendMemberList(Member member) {
    try {
        MemberListCommand memListCmd = new MemberListCommand();
        List<Member> members = new ArrayList<Member>(this.members);
        memListCmd.setMembers(members.toArray(new Member[members.size()]));
        rpcMembershipChannel.send(new Member[]{member}, memListCmd, RpcChannel.ALL_REPLY,
                                  Channel.SEND_OPTIONS_ASYNCHRONOUS |
                                  TribesConstants.MEMBERSHIP_MSG_OPTION, 10000);
        if (log.isDebugEnabled()) {
            log.debug("Sent MEMBER_LIST to " + TribesUtil.getName(member));
        }
    } catch (Exception e) {
        String errMsg = "Could not send MEMBER_LIST to member " + TribesUtil.getName(member);
        log.error(errMsg, e);
        throw new RemoteProcessException(errMsg, e);
    }
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
public EchoRpcTest(Channel channel, String name, int count, String message, long pause, int options, long timeout) {
    this.channel = channel;
    this.count = count;
    this.message = message;
    this.pause = pause;
    this.options = options;
    this.rpc = new RpcChannel(name.getBytes(),channel,this);
    this.timeout = timeout;
    this.name = name;
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Inform all members that a particular member just joined
 *
 * @param member The member who just joined
 */
public void sendMemberJoinedToAll(Member member) {
    try {

        MemberJoinedCommand cmd = new MemberJoinedCommand();
        cmd.setMember(member);
        ArrayList<Member> membersToSend = (ArrayList<Member>) (((ArrayList) members).clone());
        membersToSend.remove(member); // Do not send MEMBER_JOINED to the new member who just joined

        if (membersToSend.size() > 0) {
            rpcMembershipChannel.send(membersToSend.toArray(new Member[membersToSend.size()]), cmd,
                                      RpcChannel.ALL_REPLY,
                                      Channel.SEND_OPTIONS_ASYNCHRONOUS |
                                      TribesConstants.MEMBERSHIP_MSG_OPTION,
                                      10000);
            if (log.isDebugEnabled()) {
                log.debug("Sent MEMBER_JOINED[" + TribesUtil.getName(member) +
                          "] to all members in domain " + new String(domain));
            }
        }
    } catch (Exception e) {
        String errMsg = "Could not send MEMBER_JOINED[" + TribesUtil.getName(member) +
                        "] to all members ";
        log.error(errMsg, e);
        throw new RemoteProcessException(errMsg, e);
    }
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Initializes the map by creating the RPC channel, registering itself as a channel listener
 * This method is also responsible for initiating the state transfer
 * @param owner Object
 * @param channel Channel
 * @param mapContextName String
 * @param timeout long
 * @param channelSendOptions int
 * @param cls ClassLoader[]
 */
protected void init(MapOwner owner, Channel channel, String mapContextName, long timeout, int channelSendOptions,ClassLoader[] cls) {
    log.info("Initializing AbstractReplicatedMap with context name:"+mapContextName);
    this.mapOwner = owner;
    this.externalLoaders = cls;
    this.channelSendOptions = channelSendOptions;
    this.channel = channel;
    this.rpcTimeout = timeout;

    this.mapname = mapContextName;
    //unique context is more efficient if it is stored as bytes
    this.mapContextName = mapContextName.getBytes(CHARSET_ISO_8859_1);
    if ( log.isTraceEnabled() ) log.trace("Created Lazy Map with name:"+mapContextName+", bytes:"+Arrays.toString(this.mapContextName));

    //create an rpc channel and add the map as a listener
    this.rpcChannel = new RpcChannel(this.mapContextName, channel, this);
    //add this map as a message listener
    this.channel.addChannelListener(this);
    //listen for membership notifications
    this.channel.addMembershipListener(this);
    
    
    try {
        //broadcast our map, this just notifies other members of our existence
        broadcast(MapMessage.MSG_INIT, true);
        //transfer state from another map
        transferState();
        //state is transferred, we are ready for messaging
        broadcast(MapMessage.MSG_START, true);
    } catch (ChannelException x) {
        log.warn("Unable to send map start message.");
        // remove listener from channel
        this.rpcChannel.breakdown();
        this.channel.removeChannelListener(this);
        this.channel.removeMembershipListener(this);
        throw new RuntimeException("Unable to start replicated map.",x);
    }
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Initializes the map by creating the RPC channel, registering itself as a channel listener
 * This method is also responsible for initiating the state transfer
 * @param owner Object
 * @param channel Channel
 * @param mapContextName String
 * @param timeout long
 * @param channelSendOptions int
 * @param cls ClassLoader[]
 * @param terminate - Flag for whether to terminate this map that failed to start.
 */
protected void init(MapOwner owner, Channel channel, String mapContextName,
        long timeout, int channelSendOptions,ClassLoader[] cls, boolean terminate) {
    long start = System.currentTimeMillis();
    if (log.isInfoEnabled()) log.info("Initializing AbstractReplicatedMap with context name:"+mapContextName);
    this.mapOwner = owner;
    this.externalLoaders = cls;
    this.channelSendOptions = channelSendOptions;
    this.channel = channel;
    this.rpcTimeout = timeout;

    this.mapname = mapContextName;
    //unique context is more efficient if it is stored as bytes
    this.mapContextName = mapContextName.getBytes(CHARSET_ISO_8859_1);
    if ( log.isTraceEnabled() ) log.trace("Created Lazy Map with name:"+mapContextName+", bytes:"+Arrays.toString(this.mapContextName));

    //create an rpc channel and add the map as a listener
    this.rpcChannel = new RpcChannel(this.mapContextName, channel, this);
    //add this map as a message listener
    this.channel.addChannelListener(this);
    //listen for membership notifications
    this.channel.addMembershipListener(this);

    try {
        //broadcast our map, this just notifies other members of our existence
        broadcast(MapMessage.MSG_INIT, true);
        //transfer state from another map
        transferState();
        //state is transferred, we are ready for messaging
        broadcast(MapMessage.MSG_START, true);
    } catch (ChannelException x) {
        log.warn("Unable to send map start message.");
        if (terminate) {
            breakdown();
            throw new RuntimeException("Unable to start replicated map.",x);
        }
    }
    this.state = State.INITIALIZED;
    long complete = System.currentTimeMillis() - start;
    if (log.isInfoEnabled())
        log.info("AbstractReplicatedMap[" +mapContextName + "] initialization was completed in " + complete + " ms.");
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public V get(Object key) {
    MapEntry<K,V> entry = innerMap.get(key);
    if (log.isTraceEnabled()) log.trace("Requesting id:"+key+" entry:"+entry);
    if ( entry == null ) return null;
    if ( !entry.isPrimary() ) {
        //if the message is not primary, we need to retrieve the latest value
        try {
            Member[] backup = null;
            MapMessage msg = null;
            if (entry.isBackup()) {
                //select a new backup node
                backup = publishEntryInfo(key, entry.getValue());
            } else if ( entry.isProxy() ) {
                //make sure we don't retrieve from ourselves
                msg = new MapMessage(getMapContextName(), MapMessage.MSG_RETRIEVE_BACKUP, false,
                                     (Serializable) key, null, null, null,null);
                Response[] resp = getRpcChannel().send(entry.getBackupNodes(),msg, RpcChannel.FIRST_REPLY, getChannelSendOptions(), getRpcTimeout());
                if (resp == null || resp.length == 0 || resp[0].getMessage() == null) {
                    //no responses
                    log.warn("Unable to retrieve remote object for key:" + key);
                    return null;
                }
                msg = (MapMessage) resp[0].getMessage();
                msg.deserialize(getExternalLoaders());
                backup = entry.getBackupNodes();
                if ( msg.getValue()!=null ) entry.setValue((V) msg.getValue());

                // notify member
                msg = new MapMessage(getMapContextName(), MapMessage.MSG_NOTIFY_MAPMEMBER,false,
                        (Serializable)entry.getKey(), null, null, channel.getLocalMember(false), backup);
                if ( backup != null && backup.length > 0) {
                    getChannel().send(backup, msg, getChannelSendOptions());
                }
                
                //invalidate the previous primary
                msg = new MapMessage(getMapContextName(),MapMessage.MSG_PROXY,false,(Serializable)key,null,null,channel.getLocalMember(false),backup);
                Member[] dest = getMapMembersExcl(backup);
                if ( dest!=null && dest.length >0) {
                    getChannel().send(dest, msg, getChannelSendOptions());
                }
                if (entry.getValue() instanceof ReplicatedMapEntry) {
                    ReplicatedMapEntry val = (ReplicatedMapEntry)entry.getValue();
                    val.setOwner(getMapOwner());
                }
            } else if ( entry.isCopy() ) {
                backup = getMapMembers();
                if (backup.length > 0) {
                    msg = new MapMessage(getMapContextName(), MapMessage.MSG_NOTIFY_MAPMEMBER,false,
                            (Serializable)key,null,null,channel.getLocalMember(false),backup);
                    getChannel().send(backup, msg, getChannelSendOptions());
                }
            }
            entry.setPrimary(channel.getLocalMember(false));
            entry.setBackupNodes(backup);
            entry.setBackup(false);
            entry.setProxy(false);
            entry.setCopy(false);
            if ( getMapOwner()!=null ) getMapOwner().objectMadePrimay(key, entry.getValue());

        } catch (Exception x) {
            log.error("Unable to replicate out data for a AbstractReplicatedMap.get operation", x);
            return null;
        }
    }
    if (log.isTraceEnabled()) log.trace("Requesting id:"+key+" result:"+entry.getValue());
    return entry.getValue();
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
public RpcChannel getRpcChannel() {
    return rpcChannel;
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    long pause = 3000;
    int count = 1000000;
    int stats = 10000;
    String name = "EchoRpcId";
    int options = RpcChannel.ALL_REPLY;
    long timeout = 15000;
    String message = "EchoRpcMessage";
    if (args.length == 0) {
        usage();
        System.exit(1);
    }
    for (int i = 0; i < args.length; i++) {
        if ("-threads".equals(args[i])) {
            // Not used
        } else if ("-count".equals(args[i])) {
            count = Integer.parseInt(args[++i]);
            System.out.println("Sending "+count+" messages.");
        } else if ("-pause".equals(args[i])) {
            pause = Long.parseLong(args[++i])*1000;
        } else if ("-break".equals(args[i])) {
            // Not used
        } else if ("-stats".equals(args[i])) {
            stats = Integer.parseInt(args[++i]);
            System.out.println("Stats every "+stats+" message");
        } else if ("-timeout".equals(args[i])) {
            timeout = Long.parseLong(args[++i]);
        } else if ("-message".equals(args[i])) {
            message = args[++i];
        } else if ("-name".equals(args[i])) {
            name = args[++i];
        } else if ("-mode".equals(args[i])) {
            if ( "all".equals(args[++i]) ) options = RpcChannel.ALL_REPLY;
            else if ( "first".equals(args[i]) ) options = RpcChannel.FIRST_REPLY;
            else if ( "majority".equals(args[i]) ) options = RpcChannel.MAJORITY_REPLY;
        } else if ("-debug".equals(args[i])) {
            // Not used
        } else if ("-help".equals(args[i])) {
            usage();
            System.exit(1);
        }
    }


    ManagedChannel channel = (ManagedChannel)ChannelCreator.createChannel(args);
    EchoRpcTest test = new EchoRpcTest(channel,name,count,message,pause,options,timeout);
    channel.start(Channel.DEFAULT);
    Runtime.getRuntime().addShutdownHook(new Shutdown(channel));
    test.run();

    System.out.println("System test complete, sleeping to let threads finish.");
    Thread.sleep(60*1000*60);
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Initializes the map by creating the RPC channel, registering itself as a
 * channel listener This method is also responsible for initiating the state
 * transfer
 * 
 * @param owner
 *            Object
 * @param channel
 *            Channel
 * @param mapContextName
 *            String
 * @param timeout
 *            long
 * @param channelSendOptions
 *            int
 * @param cls
 *            ClassLoader[]
 * @param terminate
 *            - Flag for whether to terminate this map that failed to start.
 */
protected void init(MapOwner owner, Channel channel, String mapContextName, long timeout, int channelSendOptions,
		ClassLoader[] cls, boolean terminate) {
	long start = System.currentTimeMillis();
	if (log.isInfoEnabled())
		log.info("Initializing AbstractReplicatedMap with context name:" + mapContextName);
	this.mapOwner = owner;
	this.externalLoaders = cls;
	this.channelSendOptions = channelSendOptions;
	this.channel = channel;
	this.rpcTimeout = timeout;

	this.mapname = mapContextName;
	// unique context is more efficient if it is stored as bytes
	this.mapContextName = mapContextName.getBytes(CHARSET_ISO_8859_1);
	if (log.isTraceEnabled())
		log.trace(
				"Created Lazy Map with name:" + mapContextName + ", bytes:" + Arrays.toString(this.mapContextName));

	// create an rpc channel and add the map as a listener
	this.rpcChannel = new RpcChannel(this.mapContextName, channel, this);
	// add this map as a message listener
	this.channel.addChannelListener(this);
	// listen for membership notifications
	this.channel.addMembershipListener(this);

	try {
		// broadcast our map, this just notifies other members of our
		// existence
		broadcast(MapMessage.MSG_INIT, true);
		// transfer state from another map
		transferState();
		// state is transferred, we are ready for messaging
		broadcast(MapMessage.MSG_START, true);
	} catch (ChannelException x) {
		log.warn("Unable to send map start message.");
		if (terminate) {
			breakdown();
			throw new RuntimeException("Unable to start replicated map.", x);
		}
	}
	this.state = State.INITIALIZED;
	long complete = System.currentTimeMillis() - start;
	if (log.isInfoEnabled())
		log.info("AbstractReplicatedMap[" + mapContextName + "] initialization was completed in " + complete
				+ " ms.");
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public V get(Object key) {
	MapEntry<K, V> entry = innerMap.get(key);
	if (log.isTraceEnabled())
		log.trace("Requesting id:" + key + " entry:" + entry);
	if (entry == null)
		return null;
	if (!entry.isPrimary()) {
		// if the message is not primary, we need to retrieve the latest
		// value
		try {
			Member[] backup = null;
			MapMessage msg = null;
			if (entry.isBackup()) {
				// select a new backup node
				backup = publishEntryInfo(key, entry.getValue());
			} else if (entry.isProxy()) {
				// make sure we don't retrieve from ourselves
				msg = new MapMessage(getMapContextName(), MapMessage.MSG_RETRIEVE_BACKUP, false, (Serializable) key,
						null, null, null, null);
				Response[] resp = getRpcChannel().send(entry.getBackupNodes(), msg, RpcChannel.FIRST_REPLY,
						getChannelSendOptions(), getRpcTimeout());
				if (resp == null || resp.length == 0 || resp[0].getMessage() == null) {
					// no responses
					log.warn("Unable to retrieve remote object for key:" + key);
					return null;
				}
				msg = (MapMessage) resp[0].getMessage();
				msg.deserialize(getExternalLoaders());
				backup = entry.getBackupNodes();
				if (msg.getValue() != null)
					entry.setValue((V) msg.getValue());

				// notify member
				msg = new MapMessage(getMapContextName(), MapMessage.MSG_NOTIFY_MAPMEMBER, false,
						(Serializable) entry.getKey(), null, null, channel.getLocalMember(false), backup);
				if (backup != null && backup.length > 0) {
					getChannel().send(backup, msg, getChannelSendOptions());
				}

				// invalidate the previous primary
				msg = new MapMessage(getMapContextName(), MapMessage.MSG_PROXY, false, (Serializable) key, null,
						null, channel.getLocalMember(false), backup);
				Member[] dest = getMapMembersExcl(backup);
				if (dest != null && dest.length > 0) {
					getChannel().send(dest, msg, getChannelSendOptions());
				}
				if (entry.getValue() instanceof ReplicatedMapEntry) {
					ReplicatedMapEntry val = (ReplicatedMapEntry) entry.getValue();
					val.setOwner(getMapOwner());
				}
			} else if (entry.isCopy()) {
				backup = getMapMembers();
				if (backup.length > 0) {
					msg = new MapMessage(getMapContextName(), MapMessage.MSG_NOTIFY_MAPMEMBER, false,
							(Serializable) key, null, null, channel.getLocalMember(false), backup);
					getChannel().send(backup, msg, getChannelSendOptions());
				}
			}
			entry.setPrimary(channel.getLocalMember(false));
			entry.setBackupNodes(backup);
			entry.setBackup(false);
			entry.setProxy(false);
			entry.setCopy(false);
			if (getMapOwner() != null)
				getMapOwner().objectMadePrimay(key, entry.getValue());

		} catch (Exception x) {
			log.error("Unable to replicate out data for a AbstractReplicatedMap.get operation", x);
			return null;
		}
	}
	if (log.isTraceEnabled())
		log.trace("Requesting id:" + key + " result:" + entry.getValue());
	return entry.getValue();
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
public RpcChannel getRpcChannel() {
	return rpcChannel;
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Initializes the map by creating the RPC channel, registering itself as a channel listener
 * This method is also responsible for initiating the state transfer
 * @param owner Object
 * @param channel Channel
 * @param mapContextName String
 * @param timeout long
 * @param channelSendOptions int
 * @param cls ClassLoader[]
 */
protected void init(MapOwner owner, Channel channel, String mapContextName,
        long timeout, int channelSendOptions,ClassLoader[] cls, boolean terminate) {
    long start = System.currentTimeMillis();
    log.info("Initializing AbstractReplicatedMap with context name:"+mapContextName);
    this.mapOwner = owner;
    this.externalLoaders = cls;
    this.channelSendOptions = channelSendOptions;
    this.channel = channel;
    this.rpcTimeout = timeout;

    this.mapname = mapContextName;
    //unique context is more efficient if it is stored as bytes
    this.mapContextName = mapContextName.getBytes(CHARSET_ISO_8859_1);
    if ( log.isTraceEnabled() ) log.trace("Created Lazy Map with name:"+mapContextName+", bytes:"+Arrays.toString(this.mapContextName));

    //create an rpc channel and add the map as a listener
    this.rpcChannel = new RpcChannel(this.mapContextName, channel, this);
    //add this map as a message listener
    this.channel.addChannelListener(this);
    //listen for membership notifications
    this.channel.addMembershipListener(this);


    try {
        //broadcast our map, this just notifies other members of our existence
        broadcast(MapMessage.MSG_INIT, true);
        //transfer state from another map
        transferState();
        //state is transferred, we are ready for messaging
        broadcast(MapMessage.MSG_START, true);
    } catch (ChannelException x) {
        log.warn("Unable to send map start message.");
        if (terminate) {
            breakdown();
            throw new RuntimeException("Unable to start replicated map.",x);
        }
    }
    long complete = System.currentTimeMillis() - start;
    if (log.isInfoEnabled())
        log.info("AbstractReplicatedMap[" +mapContextName + "] initialization was completed in " + complete + " ms.");
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Sends a ping out to all the members in the cluster, not just map members
 * that this map is alive.
 * @param timeout long
 * @throws ChannelException
 */
protected void ping(long timeout) throws ChannelException {
    //send out a map membership message, only wait for the first reply
    MapMessage msg = new MapMessage(this.mapContextName,
                                    MapMessage.MSG_INIT,
                                    false,
                                    null,
                                    null,
                                    null,
                                    channel.getLocalMember(false),
                                    null);
    if ( channel.getMembers().length > 0 ) {
        try {
            //send a ping, wait for all nodes to reply
            Response[] resp = rpcChannel.send(channel.getMembers(),
                                              msg, RpcChannel.ALL_REPLY,
                                              (channelSendOptions),
                                              (int) accessTimeout);
            for (int i = 0; i < resp.length; i++) {
                memberAlive(resp[i].getSource());
            }
        } catch (ChannelException ce) {
            // Handle known failed members
            FaultyMember[] faultyMembers = ce.getFaultyMembers();
            for (FaultyMember faultyMember : faultyMembers) {
                memberDisappeared(faultyMember.getMember());
            }
            throw ce;
        }
    }
    //update our map of members, expire some if we didn't receive a ping back
    synchronized (mapMembers) {
        Member[] members = mapMembers.keySet().toArray(new Member[mapMembers.size()]);
        long now = System.currentTimeMillis();
        for (Member member : members) {
            long access = mapMembers.get(member);
            if ( (now - access) > timeout ) {
                memberDisappeared(member);
            }
        }
    }//synch
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Initializes the map by creating the RPC channel, registering itself as a channel listener
 * This method is also responsible for initiating the state transfer
 * @param owner Object
 * @param channel Channel
 * @param mapContextName String
 * @param timeout long
 * @param channelSendOptions int
 * @param cls ClassLoader[]
 * @param terminate - Flag for whether to terminate this map that failed to start.
 */
protected void init(MapOwner owner, Channel channel, String mapContextName,
        long timeout, int channelSendOptions,ClassLoader[] cls, boolean terminate) {
    long start = System.currentTimeMillis();
    if (log.isInfoEnabled()) log.info("Initializing AbstractReplicatedMap with context name:"+mapContextName);
    this.mapOwner = owner;
    this.externalLoaders = cls;
    this.channelSendOptions = channelSendOptions;
    this.channel = channel;
    this.rpcTimeout = timeout;

    this.mapname = mapContextName;
    //unique context is more efficient if it is stored as bytes
    this.mapContextName = mapContextName.getBytes(CHARSET_ISO_8859_1);
    if ( log.isTraceEnabled() ) log.trace("Created Lazy Map with name:"+mapContextName+", bytes:"+Arrays.toString(this.mapContextName));

    //create an rpc channel and add the map as a listener
    this.rpcChannel = new RpcChannel(this.mapContextName, channel, this);
    //add this map as a message listener
    this.channel.addChannelListener(this);
    //listen for membership notifications
    this.channel.addMembershipListener(this);


    try {
        //broadcast our map, this just notifies other members of our existence
        broadcast(MapMessage.MSG_INIT, true);
        //transfer state from another map
        transferState();
        //state is transferred, we are ready for messaging
        broadcast(MapMessage.MSG_START, true);
    } catch (ChannelException x) {
        log.warn("Unable to send map start message.");
        if (terminate) {
            breakdown();
            throw new RuntimeException("Unable to start replicated map.",x);
        }
    }
    long complete = System.currentTimeMillis() - start;
    if (log.isInfoEnabled())
        log.info("AbstractReplicatedMap[" +mapContextName + "] initialization was completed in " + complete + " ms.");
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
/**
 * Sends a ping out to all the members in the cluster, not just map members
 * that this map is alive.
 * @param timeout long
 * @throws ChannelException
 */
protected void ping(long timeout) throws ChannelException {
    //send out a map membership message, only wait for the first reply
    MapMessage msg = new MapMessage(this.mapContextName,
                                    MapMessage.MSG_INIT,
                                    false,
                                    null,
                                    null,
                                    null,
                                    channel.getLocalMember(false),
                                    null);
    if ( channel.getMembers().length > 0 ) {
        try {
            //send a ping, wait for all nodes to reply
            Response[] resp = rpcChannel.send(channel.getMembers(),
                                              msg, RpcChannel.ALL_REPLY,
                                              (channelSendOptions),
                                              (int) accessTimeout);
            for (int i = 0; i < resp.length; i++) {
                memberAlive(resp[i].getSource());
            }
        } catch (ChannelException ce) {
            // Handle known failed members
            FaultyMember[] faultyMembers = ce.getFaultyMembers();
            for (FaultyMember faultyMember : faultyMembers) {
                memberDisappeared(faultyMember.getMember());
            }
            throw ce;
        }
    }
    //update our map of members, expire some if we didn't receive a ping back
    synchronized (mapMembers) {
        Member[] members = mapMembers.keySet().toArray(new Member[mapMembers.size()]);
        long now = System.currentTimeMillis();
        for (Member member : members) {
            long access = mapMembers.get(member).longValue();
            if ( (now - access) > timeout ) {
                memberDisappeared(member);
            }
        }
    }//synch
}
 

import org.apache.catalina.tribes.group.RpcChannel; //导入依赖的package包/类
public void setRpcMembershipChannel(RpcChannel rpcMembershipChannel) {
    this.rpcMembershipChannel = rpcMembershipChannel;
}