Linux系統Nginx+Tomcat+Codis實現session共享

用戶：root

Tomcat版本：apache-tomcat-7.0.52

接上篇，本篇是將單機Redis替換爲Codis集羣實現session共享。

請先配置Codis集羣，可參考Linux系統Codis集羣安裝配置。

修改Redis源碼

• 修改源碼使用Eclipse+Maven
• pom依賴

• <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>listen</groupId>
    <artifactId>com.listen.org</artifactId>
    <packaging>war</packaging>
    <version>0.0.1-SNAPSHOT</version>
    <name>com.listen.org Maven Webapp</name>
    <url>http://maven.apache.org</url>
    <dependencies>
  <!--     <dependency> -->
  <!--     	<groupId>org.apache.httpcomponents</groupId> -->
  <!--     	<artifactId>httpclient</artifactId> -->
  <!--     	<version>4.3.1</version> -->
  <!--     </dependency> -->
  <!--     <dependency> -->
  <!-- 		<groupId>redis.clients</groupId> -->
  <!-- 		<artifactId>jedis</artifactId> -->
  <!-- 		<version>2.8.0</version> -->
  <!-- 	</dependency> -->
  <!-- 	<dependency> -->
  <!-- 		<groupId>junit</groupId> -->
  <!-- 		<artifactId>junit</artifactId> -->
  <!-- 		<version>4.12</version> -->
  <!-- 	</dependency> -->
  	<dependency>
  	  <groupId>io.codis.jodis</groupId>
  	  <artifactId>jodis</artifactId>
  	  <version>0.3.0</version>
  	</dependency>
  	<dependency>
  		<groupId>org.apache.curator</groupId>
  		<artifactId>curator-framework</artifactId>
  		<version>3.1.0</version>
  	</dependency>
    </dependencies>
    <build>
      <finalName>com.listen.org</finalName>
    </build>
  </project>

• 第三方依賴，因爲修改源碼須要gradle編譯生成的jar包做爲依賴包且這些jar包沒法使用pom依賴的方式添加到工程中，因此只能將dist目錄下的全部jar包以referenced的方式添加到工程中，且沒法使用Maven的install生成class文件，因此只能寫一個main方式，運行，以實現class文件的生成，以下圖

•  編輯源碼，並編譯

