Tomcat 6 集群配置是否有 useDirtyFlag 选项？

Question

在 Tomcat 5.0.x 中，您可以设置 useDirtyFlag="false" 来在每次请求后强制复制会话，而不是检查 set/removeAttribute 调用。

<Cluster className="org.apache.catalina.cluster.tcp.SimpleTcpCluster"
                 managerClassName="org.apache.catalina.cluster.session.SimpleTcpReplicationManager"
                 expireSessionsOnShutdown="false"
                 **useDirtyFlag="false"**
                 doClusterLog="true"
                 clusterLogName="clusterLog"> ...

server.xml 中的注释指出这可用于执行以下工作：

<%
    HashMap map = (HashMap)session.getAttribute("map");
    map.put("key","value");
%>

即更改已放入会话中的对象的状态，并且您可以确定该对象仍然是复制到集群中的其他节点。

根据 Tomcat 6 文档，您只有两个“管理器”选项 - DeltaManager 和 DeltaManager。 BackupManager ...这两个似乎都不允许此选项或类似的选项。在我测试默认设置时：

  <Cluster className="org.apache.catalina.ha.tcp.SimpleTcpCluster"/>

默认情况下获得 DeltaManager 的地方，它的行为肯定是 useDirtyFlag="true" （正如我所期望的）。

所以我的问题是 - Tomcat 6 中有等效的吗？

查看源代码，我可以看到一个管理器实现“org.apache.catalina.ha.session.SimpleTcpReplicationManager”，它确实具有 useDirtyFlag，但 javadoc 在这种状态下注释它是“Tomcat Session Replication for Tomcat 4.0”...我不知道不知道这是否可以使用 - 我猜不行，因为主集群配置文档中没有提到它。

Answer 1

我在 tomcat-users 邮件列表上发布了本质上相同的问题，对此问题的响应以及 tomcat bugzilla ([43866]) 中的一些信息使我得出以下结论：

1. 没有与 useDirtyFlag 等效的东西，如果您将可变（即更改）对象放入会话中，您需要自定义编码解决方案。
2. Tomcat ClusterValve 似乎是此解决方案的一个有效位置 - 插入集群机制，操作属性以使 DeltaManager 看到会话中的所有属性都已更改。这会强制复制整个会话。

第 1 步：编写 ForceReplicationValve（扩展 ValveBase 实现 ClusterValve）

我不会包含整个类，但会包含关键逻辑部分（删除日志记录和实例检查）：

@Override
public void invoke(Request request, Response response) 
        throws IOException, ServletException {
    getNext().invoke(request, response);
    Session session = request.getSessionInternal();        
    HttpSession deltaSession = (HttpSession) session;
    for (Enumeration<String> names = deltaSession.getAttributeNames(); 
            names.hasMoreElements(); ) {
        String name = names.nextElement();
        deltaSession.setAttribute(name, deltaSession.getAttribute(name));
    }
}

第 2 步：更改集群配置（在 conf/server.xml 中）

<Cluster className="org.apache.catalina.ha.tcp.SimpleTcpCluster"
            channelSendOptions="8">        
    <Valve className="org.apache.catalina.ha.tcp.ForceReplicationValve"/>
    <Valve className="org.apache.catalina.ha.tcp.ReplicationValve"
          filter=".*\.gif;.*\.jpg;.*\.png;.*\.js;.*\.htm;.*\.html;.*\.txt;.*\.css;"/>
    <Valve className="org.apache.catalina.ha.session.JvmRouteBinderValve"/>

    <ClusterListener className="org.apache.catalina.ha.session.JvmRouteSessionIDBinderListener"/>
    <ClusterListener className="org.apache.catalina.ha.session.ClusterSessionListener"/>
</Cluster>

将会话复制到所有集群现在，节点将在每个请求之后发生。

