如何使用Spring+redis实现对session的分布式管理

Z时代
2024-01-10
分类：综合

在Spring中实现分布式 session管理

本文主要是在Spring中实现分布式session，采用redis对session进行持久化管理，这样当应用部署的时候，不需要在Resin、Tomcat等容器里面进行分布式配置，方便加入新的节点服务器进行集群扩容，session不依赖各节点的服务器，可直接从redis获取。下面是功能的核心代码：

一、首先在web.xml里面配置

加入拦截器：


<!-- 分布式session start -->
  <filter>
    <filter-name>distributedSessionFilter</filter-name>
    <filter-class>DistributedSessionFilter</filter-class>
    <init-param>
      <!-- 必填，密钥.2种方式,1对应为bean，格式为bean:key。2字符串,格式如:afffrfgv-->
      <param-name>key</param-name>
      <param-value>xxxxxxxx</param-value>
    </init-param>
    <init-param>
      <!-- 必填，redis对应的bean,格式为bean:xx-->
      <param-name>cacheBean</param-name>
      <param-value>bean:redisPersistent</param-value>//DistributedBaseInterFace，对应于此接口，进行session的持久化操作
    </init-param>
    <init-param>
      <!-- 必填， -->
      <param-name>cookieName</param-name>
      <param-value>TESTSESSIONID</param-value>
    </init-param>
  </filter>
  <filter-mapping>
    <filter-name>distributedSessionFilter</filter-name>
    <url-pattern>*.do</url-pattern>
  </filter-mapping>
  <!-- 分布式session end -->

二、拦截器的实现，核心代码如下

主要有以下的几个类：

DistributedSessionFilter，

DistributedSessionManager，

DistributedHttpSessionWrapper，

DistributedHttpServletRequestWrapper

1、DistributedSessionFilter实现Filter：


import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import javax.servlet.Filter;
import javax.servlet.FilterChain;
import javax.servlet.FilterConfig;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.context.support.WebApplicationContextUtils;
public class DistributedSessionFilter implements Filter {
  private static final Logger log = LoggerFactory.getLogger(DistributedSessionFilter.class);
  private String cookieName;
  //主要是对session进行管理的操作
  private DistributedSessionManager distributedSessionManager;
  private String key;
}

容器启动时候的初始化方法：


@Override
  public void init(FilterConfig config) throws ServletException {
    WebApplicationContext wac = WebApplicationContextUtils.getRequiredWebApplicationContext(config
        .getServletContext());
    String key = config.getInitParameter("key");
    String cookieName = config.getInitParameter("cookieName");
    String cacheBean = config.getInitParameter("cacheBean");
    // 获取bean的名称，配置是"bean:"
    String redisBeanStr = cacheBean.substring(5);
    DistributedBaseInterFace distributedCache = (DistributedBaseInterFace) wac.getBean(redisBeanStr);
    // 获取key，有2种配置方式,1对应为bean，格式为bean:key。2字符串
    if (key.startsWith("bean:")) {
      this.key = (String) wac.getBean(key.substring(5));
    } else {
      this.key = key;
    }
    this.cookieName = cookieName;
    this.distributedSessionManager = DistributedSessionManager.getInstance(distributedCache);
    //异常处理省略。。。
  }

