distributedhashtable.java

来自「JGRoups源码」· Java 代码 · 共 646 行 · 第 1/2 页

// $Id: DistributedHashtable.java,v 1.26 2006/09/01 14:40:26 belaban Exp $

package org.jgroups.blocks;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.*;
import org.jgroups.persistence.CannotPersistException;
import org.jgroups.persistence.CannotRemoveException;
import org.jgroups.persistence.PersistenceFactory;
import org.jgroups.persistence.PersistenceManager;
import org.jgroups.util.Promise;
import org.jgroups.util.Util;

import java.io.Serializable;
import java.util.*;





/**
 * Provides the abstraction of a java.util.Hashtable that is replicated at several
 * locations. Any change to the hashtable (clear, put, remove etc) will transparently be
 * propagated to all replicas in the group. All read-only methods will always access the
 * local replica.<p>
 * Both keys and values added to the hashtable <em>must be serializable</em>, the reason
 * being that they will be sent across the network to all replicas of the group. Having said
 * this, it is now for example possible to add RMI remote objects to the hashtable as they
 * are derived from <code>java.rmi.server.RemoteObject</code> which in turn is serializable.
 * This allows to lookup shared distributed objects by their name and invoke methods on them,
 * regardless of one's onw location. A <code>DistributedHashtable</code> thus allows to
 * implement a distributed naming service in just a couple of lines.<p>
 * An instance of this class will contact an existing member of the group to fetch its
 * initial state (using the state exchange funclet <code>StateExchangeFunclet</code>.
 * @author Bela Ban
 * @author <a href="mailto:[e-mail]">Alfonso Olias-Sanz</a>
 * @version $Id: DistributedHashtable.java,v 1.26 2006/09/01 14:40:26 belaban Exp $
 */
public class DistributedHashtable extends Hashtable implements MessageListener, MembershipListener {



    public interface Notification {
        void entrySet(Object key, Object value);
        void entryRemoved(Object key);
        void viewChange(Vector new_mbrs, Vector old_mbrs);
        void contentsSet(Map new_entries);
        void contentsCleared();
    }


    private transient Channel               channel;
    protected transient RpcDispatcher       disp=null;
    private transient String                groupname=null;
    private final transient Vector          notifs=new Vector();  // to be notified when mbrship changes
    private final transient Vector          members=new Vector(); // keeps track of all DHTs
    private transient Class[]               put_signature=null;
    private transient Class[]               putAll_signature=null;
    private transient Class[]               clear_signature=null;
    private transient Class[]               remove_signature=null;
    private transient boolean               persistent=false; // whether to use PersistenceManager to save state
    private transient PersistenceManager    persistence_mgr=null;

	/** Determines when the updates have to be sent across the network, avoids sending unnecessary
     * messages when there are no member in the group */
	private transient boolean            send_message = false;

    protected final transient Promise          state_promise=new Promise();

    protected final Log log=LogFactory.getLog(this.getClass());




    /**
     * Creates a DistributedHashtable
     * @param groupname The name of the group to join
     * @param factory The ChannelFactory which will be used to create a channel
     * @param properties The property string to be used to define the channel. This will override the properties of
     * the factory. If null, then the factory properties will be used
     * @param state_timeout The time to wait until state is retrieved in milliseconds. A value of 0 means wait forever.
     */
    public DistributedHashtable(String groupname, ChannelFactory factory,
                                String properties, long state_timeout)
            throws ChannelException {
        this.groupname=groupname;
        initSignatures();
        if(factory != null) {
            channel=properties != null? factory.createChannel(properties) : factory.createChannel();
        }
        else {
            channel=new JChannel(properties);
        }
        disp=new RpcDispatcher(channel, this, this, this);
        channel.connect(groupname);
        start(state_timeout);
    }

    /**
     * Creates a DisttributedHashtable. Optionally the contents can be saved to
     * persistemt storage using the {@link PersistenceManager}.
     * @param groupname Name of the group to join
     * @param factory Instance of a ChannelFactory to create the channel
     * @param properties Protocol stack properties. This will override the properties of the factory. If
     * null, then the factory properties will be used
     * @param persistent Whether the contents should be persisted
     * @param state_timeout Max number of milliseconds to wait until state is
     * retrieved
     */
    public DistributedHashtable(String groupname, ChannelFactory factory, String properties,
                                boolean persistent, long state_timeout)
            throws ChannelException {
        this.groupname=groupname;
        this.persistent=persistent;
        initSignatures();
        if(factory != null) {
            channel=properties != null? factory.createChannel(properties) : factory.createChannel();
        }
        else {
            channel=new JChannel(properties);
        }
        disp=new RpcDispatcher(channel, this, this, this);
        channel.connect(groupname);
        start(state_timeout);
    }


    public DistributedHashtable(JChannel channel, long state_timeout) {
        this(channel, false, state_timeout);
    }


    public DistributedHashtable(JChannel channel, boolean persistent, long state_timeout) {
        this.groupname = channel.getClusterName();
        this.channel = channel;
        this.persistent=persistent;
        init(state_timeout);
    }