• package com.orangefunction.tomcat.redissessions;
  
  import io.codis.jodis.JedisResourcePool;
  import io.codis.jodis.RoundRobinJedisPool;
  
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.EnumSet;
  import java.util.Enumeration;
  import java.util.Iterator;
  import java.util.Set;
  
  import org.apache.catalina.Lifecycle;
  import org.apache.catalina.LifecycleException;
  import org.apache.catalina.LifecycleListener;
  import org.apache.catalina.LifecycleState;
  import org.apache.catalina.Loader;
  import org.apache.catalina.Session;
  import org.apache.catalina.Valve;
  import org.apache.catalina.session.ManagerBase;
  import org.apache.catalina.util.LifecycleSupport;
  import org.apache.juli.logging.Log;
  import org.apache.juli.logging.LogFactory;
  
  import redis.clients.jedis.Jedis;
  import redis.clients.jedis.JedisPoolConfig;
  import redis.clients.jedis.Protocol;
  
  
  public class RedisSessionManager extends ManagerBase implements Lifecycle {
  	
  	public static void main(String[] a){
  		System.out.println("aaa");
  	}
  
    enum SessionPersistPolicy {
      DEFAULT,
      SAVE_ON_CHANGE,
      ALWAYS_SAVE_AFTER_REQUEST;
  
      static SessionPersistPolicy fromName(String name) {
        for (SessionPersistPolicy policy : SessionPersistPolicy.values()) {
          if (policy.name().equalsIgnoreCase(name)) {
            return policy;
          }
        }
        throw new IllegalArgumentException("Invalid session persist policy [" + name + "]. Must be one of " + Arrays.asList(SessionPersistPolicy.values())+ ".");
      }
    }
  
    protected byte[] NULL_SESSION = "null".getBytes();
  
    private final Log log = LogFactory.getLog(RedisSessionManager.class);
  
    protected int timeout = Protocol.DEFAULT_TIMEOUT;
  
    protected RedisSessionHandlerValve handlerValve;
    protected ThreadLocal<RedisSession> currentSession = new ThreadLocal<>();
    protected ThreadLocal<SessionSerializationMetadata> currentSessionSerializationMetadata = new ThreadLocal<>();
    protected ThreadLocal<String> currentSessionId = new ThreadLocal<>();
    protected ThreadLocal<Boolean> currentSessionIsPersisted = new ThreadLocal<>();
    protected Serializer serializer;
  
    protected static String name = "RedisSessionManager";
  
    protected String serializationStrategyClass = "com.orangefunction.tomcat.redissessions.JavaSerializer";
  
    protected EnumSet<SessionPersistPolicy> sessionPersistPoliciesSet = EnumSet.of(SessionPersistPolicy.DEFAULT);
  
    /**
     * The lifecycle event support for this component.
     */
    protected LifecycleSupport lifecycle = new LifecycleSupport(this);
  
  
    public int getTimeout() {
      return timeout;
    }
  
    public void setTimeout(int timeout) {
      this.timeout = timeout;
    }
  
    public void setSerializationStrategyClass(String strategy) {
      this.serializationStrategyClass = strategy;
    }
  
    public String getSessionPersistPolicies() {
      StringBuilder policies = new StringBuilder();
      for (Iterator<SessionPersistPolicy> iter = this.sessionPersistPoliciesSet.iterator(); iter.hasNext();) {
        SessionPersistPolicy policy = iter.next();
        policies.append(policy.name());
        if (iter.hasNext()) {
          policies.append(",");
        }
      }
      return policies.toString();
    }
  
    public void setSessionPersistPolicies(String sessionPersistPolicies) {
      String[] policyArray = sessionPersistPolicies.split(",");
      EnumSet<SessionPersistPolicy> policySet = EnumSet.of(SessionPersistPolicy.DEFAULT);
      for (String policyName : policyArray) {
        SessionPersistPolicy policy = SessionPersistPolicy.fromName(policyName);
        policySet.add(policy);
      }
      this.sessionPersistPoliciesSet = policySet;
    }
  
    public boolean getSaveOnChange() {
      return this.sessionPersistPoliciesSet.contains(SessionPersistPolicy.SAVE_ON_CHANGE);
    }
  
    public boolean getAlwaysSaveAfterRequest() {
      return this.sessionPersistPoliciesSet.contains(SessionPersistPolicy.ALWAYS_SAVE_AFTER_REQUEST);
    }
  
    protected Jedis acquireConnection() {
  //    Jedis jedis = connectionPool.getResource();
  	  JedisPoolConfig poolConfig = new JedisPoolConfig();
  		poolConfig.setMaxTotal(4096);
  		poolConfig.setMaxIdle(200);
  		poolConfig.setMaxWaitMillis(3000);
  		poolConfig.setTestOnBorrow(true);
  		poolConfig.setTestOnReturn(true);
  		poolConfig.setMinIdle(20);
  		JedisResourcePool jedisPool = RoundRobinJedisPool.create()
  				.curatorClient("192.168.75.141:2181,192.168.75.141:2182,192.168.75.141:2183,192.168.75.141:2184,192.168.75.141:2185", 60000)
  				.zkProxyDir("/zk/codis/db_codis/proxy").poolConfig(poolConfig).build();
  		Jedis jedis = jedisPool.getResource();
  		log.info("current jedis:" + jedisToString(jedis));
  		System.out.println("current jedis:" + jedisToString(jedis));
      return jedis;
    }
    
    private String jedisToString(Jedis jedis){
  	  StringBuffer sb = new StringBuffer();
  	  try{
  		  if(jedis != null){
  			  sb.append(jedis.info());
  		  }
  	  }
  	  catch (Exception e){
  		  e.printStackTrace();
  	  }
  	  return sb.toString();
    }
  
    /**
     * Add a lifecycle event listener to this component.
     *
     * @param listener The listener to add
     */
    @Override
    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.
     */
    @Override
    public LifecycleListener[] findLifecycleListeners() {
      return lifecycle.findLifecycleListeners();
    }
  
  
    /**
     * Remove a lifecycle event listener from this component.
     *
     * @param listener The listener to remove
     */
    @Override
    public void removeLifecycleListener(LifecycleListener listener) {
      lifecycle.removeLifecycleListener(listener);
    }
  
    /**
     * Start this component and implement the requirements
     * of {@link org.apache.catalina.util.LifecycleBase#startInternal()}.
     *
     * @exception LifecycleException if this component detects a fatal error
     *  that prevents this component from being used
     */
    @Override
    protected synchronized void startInternal() throws LifecycleException {
      super.startInternal();
  
      setState(LifecycleState.STARTING);
  
      Boolean attachedToValve = false;
      for (Valve valve : getContainer().getPipeline().getValves()) {
        if (valve instanceof RedisSessionHandlerValve) {
          this.handlerValve = (RedisSessionHandlerValve) valve;
          this.handlerValve.setRedisSessionManager(this);
          log.info("Attached to RedisSessionHandlerValve");
          attachedToValve = true;
          break;
        }
      }
  
      if (!attachedToValve) {
        String error = "Unable to attach to session handling valve; sessions cannot be saved after the request without the valve starting properly.";
        log.fatal(error);
        throw new LifecycleException(error);
      }
  
      try {
        initializeSerializer();
      } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
        log.fatal("Unable to load serializer", e);
        throw new LifecycleException(e);
      }
  
      log.info("Will expire sessions after " + getMaxInactiveInterval() + " seconds");
  
      setDistributable(true);
    }
  
  
    /**
     * Stop this component and implement the requirements
     * of {@link org.apache.catalina.util.LifecycleBase#stopInternal()}.
     *
     * @exception LifecycleException if this component detects a fatal error
     *  that prevents this component from being used
     */
    @Override
    protected synchronized void stopInternal() throws LifecycleException {
      if (log.isDebugEnabled()) {
        log.debug("Stopping");
      }
  
      setState(LifecycleState.STOPPING);
  
      // Require a new random number generator if we are restarted
      super.stopInternal();
    }
  
    @Override
    public Session createSession(String requestedSessionId) {
      RedisSession session = null;
      String sessionId = null;
      String jvmRoute = getJvmRoute();
  
      Boolean error = true;
      Jedis jedis = null;
      try {
        jedis = acquireConnection();
  
        // Ensure generation of a unique session identifier.
        if (null != requestedSessionId) {
          sessionId = sessionIdWithJvmRoute(requestedSessionId, jvmRoute);
          if (jedis.setnx(sessionId.getBytes(), NULL_SESSION) == 0L) {
            sessionId = null;
          }
        } else {
          do {
            sessionId = sessionIdWithJvmRoute(generateSessionId(), jvmRoute);
          } while (jedis.setnx(sessionId.getBytes(), NULL_SESSION) == 0L); // 1 = key set; 0 = key already existed
        }
  
        /* Even though the key is set in Redis, we are not going to flag
           the current thread as having had the session persisted since
           the session isn't actually serialized to Redis yet.
           This ensures that the save(session) at the end of the request
           will serialize the session into Redis with 'set' instead of 'setnx'. */
  
        error = false;
  
        if (null != sessionId) {
          session = (RedisSession)createEmptySession();
          session.setNew(true);
          session.setValid(true);
          session.setCreationTime(System.currentTimeMillis());
          session.setMaxInactiveInterval(getMaxInactiveInterval());
          session.setId(sessionId);
          session.tellNew();
        }
  
        currentSession.set(session);
        currentSessionId.set(sessionId);
        currentSessionIsPersisted.set(false);
        currentSessionSerializationMetadata.set(new SessionSerializationMetadata());
  
        if (null != session) {
          try {
            error = saveInternal(jedis, session, true);
          } catch (IOException ex) {
            log.error("Error saving newly created session: " + ex.getMessage());
            currentSession.set(null);
            currentSessionId.set(null);
            session = null;
          }
        }
      } finally {
        if (jedis != null) {
        }
      }
  
      return session;
    }
  
    private String sessionIdWithJvmRoute(String sessionId, String jvmRoute) {
      if (jvmRoute != null) {
        String jvmRoutePrefix = '.' + jvmRoute;
        return sessionId.endsWith(jvmRoutePrefix) ? sessionId : sessionId + jvmRoutePrefix;
      }
      return sessionId;
    }
  
    @Override
    public Session createEmptySession() {
      return new RedisSession(this);
    }
  
    @Override
    public void add(Session session) {
      try {
        save(session);
      } catch (IOException ex) {
        log.warn("Unable to add to session manager store: " + ex.getMessage());
        throw new RuntimeException("Unable to add to session manager store.", ex);
      }
    }
  
    @Override
    public Session findSession(String id) throws IOException {
      RedisSession session = null;
  
      if (null == id) {
        currentSessionIsPersisted.set(false);
        currentSession.set(null);
        currentSessionSerializationMetadata.set(null);
        currentSessionId.set(null);
      } else if (id.equals(currentSessionId.get())) {
        session = currentSession.get();
      } else {
        byte[] data = loadSessionDataFromRedis(id);
        if (data != null) {
          DeserializedSessionContainer container = sessionFromSerializedData(id, data);
          session = container.session;
          currentSession.set(session);
          currentSessionSerializationMetadata.set(container.metadata);
          currentSessionIsPersisted.set(true);
          currentSessionId.set(id);
        } else {
          currentSessionIsPersisted.set(false);
          currentSession.set(null);
          currentSessionSerializationMetadata.set(null);
          currentSessionId.set(null);
        }
      }
  
      return session;
    }
  
    public void clear() {
      Jedis jedis = null;
      Boolean error = true;
      try {
        jedis = acquireConnection();
        jedis.flushDB();
        error = false;
      } finally {
        if (jedis != null) {
        }
      }
    }
  
    public int getSize() throws IOException {
      Jedis jedis = null;
      Boolean error = true;
      try {
        jedis = acquireConnection();
        int size = jedis.dbSize().intValue();
        error = false;
        return size;
      } finally {
        if (jedis != null) {
        }
      }
    }
  
    public String[] keys() throws IOException {
      Jedis jedis = null;
      Boolean error = true;
      try {
        jedis = acquireConnection();
        Set<String> keySet = jedis.keys("*");
        error = false;
        return keySet.toArray(new String[keySet.size()]);
      } finally {
        if (jedis != null) {
        }
      }
    }
  
    public byte[] loadSessionDataFromRedis(String id) throws IOException {
      Jedis jedis = null;
      Boolean error = true;
  
      try {
        log.trace("Attempting to load session " + id + " from Redis");
  
        jedis = acquireConnection();
        byte[] data = jedis.get(id.getBytes());
        error = false;
  
        if (data == null) {
          log.trace("Session " + id + " not found in Redis");
        }
  
        return data;
      } finally {
        if (jedis != null) {
        }
      }
    }
  
    public DeserializedSessionContainer sessionFromSerializedData(String id, byte[] data) throws IOException {
      log.trace("Deserializing session " + id + " from Redis");
  
      if (Arrays.equals(NULL_SESSION, data)) {
        log.error("Encountered serialized session " + id + " with data equal to NULL_SESSION. This is a bug.");
        throw new IOException("Serialized session data was equal to NULL_SESSION");
      }
  
      RedisSession session = null;
      SessionSerializationMetadata metadata = new SessionSerializationMetadata();
  
      try {
        session = (RedisSession)createEmptySession();
  
        serializer.deserializeInto(data, session, metadata);
  
        session.setId(id);
        session.setNew(false);
        session.setMaxInactiveInterval(getMaxInactiveInterval());
        session.access();
        session.setValid(true);
        session.resetDirtyTracking();
  
        if (log.isTraceEnabled()) {
          log.trace("Session Contents [" + id + "]:");
          Enumeration en = session.getAttributeNames();
          while(en.hasMoreElements()) {
            log.trace("  " + en.nextElement());
          }
        }
      } catch (ClassNotFoundException ex) {
        log.fatal("Unable to deserialize into session", ex);
        throw new IOException("Unable to deserialize into session", ex);
      }
  
      return new DeserializedSessionContainer(session, metadata);
    }
  
    public void save(Session session) throws IOException {
      save(session, false);
    }
  
    public void save(Session session, boolean forceSave) throws IOException {
      Jedis jedis = null;
      Boolean error = true;
  
      try {
        jedis = acquireConnection();
        error = saveInternal(jedis, session, forceSave);
      } catch (IOException e) {
        throw e;
      } finally {
        if (jedis != null) {
        }
      }
    }
  
    protected boolean saveInternal(Jedis jedis, Session session, boolean forceSave) throws IOException {
      Boolean error = true;
  
      try {
        log.trace("Saving session " + session + " into Redis");
  
        RedisSession redisSession = (RedisSession)session;
  
        if (log.isTraceEnabled()) {
          log.trace("Session Contents [" + redisSession.getId() + "]:");
          Enumeration en = redisSession.getAttributeNames();
          while(en.hasMoreElements()) {
            log.trace("  " + en.nextElement());
          }
        }
  
        byte[] binaryId = redisSession.getId().getBytes();
  
        Boolean isCurrentSessionPersisted;
        SessionSerializationMetadata sessionSerializationMetadata = currentSessionSerializationMetadata.get();
        byte[] originalSessionAttributesHash = sessionSerializationMetadata.getSessionAttributesHash();
        byte[] sessionAttributesHash = null;
        if (
             forceSave
             || redisSession.isDirty()
             || null == (isCurrentSessionPersisted = this.currentSessionIsPersisted.get())
              || !isCurrentSessionPersisted
             || !Arrays.equals(originalSessionAttributesHash, (sessionAttributesHash = serializer.attributesHashFrom(redisSession)))
           ) {
  
          log.trace("Save was determined to be necessary");
  
          if (null == sessionAttributesHash) {
            sessionAttributesHash = serializer.attributesHashFrom(redisSession);
          }
  
          SessionSerializationMetadata updatedSerializationMetadata = new SessionSerializationMetadata();
          updatedSerializationMetadata.setSessionAttributesHash(sessionAttributesHash);
          log.info("current jedis:" + jedisToString(jedis));
  		System.out.println("current jedis:" + jedisToString(jedis));
          jedis.set(binaryId, serializer.serializeFrom(redisSession, updatedSerializationMetadata));
  
          redisSession.resetDirtyTracking();
          currentSessionSerializationMetadata.set(updatedSerializationMetadata);
          currentSessionIsPersisted.set(true);
        } else {
          log.trace("Save was determined to be unnecessary");
        }
  
        log.trace("Setting expire timeout on session [" + redisSession.getId() + "] to " + getMaxInactiveInterval());
        jedis.expire(binaryId, getMaxInactiveInterval());
  
        error = false;
  
        return error;
      } catch (IOException e) {
        log.error(e.getMessage());
  
        throw e;
      } finally {
        return error;
      }
    }
  
    @Override
    public void remove(Session session) {
      remove(session, false);
    }
  
    @Override
    public void remove(Session session, boolean update) {
      Jedis jedis = null;
      Boolean error = true;
  
      log.trace("Removing session ID : " + session.getId());
  
      try {
        jedis = acquireConnection();
        jedis.del(session.getId());
        error = false;
      } finally {
        if (jedis != null) {
        }
      }
    }
  
    public void afterRequest() {
      RedisSession redisSession = currentSession.get();
      if (redisSession != null) {
        try {
          if (redisSession.isValid()) {
            log.trace("Request with session completed, saving session " + redisSession.getId());
            save(redisSession, getAlwaysSaveAfterRequest());
          } else {
            log.trace("HTTP Session has been invalidated, removing :" + redisSession.getId());
            remove(redisSession);
          }
        } catch (Exception e) {
          log.error("Error storing/removing session", e);
        } finally {
          currentSession.remove();
          currentSessionId.remove();
          currentSessionIsPersisted.remove();
          log.trace("Session removed from ThreadLocal :" + redisSession.getIdInternal());
        }
      }
    }
  
    @Override
    public void processExpires() {
      // We are going to use Redis's ability to expire keys for session expiration.
  
      // Do nothing.
    }
  
  
    private void initializeSerializer() throws ClassNotFoundException, IllegalAccessException, InstantiationException {
      log.info("Attempting to use serializer :" + serializationStrategyClass);
      serializer = (Serializer) Class.forName(serializationStrategyClass).newInstance();
  
      Loader loader = null;
  
      if (getContainer() != null) {
        loader = getContainer().getLoader();
      }
  
      ClassLoader classLoader = null;
  
      if (loader != null) {
        classLoader = loader.getClassLoader();
      }
      serializer.setClassLoader(classLoader);
    }
  
  @Override
  public void load() throws ClassNotFoundException, IOException {
  }
  
  @Override
  public void unload() throws IOException {
  }
  
  
    // Connection Pool Config Accessors
  
    // - from org.apache.commons.pool2.impl.GenericObjectPoolConfig
  }
  
  class DeserializedSessionContainer {
    public final RedisSession session;
    public final SessionSerializationMetadata metadata;
    public DeserializedSessionContainer(RedisSession session, SessionSerializationMetadata metadata) {
      this.session = session;
      this.metadata = metadata;
    }
  }

  替換編譯後的class字節碼文件