进行实际的请求拦截：


  @Override
  public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain)
      throws ServletException, IOException {
    DistributedHttpServletRequestWrapper distReq = null;
    try {
      //请求处理
      distReq = createDistributedRequest(servletRequest, servletResponse);
      filterChain.doFilter(distReq, servletResponse);
    } catch (Throwable e) {
      //省略。。。
    } finally {
      if (distReq != null) {
        try {
          //处理完成request后，处理session（主要是保存session会话）
          dealSessionAfterRequest(distReq.getSession());
        } catch (Throwable e2) {
          //省略。。。
        }
      }
    }
  }
  //分布式请求
  private DistributedHttpServletRequestWrapper createDistributedRequest(ServletRequest servletRequest,
      ServletResponse servletResponse) throws IOException, ServletException {
    HttpServletRequest request = (HttpServletRequest) servletRequest;
    HttpServletResponse response = (HttpServletResponse) servletResponse;
    String userSid = CookieUtil.getCookie(cookieName, request);
    String actualSid = distributedSessionManager.getActualSid(userSid, request, key);
    if (StringUtil.isBlank(actualSid)) {
      if (StringUtil.isNotBlank(userSid)) {
        log.info("userSid[{}]验证不通过", userSid);
      }
      // 写cookie
      String[] userSidArr = distributedSessionManager.createUserSid(request, key);
      userSid = userSidArr[0];
      CookieUtil.setCookie(cookieName, userSid, request, response);
      actualSid = userSidArr[1];
    }
    actualSid = "sid:" + actualSid;
    DistributedHttpSessionWrapper distSession = null;
    try {
      Map<String, Object> allAttribute = distributedSessionManager.getSession(actualSid, request.getSession()
          .getMaxInactiveInterval());
      distSession = new DistributedHttpSessionWrapper(actualSid, request.getSession(), allAttribute);
    } catch (Throwable e) {
      // 出错，删掉缓存数据
      log.error(e.getMessage(), e);
      Map<String, Object> allAttribute = new HashMap<String, Object>();
      distSession = new DistributedHttpSessionWrapper(actualSid, request.getSession(), allAttribute);
      distributedSessionManager.removeSession(distSession);
    }
    DistributedHttpServletRequestWrapper requestWrapper = new DistributedHttpServletRequestWrapper(request,
        distSession);
    return requestWrapper;
  }
  // request处理完时操作session
  private void dealSessionAfterRequest(DistributedHttpSessionWrapper session) {
    if (session == null) {
      return;
    }
    if (session.changed) {
      distributedSessionManager.saveSession(session);
    } else if (session.invalidated) {
      distributedSessionManager.removeSession(session);
    } else {
      distributedSessionManager.expire(session);
    }
  }

2、DistributedSessionManager，主要处理分布式session，核心代码：


class DistributedSessionManager {
  protected static final Logger log = LoggerFactory.getLogger(DistributedSessionManager.class);
  private static DistributedSessionManager instance = null;
  //redis处理session的接口，自己根据情况实现
  private DistributedBaseInterFace distributedBaseInterFace;
  private static byte[] lock = new byte[1];
  private DistributedSessionManager(DistributedBaseInterFace distributedBaseInterFace) {
    this.distributedBaseInterFace = distributedBaseInterFace;
  }
  public static DistributedSessionManager getInstance(DistributedBaseInterFace redis) {
    if (instance == null) {
      synchronized (lock) {
        if (instance == null) {
          instance = new DistributedSessionManager(redis);
        }
      }
    }
    return instance;
  }
  //获取session
  public Map<String, Object> getSession(String sid,int second) {
    String json = this.distributedBaseInterFace.get(sid,second);
    if (StringUtil.isNotBlank(json)) {
      return JsonUtil.unserializeMap(json);
    }
    return new HashMap<String, Object>(1);
  }
  //保存session
  public void saveSession(DistributedHttpSessionWrapper session) {
    Map<String, Object> map=session.allAttribute;
    if(MapUtil.isEmpty(map)){
      return;
    }
    String json = JsonUtil.serializeMap(map);
    this.distributedBaseInterFace.set(session.getId(), json, session.getMaxInactiveInterval());
  }
  //删除session
  public void removeSession(DistributedHttpSessionWrapper session) {
    distributedBaseInterFace.del(session.getId());
  }
  public void expire(DistributedHttpSessionWrapper session) {
    distributedBaseInterFace.expire(session.getId(), session.getMaxInactiveInterval());
  }
  /**
   * 创建cookie的sid
   */
  public String[] createUserSid(HttpServletRequest request, String key) {
    //...
  }
  public String getActualSid(String userSid, HttpServletRequest request, String key) {
    //...
  }
}

3、DistributedHttpSessionWrapper 实现了 HttpSession，进行分布式session包装，核心代码：