另外：请注意 channelSendOptions 设置。这取代了 Tomcat 5.0.x 中的 replicationMode=asynchronous/synchronous/pooled。有关可能的 int 值，请参阅集群文档。

附录：按要求提供完整阀门源

package org.apache.catalina.ha.tcp;

import java.io.IOException;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.List;

import javax.servlet.ServletException;
import javax.servlet.http.HttpSession;

import org.apache.catalina.Lifecycle;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.LifecycleListener;
import org.apache.catalina.Session;
import org.apache.catalina.connector.Request;
import org.apache.catalina.connector.Response;
import org.apache.catalina.ha.CatalinaCluster;
import org.apache.catalina.ha.ClusterValve;
import org.apache.catalina.ha.session.ReplicatedSession;
import org.apache.catalina.ha.session.SimpleTcpReplicationManager;
import org.apache.catalina.util.LifecycleSupport;
//import org.apache.catalina.util.StringManager;
import org.apache.catalina.valves.ValveBase;

/**
 * <p>With the {@link SimpleTcpReplicationManager} effectively deprecated, this allows
 * mutable objects to be replicated in the cluster by forcing the "dirty" status on 
 * every request.</p> 
 * 
 * @author Jon Brisbin (via post on tomcat-users http://markmail.org/thread/rdo3drcir75dzzrq)
 * @author Kevin Jansz
 */
public class ForceReplicationValve extends ValveBase implements Lifecycle, ClusterValve {
    private static org.apache.juli.logging.Log log =
        org.apache.juli.logging.LogFactory.getLog( ForceReplicationValve.class );

    @SuppressWarnings("hiding")
    protected static final String info = "org.apache.catalina.ha.tcp.ForceReplicationValve/1.0";

// this could be used if ForceReplicationValve messages were setup 
// in org/apache/catalina/ha/tcp/LocalStrings.properties
//    
//    /**
//     * The StringManager for this package.
//     */
//    @SuppressWarnings("hiding")
//    protected static StringManager sm =
//        StringManager.getManager(Constants.Package);

    /** 
     * Not actually required but this must implement {@link ClusterValve} to 
     * be allowed to be added to the Cluster.
     */
    private CatalinaCluster cluster = null ;

    /**
     * Also not really required, implementing {@link Lifecycle} to allow 
     * initialisation and shutdown to be logged. 
     */
    protected LifecycleSupport lifecycle = new LifecycleSupport(this);    


    /**
     * Default constructor
     */
    public ForceReplicationValve() {
        super();
        if (log.isInfoEnabled()) {
            log.info(getInfo() + ": created");
        }
    }

    @Override
    public String getInfo() {
        return info;
    }

    @Override
    public void invoke(Request request, Response response) throws IOException,
            ServletException {

        getNext().invoke(request, response);

        Session session = null;
        try {
            session = request.getSessionInternal();
        } catch (Throwable e) {
            log.error(getInfo() + ": Unable to perform replication request.", e);
        }

        String context = request.getContext().getName();
        String task = request.getPathInfo();
        if(task == null) {
            task = request.getRequestURI();
        }
        if (session != null) {
            if (log.isDebugEnabled()) {
                log.debug(getInfo() + ": [session=" + session.getId() + ", instanceof=" + session.getClass().getName() + ", context=" + context + ", request=" + task + "]");
            }
            if (session instanceof ReplicatedSession) {
                // it's a SimpleTcpReplicationManager - can just set to dirty
                ((ReplicatedSession) session).setIsDirty(true);
                if (log.isDebugEnabled()) {
                    log.debug(getInfo() + ": [session=" + session.getId() + ", context=" + context + ", request=" + task + "] maked DIRTY");
                }
            } else {
                // for everything else - cycle all attributes
                List cycledNames = new LinkedList();

                // in a cluster where the app is <distributable/> this should be
                // org.apache.catalina.ha.session.DeltaSession - implements HttpSession
                HttpSession deltaSession = (HttpSession) session;
                for (Enumeration<String> names = deltaSession.getAttributeNames(); names.hasMoreElements(); ) {
                    String name = names.nextElement();
                    deltaSession.setAttribute(name, deltaSession.getAttribute(name));

                    cycledNames.add(name);                    
                }

                if (log.isDebugEnabled()) {
                    log.debug(getInfo() + ": [session=" + session.getId() + ", context=" + context + ", request=" + task + "] cycled atrributes=" + cycledNames + "");
                }
            }
        } else {
            String id = request.getRequestedSessionId();
            log.warn(getInfo()  + ": [session=" + id + ", context=" + context + ", request=" + task + "] Session not available, unable to send session over cluster.");
        }
    }