• 將編譯後的class文件替換到tomcat-redis-session-manager-master-2.0.0.jar中，是有兩個class文件

• 替換覆蓋jar包

• 將替換後的tomcat-redis-session-manager-master-2.0.0.jar包覆蓋到{tomat_home}/libs中（2個tomcat都要替換）
• 修改tomcat配置文件content.xml

• <?xml version='1.0' encoding='utf-8'?>
  <!--
    Licensed to the Apache Software Foundation (ASF) under one or more
    contributor license agreements.  See the NOTICE file distributed with
    this work for additional information regarding copyright ownership.
    The ASF licenses this file to You under the Apache License, Version 2.0
    (the "License"); you may not use this file except in compliance with
    the License.  You may obtain a copy of the License at
  
        http://www.apache.org/licenses/LICENSE-2.0
  
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
  -->
  <!-- The contents of this file will be loaded for each web application -->
  <Context>
  
      <!-- Default set of monitored resources -->
      <WatchedResource>WEB-INF/web.xml</WatchedResource>
  
      <!-- Uncomment this to disable session persistence across Tomcat restarts -->
      <!--
      <Manager pathname="" />
      -->
  
      <!-- Uncomment this to enable Comet connection tacking (provides events
           on session expiration as well as webapp lifecycle) -->
      <!--
      <Valve className="org.apache.catalina.valves.CometConnectionManagerValve" />
      -->
  
  <!--如下爲增長的內容-->
  <Valve className="com.orangefunction.tomcat.redissessions.RedisSessionHandlerValve" />  
  <Manager className="com.orangefunction.tomcat.redissessions.RedisSessionManager"  
           
           maxInactiveInterval="60" />  
  
  </Context>

• 測試

• 先啓動Codis集羣，集羣的啓動方法可參考Linux系統Codis集羣安裝配置 。而後重啓兩個tomcat，別忘了啓動Nginx。
• 測試session共享，先使用Redis客戶端鏈接到Codis集羣，查詢sessionId是保存在哪臺master上的，而後主從切換對應的master，查看session頁面是否還能正常顯示，結果是能夠正常顯示。
• 總結：修改源碼其實就是修改Redis鏈接的獲取方式，將以前的單機鏈接修改成集羣獲取鏈接的方式，其餘沒有改變。