    /**
     * Uses a user-provided PullPushAdapter to create the dispatcher rather than a Channel. If id is non-null, it will be
     * used to register under that id. This is typically used when another building block is already using
     * PullPushAdapter, and we want to add this building block in addition. The id is the used to discriminate
     * between messages for the various blocks on top of PullPushAdapter. If null, we will assume we are the
     * first block created on PullPushAdapter.
     * @param adapter The PullPushAdapter which to use as underlying transport
     * @param id A serializable object (e.g. an Integer) used to discriminate (multiplex/demultiplex) between
     *           requests/responses for different building blocks on top of PullPushAdapter.
     * @param state_timeout Max number of milliseconds to wait until state is
     * retrieved
     */
    public DistributedHashtable(PullPushAdapter adapter, Serializable id, long state_timeout)
        throws ChannelNotConnectedException, ChannelClosedException {
        initSignatures();
        this.channel = (Channel)adapter.getTransport();
        this.groupname = this.channel.getClusterName();
        disp=new RpcDispatcher(adapter, id, this, this, this);
        start(state_timeout);
    }

    public DistributedHashtable(PullPushAdapter adapter, Serializable id) {
        initSignatures();
        this.channel = (Channel)adapter.getTransport();
        this.groupname = this.channel.getClusterName();
        disp=new RpcDispatcher(adapter, id, this, this, this);
    }

    protected final void init(long state_timeout) {
        initSignatures();
        disp = new RpcDispatcher(channel, this, this, this);

        // Changed by bela (jan 20 2003): start() has to be called by user (only when providing
        // own channel). First, Channel.connect() has to be called, then start().
        // start(state_timeout);
    }


    /**
     * Fetches the state
     * @param state_timeout
     * @throws ChannelClosedException
     * @throws ChannelNotConnectedException
     */
    public final void start(long state_timeout) throws ChannelClosedException, ChannelNotConnectedException {
        boolean rc;
        if(persistent) {
            if(log.isInfoEnabled()) log.info("fetching state from database");
            try {
                persistence_mgr=PersistenceFactory.getInstance().createManager();
            }
            catch(Throwable ex) {
                if(log.isErrorEnabled()) log.error("failed creating PersistenceManager, " +
                            "turning persistency off. Exception: " + Util.printStackTrace(ex));
                persistent=false;
            }
        }

        state_promise.reset();
        rc=channel.getState(null, state_timeout);
        if(rc) {
            if(log.isInfoEnabled()) log.info("state was retrieved successfully, waiting for setState()");
            Boolean result=(Boolean)state_promise.getResult(state_timeout);
            if(result == null) {
                if(log.isErrorEnabled()) log.error("setState() never got called");
            }
            else {
                if(log.isInfoEnabled()) log.info("setState() was called");
            }
        }
        else {
            if(log.isInfoEnabled()) log.info("state could not be retrieved (first member)");
            if(persistent) {
                if(log.isInfoEnabled()) log.info("fetching state from database");
                try {
                    Map m=persistence_mgr.retrieveAll();
                    if(m != null) {
                        Map.Entry entry;
                        Object key, val;
                        for(Iterator it=m.entrySet().iterator(); it.hasNext();) {
                            entry=(Map.Entry)it.next();
                            key=entry.getKey();
                            val=entry.getValue();
                            if(log.isInfoEnabled()) log.info("inserting " + key + " --> " + val);
                            put(key, val);  // will replicate key and value
                        }
                    }
                }
                catch(Throwable ex) {
                    if(log.isErrorEnabled()) log.error("failed creating PersistenceManager, " +
                                "turning persistency off. Exception: " + Util.printStackTrace(ex));
                    persistent=false;
                }
            }
        }
    }


    public Address getLocalAddress()        {return channel != null ? channel.getLocalAddress() : null;}
    public String  getGroupName()           {return groupname;}
    public Channel getChannel()             {return channel;}
    public boolean getPersistent()          {return persistent;}
    public void    setPersistent(boolean p) {persistent=p;}


    public void setDeadlockDetection(boolean flag) {
        if(disp != null)
            disp.setDeadlockDetection(flag);
    }

    public void addNotifier(Notification n) {
        if(!notifs.contains(n))
            notifs.addElement(n);
    }

    public void removeNotifier(Notification n) {
        if(notifs.contains(n))
            notifs.removeElement(n);
    }

    public void stop() {
        if(disp != null) {
            disp.stop();
            disp=null;
        }
        if(channel != null) {
            channel.close();
            channel=null;
        }
    }


	/**
	 * Maps the specified key to the specified value in the hashtable. Neither of both parameters can be null
	 * @param key - the hashtable key
	 * @param value - the value
	 * @return the previous value of the specified key in this hashtable, or null if it did not have one
	 */
    public Object put(Object key, Object value) {
        Object prev_val=get(key);

        //Changes done by <aos>
        //if true, propagate action to the group
        if(send_message == true) {
            try {
                disp.callRemoteMethods(
                        null, "_put", new Object[]{key,value},
                        put_signature,
                        GroupRequest.GET_ALL,
                        0);
            }
            catch(Exception e) {
                //return null;
            }
        }
        else {
            _put(key, value);
            //don't have to do prev_val = super.put(..) as is done at the beginning
        }
        return prev_val;
	}

	/**
	 * Copies all of the mappings from the specified Map to this Hashtable These mappings will replace any mappings that this Hashtable had for any of the keys currently in the specified Map.
	 * @param m - Mappings to be stored in this map
	 */
	public void putAll(Map m) {
		//Changes done by <aos>
		//if true, propagate action to the group
        if(send_message == true) {
            try {
                disp.callRemoteMethods(
                        null, "_putAll", new Object[]{m},
                        putAll_signature,
                        GroupRequest.GET_ALL,
                        0);
            }
            catch(Throwable t) {
            }
        }
        else {
            _putAll(m);
        }