    /* 
     * ClusterValve methods - implemented to ensure this valve is not ignored by Cluster  
     */

    public CatalinaCluster getCluster() {
        return cluster;
    }

    public void setCluster(CatalinaCluster cluster) {
        this.cluster = cluster;
    }


    /* 
     * Lifecycle methods - currently implemented just for logging startup 
     */

    /**
     * Add a lifecycle event listener to this component.
     *
     * @param listener The listener to add
     */
    public void addLifecycleListener(LifecycleListener listener) {
        lifecycle.addLifecycleListener(listener);
    }

    /**
     * Get the lifecycle listeners associated with this lifecycle. If this 
     * Lifecycle has no listeners registered, a zero-length array is returned.
     */
    public LifecycleListener[] findLifecycleListeners() {
        return lifecycle.findLifecycleListeners();
    }

    /**
     * Remove a lifecycle event listener from this component.
     *
     * @param listener The listener to remove
     */
    public void removeLifecycleListener(LifecycleListener listener) {
        lifecycle.removeLifecycleListener(listener);
    }

    public void start() throws LifecycleException {
        lifecycle.fireLifecycleEvent(START_EVENT, null);
        if (log.isInfoEnabled()) {
            log.info(getInfo() + ": started");
        }
    }

    public void stop() throws LifecycleException {
        lifecycle.fireLifecycleEvent(STOP_EVENT, null);
        if (log.isInfoEnabled()) {
            log.info(getInfo() + ": stopped");
        }
    }

}

Answer 2

非常感谢 kevinjansz 提供 ForceReplicationValve 的源代码。

我针对Tomcat7进行了调整，如果有人需要的话，这里是：

package org.apache.catalina.ha.tcp;

import java.io.IOException;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.List;

import javax.servlet.ServletException;
import javax.servlet.http.HttpSession;

import org.apache.catalina.Lifecycle;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.LifecycleListener;
import org.apache.catalina.Session;
import org.apache.catalina.connector.Request;
import org.apache.catalina.connector.Response;
import org.apache.catalina.ha.CatalinaCluster;
import org.apache.catalina.ha.ClusterValve;
import org.apache.catalina.util.LifecycleSupport;
import org.apache.catalina.valves.ValveBase;
import org.apache.catalina.LifecycleState;
// import org.apache.tomcat.util.res.StringManager;

/**
 * <p>With the {@link SimpleTcpReplicationManager} effectively deprecated, this allows
 * mutable objects to be replicated in the cluster by forcing the "dirty" status on 
 * every request.</p> 
 * 
 * @author Jon Brisbin (via post on tomcat-users http://markmail.org/thread/rdo3drcir75dzzrq)
 * @author Kevin Jansz
 */
public class ForceReplicationValve extends ValveBase implements Lifecycle, ClusterValve {
    private static org.apache.juli.logging.Log log =
        org.apache.juli.logging.LogFactory.getLog( ForceReplicationValve.class );

    @SuppressWarnings("hiding")
    protected static final String info = "org.apache.catalina.ha.tcp.ForceReplicationValve/1.0";

// this could be used if ForceReplicationValve messages were setup 
// in org/apache/catalina/ha/tcp/LocalStrings.properties
//    
//    /**
//     * The StringManager for this package.
//     */
//    @SuppressWarnings("hiding")
//    protected static StringManager sm =
//        StringManager.getManager(Constants.Package);