public class DistributedHttpSessionWrapper implements HttpSession {
  private HttpSession orgiSession;
  private String sid;
  boolean changed = false;
  boolean invalidated = false;
  Map<String, Object> allAttribute;
  public DistributedHttpSessionWrapper(String sid, HttpSession session, Map<String, Object> allAttribute) {
    this.orgiSession = session;
    this.sid = sid;
    this.allAttribute = allAttribute;
  }
  @Override
  public String getId() {
    return this.sid;
  }
  @Override
  public void setAttribute(String name, Object value) {
    changed = true;
    allAttribute.put(name, value);
  }
  @Override
  public Object getAttribute(String name) {
    return allAttribute.get(name);
  }
  @Override
  public Enumeration<String> getAttributeNames() {
    Set<String> set = allAttribute.keySet();
    Iterator<String> iterator = set.iterator();
    return new MyEnumeration<String>(iterator);
  }
  private class MyEnumeration<T> implements Enumeration<T> {
    Iterator<T> iterator;
    public MyEnumeration(Iterator<T> iterator) {
      super();
      this.iterator = iterator;
    }
    @Override
    public boolean hasMoreElements() {
      return iterator.hasNext();
    }
    @Override
    public T nextElement() {
      return iterator.next();
    }
  }
  @Override
  public void invalidate() {
    this.invalidated = true;
  }
  @Override
  public void removeAttribute(String name) {
    changed = true;
    allAttribute.remove(name);
  }
  @Override
  public long getCreationTime() {
    return orgiSession.getCreationTime();
  }
  @Override
  public long getLastAccessedTime() {
    return orgiSession.getLastAccessedTime();
  }
  @Override
  public int getMaxInactiveInterval() {
    return orgiSession.getMaxInactiveInterval();
  }
  @Override
  public ServletContext getServletContext() {
    return orgiSession.getServletContext();
  }
  @Override
  public Object getValue(String arg0) {
    return orgiSession.getValue(arg0);
  }
  @Override
  public String[] getValueNames() {
    return orgiSession.getValueNames();
  }
  @Override
  public boolean isNew() {
    return orgiSession.isNew();
  }
  @Override
  public void putValue(String arg0, Object arg1) {
    orgiSession.putValue(arg0, arg1);
  }
  @Override
  public void removeValue(String arg0) {
    orgiSession.removeValue(arg0);
  }
  @Override
  public void setMaxInactiveInterval(int arg0) {
    orgiSession.setMaxInactiveInterval(arg0);
  }
  @Override
  public HttpSessionContext getSessionContext() {
    return orgiSession.getSessionContext();
  }

4、DistributedHttpServletRequestWrapper 实现了 HttpServletRequestWrapper，包装处理过的session和原始request，核心代码：


public class DistributedHttpServletRequestWrapper extends javax.servlet.http.HttpServletRequestWrapper {
  private HttpServletRequest orgiRequest;
  private DistributedHttpSessionWrapper session;
  public DistributedHttpServletRequestWrapper(HttpServletRequest request, DistributedHttpSessionWrapper session) {
    super(request);
    if (session == null){
      //异常处理。。
    }
    if (request == null){
      //异常处理。。
    }
    this.orgiRequest = request;
    this.session = session;
  }
  public DistributedHttpSessionWrapper getSession(boolean create) {
    orgiRequest.getSession(create);
    return session;
  }
  public DistributedHttpSessionWrapper getSession() {
    return session;
  }
}

5、另外，定义DistributedBaseInterFace接口，用来处理session入redis进行持久化操作：


public interface DistributedBaseInterFace {
  /**
   * 根据key获取缓存数据
   * @param key  
   * @param seconds  
   */
  public String get(String key,int seconds);
  /**
   * 更新缓存数据
   * @param key  
   * @param json
   * @param seconds
   */
  public void set(String key, String json,int seconds);
  /**
   * 删除缓存
   * @param key
   */
  public void del(String key);
  /**
   * 设置过期数据
   * @param key
   * @param seconds
   */
  public void expire(String key,int seconds);