    /** 
     * Not actually required but this must implement {@link ClusterValve} to 
     * be allowed to be added to the Cluster.
     */
    private CatalinaCluster cluster = null;

    /**
     * Also not really required, implementing {@link Lifecycle} to allow 
     * initialisation and shutdown to be logged. 
     */
    protected LifecycleSupport lifecycle = new LifecycleSupport(this);    


    /**
     * Default constructor
     */
    public ForceReplicationValve() {
        super();
        if (log.isInfoEnabled()) {
            log.info(getInfo() + ": created");
        }
    }

    @Override
    public String getInfo() {
        return info;
    }

    @Override
    public void invoke(Request request, Response response) throws IOException,
            ServletException {

        getNext().invoke(request, response);

        Session session = null;
        try {
            session = request.getSessionInternal();
        } catch (Throwable e) {
            log.error(getInfo() + ": Unable to perform replication request.", e);
        }

        String context = request.getContext().getName();
        String task = request.getPathInfo();
        if(task == null) {
            task = request.getRequestURI();
        }
        if (session != null) {
            if (log.isDebugEnabled()) {
                log.debug(getInfo() + ": [session=" + session.getId() + ", instanceof=" + session.getClass().getName() + ", context=" + context + ", request=" + task + "]");
            }
            //cycle all attributes
            List<String> cycledNames = new LinkedList<String>();

            // in a cluster where the app is <distributable/> this should be
            // org.apache.catalina.ha.session.DeltaSession - implements HttpSession
            HttpSession deltaSession = (HttpSession) session;
            for (Enumeration<String> names = deltaSession.getAttributeNames(); names.hasMoreElements(); ) {
                String name = names.nextElement();
                deltaSession.setAttribute(name, deltaSession.getAttribute(name));

                cycledNames.add(name);                    
            }

            if (log.isDebugEnabled()) {
                log.debug(getInfo() + ": [session=" + session.getId() + ", context=" + context + ", request=" + task + "] cycled atrributes=" + cycledNames + "");
            }
        } else {
            String id = request.getRequestedSessionId();
            log.warn(getInfo()  + ": [session=" + id + ", context=" + context + ", request=" + task + "] Session not available, unable to send session over cluster.");
        }
    }


    /* 
     * ClusterValve methods - implemented to ensure this valve is not ignored by Cluster  
     */

    public CatalinaCluster getCluster() {
        return cluster;
    }

    public void setCluster(CatalinaCluster cluster) {
        this.cluster = cluster;
    }


    /* 
     * Lifecycle methods - currently implemented just for logging startup 
     */

    /**
     * Add a lifecycle event listener to this component.
     *
     * @param listener The listener to add
     */
    public void addLifecycleListener(LifecycleListener listener) {
        lifecycle.addLifecycleListener(listener);
    }

    /**
     * Get the lifecycle listeners associated with this lifecycle. If this 
     * Lifecycle has no listeners registered, a zero-length array is returned.
     */
    public LifecycleListener[] findLifecycleListeners() {
        return lifecycle.findLifecycleListeners();
    }

    /**
     * Remove a lifecycle event listener from this component.
     *
     * @param listener The listener to remove
     */
    public void removeLifecycleListener(LifecycleListener listener) {
        lifecycle.removeLifecycleListener(listener);
    }

    protected synchronized void startInternal() throws LifecycleException {
        setState(LifecycleState.STARTING);
        if (log.isInfoEnabled()) {
            log.info(getInfo() + ": started");
        }
    }

    protected synchronized void stopInternal() throws LifecycleException {
        setState(LifecycleState.STOPPING);
        if (log.isInfoEnabled()) {
            log.info(getInfo() + ": stopped");
        }
    }

}