Spring AOP 源码解析

Posted on 2018-03-09 Edited on 2022-01-28 In 源码

首先通过继承BeanDefinitionParser的触发过程看http://www1350.github.io/#post/84

sequencediagram11

通过继承NamespaceHandlerSupport来注册，最核心的类就是AnnotationAwareAspectJAutoProxyCreator

Mybatis/Mybatis-Spring源码

Posted on 2018-03-07 Edited on 2022-01-28 In ORM , 源码

1.SqlSessionTemplate解析

我们要使用的时候通常

1
2
3

<bean id="sqlSessionTemplate" class="org.mybatis.spring.SqlSessionTemplate">
  <constructor-arg ref="sqlSessionFactory" />
</bean>

  public SqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {
    this(sqlSessionFactory, sqlSessionFactory.getConfiguration().getDefaultExecutorType());
  }

  public SqlSessionTemplate(SqlSessionFactory sqlSessionFactory, ExecutorType executorType) {
    this(sqlSessionFactory, executorType,
        new MyBatisExceptionTranslator(
            sqlSessionFactory.getConfiguration().getEnvironment().getDataSource(), true));
  }

public SqlSessionTemplate(SqlSessionFactory sqlSessionFactory, ExecutorType executorType,
      PersistenceExceptionTranslator exceptionTranslator) {

    notNull(sqlSessionFactory, "Property 'sqlSessionFactory' is required");
    notNull(executorType, "Property 'executorType' is required");

    this.sqlSessionFactory = sqlSessionFactory;
    this.executorType = executorType;
    this.exceptionTranslator = exceptionTranslator;
    //动态代理
    this.sqlSessionProxy = (SqlSession) newProxyInstance(
        SqlSessionFactory.class.getClassLoader(),
        new Class[] { SqlSession.class },
        new SqlSessionInterceptor());
  }

我们可以注意到这里是使用代理模式，SqlSessionInterceptor实现了InvocationHandler

private class SqlSessionInterceptor implements InvocationHandler {
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//获取SqlSession
      SqlSession sqlSession = getSqlSession(
          SqlSessionTemplate.this.sqlSessionFactory,
          SqlSessionTemplate.this.executorType,
          SqlSessionTemplate.this.exceptionTranslator);
      try {
//
        Object result = method.invoke(sqlSession, args);
//如果不被spring托管则强制commit
        if (!isSqlSessionTransactional(sqlSession, SqlSessionTemplate.this.sqlSessionFactory)) {
          // force commit even on non-dirty sessions because some databases require
          // a commit/rollback before calling close()
          sqlSession.commit(true);
        }
        return result;
      } catch (Throwable t) {
        Throwable unwrapped = unwrapThrowable(t);
        if (SqlSessionTemplate.this.exceptionTranslator != null && unwrapped instanceof PersistenceException) {
          // release the connection to avoid a deadlock if the translator is no loaded. See issue #22
          closeSqlSession(sqlSession, SqlSessionTemplate.this.sqlSessionFactory);
          sqlSession = null;
          Throwable translated = SqlSessionTemplate.this.exceptionTranslator.translateExceptionIfPossible((PersistenceException) unwrapped);
          if (translated != null) {
            unwrapped = translated;
          }
        }
        throw unwrapped;
      } finally {
        if (sqlSession != null) {
//如果SqlSession是被spring托管则调用SqlSessionHolder#released()释放（其实是通过减小计数器实现的）
          closeSqlSession(sqlSession, SqlSessionTemplate.this.sqlSessionFactory);
        }
      }
    }
  }


public static SqlSession getSqlSession(SqlSessionFactory sessionFactory, ExecutorType executorType, PersistenceExceptionTranslator exceptionTranslator) {

    notNull(sessionFactory, NO_SQL_SESSION_FACTORY_SPECIFIED);
    notNull(executorType, NO_EXECUTOR_TYPE_SPECIFIED);
//根据SqlSessionFactory获取当前线程名为”Transactional resources“的ThreadLocal里面map对应key的 SqlSessionHolder
    SqlSessionHolder holder = (SqlSessionHolder) TransactionSynchronizationManager.getResource(sessionFactory);
//SqlSessionHolder拿出SqlSession
    SqlSession session = sessionHolder(executorType, holder);
    if (session != null) {
      return session;
    }
//SqlSessionHolder里面没有SqlSession则新建一个绑定到线程
    LOGGER.debug(() -> "Creating a new SqlSession");
    session = sessionFactory.openSession(executorType);

    registerSessionHolder(sessionFactory, executorType, exceptionTranslator, session);

    return session;
  }

2. DefaultSqlSession不是线程安全的

SqlSessionManager如何保证线程安全？

他的构造方法是私有的，只能通过newInstance构造

private SqlSessionManager(SqlSessionFactory sqlSessionFactory) {
   this.sqlSessionFactory = sqlSessionFactory;
   this.sqlSessionProxy = (SqlSession) Proxy.newProxyInstance(
       SqlSessionFactory.class.getClassLoader(),
       new Class[]{SqlSession.class},
       new SqlSessionInterceptor());
 }

这里又用了代理

private class SqlSessionInterceptor implements InvocationHandler {
    public SqlSessionInterceptor() {
        // Prevent Synthetic Access
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//注意这里。线程安全通过ThreadLocal实现
      final SqlSession sqlSession = SqlSessionManager.this.localSqlSession.get();
      if (sqlSession != null) {
        try {
          return method.invoke(sqlSession, args);
        } catch (Throwable t) {
          throw ExceptionUtil.unwrapThrowable(t);
        }
      } else {
        final SqlSession autoSqlSession = openSession();
        try {
          final Object result = method.invoke(autoSqlSession, args);
          autoSqlSession.commit();
          return result;
        } catch (Throwable t) {
          autoSqlSession.rollback();
          throw ExceptionUtil.unwrapThrowable(t);
        } finally {
          autoSqlSession.close();
        }
      }
    }
  }

3.DefaultSqlSessionFactory 解析

一般我们这么写

<bean id="mybatisSqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
	<property name="configLocation" value="classpath:mybatis-config.xml" />
	<property name="mapperLocations">
		<list>
			<value>classpath*:sqlmap/**/*.xml</value>
		</list>
	</property>
	<property name="dataSource" ref="dataSource" />

</bean>

我们可以看到SqlSessionFactoryBean的构建，另外通过解析configLocation配置的xml获取Configuration配置

  @Override
  public void afterPropertiesSet() throws Exception {
    notNull(dataSource, "Property 'dataSource' is required");
    notNull(sqlSessionFactoryBuilder, "Property 'sqlSessionFactoryBuilder' is required");
    state((configuration == null && configLocation == null) || !(configuration != null && configLocation != null),
              "Property 'configuration' and 'configLocation' can not specified with together");

//使用this.sqlSessionFactoryBuilder.build(configuration);
    this.sqlSessionFactory = buildSqlSessionFactory();
  }

 public SqlSessionFactory build(Configuration config) {
    return new DefaultSqlSessionFactory(config);
  }

所以我们来分析下DefaultSqlSessionFactory

我们知道SqlSessionTemplate是通过DefaultSqlSessionFactory的openSession获取SqlSession

  @Override
  public SqlSession openSession() {
    return openSessionFromDataSource(configuration.getDefaultExecutorType(), null, false);
  }

  private SqlSession openSessionFromDataSource(ExecutorType execType, TransactionIsolationLevel level, boolean autoCommit) {
    Transaction tx = null;
    try {
      final Environment environment = configuration.getEnvironment();
      final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment);
      tx = transactionFactory.newTransaction(environment.getDataSource(), level, autoCommit);
//通过策略模式获取Executor
      final Executor executor = configuration.newExecutor(tx, execType);
      return new DefaultSqlSession(configuration, executor, autoCommit);
    } catch (Exception e) {
      closeTransaction(tx); // may have fetched a connection so lets call close()
      throw ExceptionFactory.wrapException("Error opening session.  Cause: " + e, e);
    } finally {
      ErrorContext.instance().reset();
    }
  }

<settings>
  <setting name="cacheEnabled" value="true"/>
  <setting name="lazyLoadingEnabled" value="true"/>
  <setting name="multipleResultSetsEnabled" value="true"/>
  <setting name="useColumnLabel" value="true"/>
  <setting name="useGeneratedKeys" value="false"/>
  <setting name="autoMappingBehavior" value="PARTIAL"/>
  <setting name="autoMappingUnknownColumnBehavior" value="WARNING"/>
  <setting name="defaultExecutorType" value="SIMPLE"/>
  <setting name="defaultStatementTimeout" value="25"/>
  <setting name="defaultFetchSize" value="100"/>
  <setting name="safeRowBoundsEnabled" value="false"/>
  <setting name="mapUnderscoreToCamelCase" value="false"/>
  <setting name="localCacheScope" value="SESSION"/>
  <setting name="jdbcTypeForNull" value="OTHER"/>
  <setting name="lazyLoadTriggerMethods"
    value="equals,clone,hashCode,toString"/>
</settings>

Configuration的defaultExecutorType配置，我们回顾下：1.SIMPLE是普通的执行器2.REUSE执行器会重用预处理语句3.BATCH会重用预处理并执行批量更新，我们来看下是如何实现的

  public Executor newExecutor(Transaction transaction, ExecutorType executorType) {
    executorType = executorType == null ? defaultExecutorType : executorType;
    executorType = executorType == null ? ExecutorType.SIMPLE : executorType;
    Executor executor;
    if (ExecutorType.BATCH == executorType) {
      executor = new BatchExecutor(this, transaction);
    } else if (ExecutorType.REUSE == executorType) {
      executor = new ReuseExecutor(this, transaction);
    } else {
      executor = new SimpleExecutor(this, transaction);
    }
    if (cacheEnabled) {
      executor = new CachingExecutor(executor);
    }
    executor = (Executor) interceptorChain.pluginAll(executor);
    return executor;
  }


public enum ExecutorType {
  SIMPLE, REUSE, BATCH
}

看下SqlSessionTemplate在执行update的时候

  @Override
  public int update(String statement, Object parameter) {
    return this.sqlSessionProxy.update(statement, parameter);
  }

//DefaultSqlSession
  @Override
  public int update(String statement, Object parameter) {
    try {
      dirty = true;
      MappedStatement ms = configuration.getMappedStatement(statement);
      return executor.update(ms, wrapCollection(parameter));
    } catch (Exception e) {
      throw ExceptionFactory.wrapException("Error updating database.  Cause: " + e, e);
    } finally {
      ErrorContext.instance().reset();
    }
  }

//Configuration
  public MappedStatement getMappedStatement(String id, boolean validateIncompleteStatements) {
    if (validateIncompleteStatements) {
      buildAllStatements();
    }
    return mappedStatements.get(id);
  }

先看下BATCH，对应BatchExecutor

  @Override
  public int update(MappedStatement ms, Object parameter) throws SQLException {
    ErrorContext.instance().resource(ms.getResource()).activity("executing an update").object(ms.getId());
    if (closed) {
      throw new ExecutorException("Executor was closed.");
    }
    clearLocalCache();
    return doUpdate(ms, parameter);
  }

@Override
  public int doUpdate(MappedStatement ms, Object parameterObject) throws SQLException {
    final Configuration configuration = ms.getConfiguration();
//获得StatementHandler  
    final StatementHandler handler = configuration.newStatementHandler(this, ms, parameterObject, RowBounds.DEFAULT, null, null);
    final BoundSql boundSql = handler.getBoundSql();
//获得Sql语句  
    final String sql = boundSql.getSql();
    final Statement stmt;
    if (sql.equals(currentSql) && ms.equals(currentStatement)) {
//拿到最后一个
      int last = statementList.size() - 1;
      stmt = statementList.get(last);
      applyTransactionTimeout(stmt);
     handler.parameterize(stmt);//fix Issues 322
      BatchResult batchResult = batchResultList.get(last);
      batchResult.addParameterObject(parameterObject);
    } else {
      Connection connection = getConnection(ms.getStatementLog());
      stmt = handler.prepare(connection, transaction.getTimeout());
      handler.parameterize(stmt);    //fix Issues 322
      currentSql = sql;
      currentStatement = ms;
      statementList.add(stmt);
      batchResultList.add(new BatchResult(ms, sql, parameterObject));
    }
  // 使用batch
/***
*
  @Override
  public void batch(Statement statement) throws SQLException {
    PreparedStatement ps = (PreparedStatement) statement;
    ps.addBatch();
  }
*
*/
    handler.batch(stmt);
    return BATCH_UPDATE_RETURN_VALUE;
  }

为什么这么写呢。我们回顾下jdbc的用法

//获取连接
Connection    connection =getConnection();
//不自动 Commit
connection.setAutoCommit(false);   

//预编译
PreparedStatement statement = connection.prepareStatement("INSERT INTO TABLEX VALUES(?, ?)");   

//记录1
statement.setInt(1, 1); 
statement.setString(2, "Cujo"); 
statement.addBatch();   
//记录2
statement.setInt(1, 2); 
statement.setString(2, "Fred"); 
statement.addBatch(); 
//记录3
statement.setInt(1, 3); 
statement.setString(2, "Mark"); 
statement.addBatch();   

//execu
int [] counts = statement.executeBatch();   

//commit
connection.commit();

其中里面事务部分是托管给spring代理的

4.MapperFactoryBean 解析

我们通常会这么使用

<bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
    <property name="basePackage" value="org.mybatis.spring.sample.mapper" />
    <property name="sqlSessionFactoryBeanName" value="sqlSessionFactory" />
</bean>

或

<bean id="baseMapper" class="org.mybatis.spring.mapper.MapperFactoryBean" abstract="true" lazy-init="true">
  <property name="sqlSessionFactory" ref="sqlSessionFactory" />
</bean>
 
<bean id="oneMapper" parent="baseMapper">
  <property name="mapperInterface" value="my.package.MyMapperInterface" />
</bean>
 
<bean id="anotherMapper" parent="baseMapper">
  <property name="mapperInterface" value="my.package.MyAnotherMapperInterface" />
</bean>

MapperFactoryBean,作用就是把接口进行代理切入

public abstract class DaoSupport implements InitializingBean {
	@Override
	public final void afterPropertiesSet() throws IllegalArgumentException, BeanInitializationException {
		checkDaoConfig();

		// Let concrete implementations initialize themselves.
		try {
			initDao();
		}
		catch (Exception ex) {
			throw new BeanInitializationException("Initialization of DAO failed", ex);
		}
	}

	protected abstract void checkDaoConfig() throws IllegalArgumentException;
}


//这里把所有相关的都扫描进去了
 @Override
  protected void checkDaoConfig() {
    super.checkDaoConfig();

    notNull(this.mapperInterface, "Property 'mapperInterface' is required");

    Configuration configuration = getSqlSession().getConfiguration();
    if (this.addToConfig && !configuration.hasMapper(this.mapperInterface)) {
      try {
        configuration.addMapper(this.mapperInterface);
      } catch (Exception e) {
        logger.error("Error while adding the mapper '" + this.mapperInterface + "' to configuration.", e);
        throw new IllegalArgumentException(e);
      } finally {
        ErrorContext.instance().reset();
      }
    }
  }

  @Override
  public T getObject() throws Exception {
    return getSqlSession().getMapper(this.mapperInterface);
  }


//SqlSessionTemplate
  @Override
  public <T> T getMapper(Class<T> type) {
    return getConfiguration().getMapper(type, this);
  }

  public <T> T getMapper(Class<T> type, SqlSession sqlSession) {
    return mapperRegistry.getMapper(type, sqlSession);
  }

//MapperRegistry 这里又是代理
  public <T> T getMapper(Class<T> type, SqlSession sqlSession) {
    final MapperProxyFactory<T> mapperProxyFactory = (MapperProxyFactory<T>) knownMappers.get(type);
    if (mapperProxyFactory == null) {
      throw new BindingException("Type " + type + " is not known to the MapperRegistry.");
    }
    try {
      return mapperProxyFactory.newInstance(sqlSession);
    } catch (Exception e) {
      throw new BindingException("Error getting mapper instance. Cause: " + e, e);
    }
  }

public class MapperProxyFactory<T> {

  private final Class<T> mapperInterface;
  private final Map<Method, MapperMethod> methodCache = new ConcurrentHashMap<Method, MapperMethod>();

  public MapperProxyFactory(Class<T> mapperInterface) {
    this.mapperInterface = mapperInterface;
  }

  public Class<T> getMapperInterface() {
    return mapperInterface;
  }

  public Map<Method, MapperMethod> getMethodCache() {
    return methodCache;
  }

  @SuppressWarnings("unchecked")
  protected T newInstance(MapperProxy<T> mapperProxy) {
    return (T) Proxy.newProxyInstance(mapperInterface.getClassLoader(), new Class[] { mapperInterface }, mapperProxy);
  }

  public T newInstance(SqlSession sqlSession) {
    final MapperProxy<T> mapperProxy = new MapperProxy<T>(sqlSession, mapperInterface, methodCache);
    return newInstance(mapperProxy);
  }

}



public class MapperProxy<T> implements InvocationHandler, Serializable {

  private static final long serialVersionUID = -6424540398559729838L;
  private final SqlSession sqlSession;
  private final Class<T> mapperInterface;
  private final Map<Method, MapperMethod> methodCache;

  public MapperProxy(SqlSession sqlSession, Class<T> mapperInterface, Map<Method, MapperMethod> methodCache) {
    this.sqlSession = sqlSession;
    this.mapperInterface = mapperInterface;
    this.methodCache = methodCache;
  }

  @Override
  public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    try {
      if (Object.class.equals(method.getDeclaringClass())) {
        return method.invoke(this, args);
      } else if (isDefaultMethod(method)) {
        return invokeDefaultMethod(proxy, method, args);
      }
    } catch (Throwable t) {
      throw ExceptionUtil.unwrapThrowable(t);
    }
    final MapperMethod mapperMethod = cachedMapperMethod(method);
    return mapperMethod.execute(sqlSession, args);
  }

  private MapperMethod cachedMapperMethod(Method method) {
    MapperMethod mapperMethod = methodCache.get(method);
    if (mapperMethod == null) {
      mapperMethod = new MapperMethod(mapperInterface, method, sqlSession.getConfiguration());
      methodCache.put(method, mapperMethod);
    }
    return mapperMethod;
  }

  @UsesJava7
  private Object invokeDefaultMethod(Object proxy, Method method, Object[] args)
      throws Throwable {
    final Constructor<MethodHandles.Lookup> constructor = MethodHandles.Lookup.class
        .getDeclaredConstructor(Class.class, int.class);
    if (!constructor.isAccessible()) {
      constructor.setAccessible(true);
    }
    final Class<?> declaringClass = method.getDeclaringClass();
    return constructor
        .newInstance(declaringClass,
            MethodHandles.Lookup.PRIVATE | MethodHandles.Lookup.PROTECTED
                | MethodHandles.Lookup.PACKAGE | MethodHandles.Lookup.PUBLIC)
        .unreflectSpecial(method, declaringClass).bindTo(proxy).invokeWithArguments(args);
  }

  /**
   * Backport of java.lang.reflect.Method#isDefault()
   */
  private boolean isDefaultMethod(Method method) {
    return ((method.getModifiers()
        & (Modifier.ABSTRACT | Modifier.PUBLIC | Modifier.STATIC)) == Modifier.PUBLIC)
        && method.getDeclaringClass().isInterface();
  }
}

最后我们知道方法都通过MapperMethod 了

public Object execute(SqlSession sqlSession, Object[] args) {
    Object result;
    switch (command.getType()) {
      case INSERT: {
      Object param = method.convertArgsToSqlCommandParam(args);
        result = rowCountResult(sqlSession.insert(command.getName(), param));
        break;
      }
      case UPDATE: {
        Object param = method.convertArgsToSqlCommandParam(args);
        result = rowCountResult(sqlSession.update(command.getName(), param));
        break;
      }
      case DELETE: {
        Object param = method.convertArgsToSqlCommandParam(args);
        result = rowCountResult(sqlSession.delete(command.getName(), param));
        break;
      }
      case SELECT:
        if (method.returnsVoid() && method.hasResultHandler()) {
          executeWithResultHandler(sqlSession, args);
          result = null;
        } else if (method.returnsMany()) {
          result = executeForMany(sqlSession, args);
        } else if (method.returnsMap()) {
          result = executeForMap(sqlSession, args);
        } else if (method.returnsCursor()) {
          result = executeForCursor(sqlSession, args);
        } else {
          Object param = method.convertArgsToSqlCommandParam(args);
          result = sqlSession.selectOne(command.getName(), param);
        }
        break;
      case FLUSH:
        result = sqlSession.flushStatements();
        break;
      default:
        throw new BindingException("Unknown execution method for: " + command.getName());
    }
    if (result == null && method.getReturnType().isPrimitive() && !method.returnsVoid()) {
      throw new BindingException("Mapper method '" + command.getName() 
          + " attempted to return null from a method with a primitive return type (" + method.getReturnType() + ").");
    }
    return result;
  }

整个过程如下
sequencediagram1

5.插件源码解析

让我们回顾下http://www1350.github.io/#post/88 的插件写法

<plugins>
     <plugin interceptor="com.xxx.DynamicPlugin">
     </plugin>
 </plugins>

源码如下

//编程api方式加入
public void setPlugins(Interceptor[] plugins) {
    this.plugins = plugins;
  }

    if (!isEmpty(this.plugins)) {
      for (Interceptor plugin : this.plugins) {
        configuration.addInterceptor(plugin);
        LOGGER.debug(() -> "Registered plugin: '" + plugin + "'");
      }
    }

//XML

pluginElement(root.evalNode("plugins"));

  private void pluginElement(XNode parent) throws Exception {
    if (parent != null) {
      for (XNode child : parent.getChildren()) {
        String interceptor = child.getStringAttribute("interceptor");
        Properties properties = child.getChildrenAsProperties();
        Interceptor interceptorInstance = (Interceptor) resolveClass(interceptor).newInstance();
        interceptorInstance.setProperties(properties);
        configuration.addInterceptor(interceptorInstance);
      }
    }
  }


  public void addInterceptor(Interceptor interceptor) {
    interceptorChain.addInterceptor(interceptor);
  }

//责任链模式

public interface Interceptor {

  Object intercept(Invocation invocation) throws Throwable;

  Object plugin(Object target);

  void setProperties(Properties properties);

}


public class InterceptorChain {

  private final List<Interceptor> interceptors = new ArrayList<Interceptor>();

//调用每个实现Interceptor接口的plugin方法
  public Object pluginAll(Object target) {
    for (Interceptor interceptor : interceptors) {
      target = interceptor.plugin(target);
    }
    return target;
  }

  public void addInterceptor(Interceptor interceptor) {
    interceptors.add(interceptor);
  }
  
  public List<Interceptor> getInterceptors() {
    return Collections.unmodifiableList(interceptors);
  }

}


public class Invocation {

  private final Object target;
  private final Method method;
  private final Object[] args;

  public Invocation(Object target, Method method, Object[] args) {
    this.target = target;
    this.method = method;
    this.args = args;
  }

  public Object getTarget() {
    return target;
  }

  public Method getMethod() {
    return method;
  }

  public Object[] getArgs() {
    return args;
  }

  public Object proceed() throws InvocationTargetException, IllegalAccessException {
    return method.invoke(target, args);
  }

}



public class Plugin implements InvocationHandler {

  private final Object target;
  private final Interceptor interceptor;
  private final Map<Class<?>, Set<Method>> signatureMap;

  private Plugin(Object target, Interceptor interceptor, Map<Class<?>, Set<Method>> signatureMap) {
    this.target = target;
    this.interceptor = interceptor;
    this.signatureMap = signatureMap;
  }

//每个使用Plugin.wrap 的时候将会代理那个类，然后将会调用interceptor的intercept方法
  public static Object wrap(Object target, Interceptor interceptor) {
    Map<Class<?>, Set<Method>> signatureMap = getSignatureMap(interceptor);
    Class<?> type = target.getClass();
    Class<?>[] interfaces = getAllInterfaces(type, signatureMap);
    if (interfaces.length > 0) {
      return Proxy.newProxyInstance(
          type.getClassLoader(),
          interfaces,
          new Plugin(target, interceptor, signatureMap));
    }
    return target;
  }

  @Override
  public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    try {
      Set<Method> methods = signatureMap.get(method.getDeclaringClass());
//被Signature设置的调用的时候才调用intercept，否则调用原方法
      if (methods != null && methods.contains(method)) {
        return interceptor.intercept(new Invocation(target, method, args));
      }
      return method.invoke(target, args);
    } catch (Exception e) {
      throw ExceptionUtil.unwrapThrowable(e);
    }
  }

  private static Map<Class<?>, Set<Method>> getSignatureMap(Interceptor interceptor) {
    Intercepts interceptsAnnotation = interceptor.getClass().getAnnotation(Intercepts.class);
    // issue #251
    if (interceptsAnnotation == null) {
      throw new PluginException("No @Intercepts annotation was found in interceptor " + interceptor.getClass().getName());      
    }
    Signature[] sigs = interceptsAnnotation.value();
    Map<Class<?>, Set<Method>> signatureMap = new HashMap<Class<?>, Set<Method>>();
    for (Signature sig : sigs) {
      Set<Method> methods = signatureMap.get(sig.type());
      if (methods == null) {
        methods = new HashSet<Method>();
        signatureMap.put(sig.type(), methods);
      }
      try {
        Method method = sig.type().getMethod(sig.method(), sig.args());
        methods.add(method);
      } catch (NoSuchMethodException e) {
        throw new PluginException("Could not find method on " + sig.type() + " named " + sig.method() + ". Cause: " + e, e);
      }
    }
    return signatureMap;
  }

  private static Class<?>[] getAllInterfaces(Class<?> type, Map<Class<?>, Set<Method>> signatureMap) {
    Set<Class<?>> interfaces = new HashSet<Class<?>>();
    while (type != null) {
      for (Class<?> c : type.getInterfaces()) {
        if (signatureMap.containsKey(c)) {
          interfaces.add(c);
        }
      }
      type = type.getSuperclass();
    }
    return interfaces.toArray(new Class<?>[interfaces.size()]);
  }

}

最后我们知道，入口都在pluginAll，我们看下哪里调用了

public ParameterHandler newParameterHandler(MappedStatement mappedStatement, Object parameterObject, BoundSql boundSql) {
  ParameterHandler parameterHandler = mappedStatement.getLang().createParameterHandler(mappedStatement, parameterObject, boundSql);
  parameterHandler = (ParameterHandler) interceptorChain.pluginAll(parameterHandler);
  return parameterHandler;
}

public ResultSetHandler newResultSetHandler(Executor executor, MappedStatement mappedStatement, RowBounds rowBounds, ParameterHandler parameterHandler,
    ResultHandler resultHandler, BoundSql boundSql) {
  ResultSetHandler resultSetHandler = new DefaultResultSetHandler(executor, mappedStatement, parameterHandler, resultHandler, boundSql, rowBounds);
  resultSetHandler = (ResultSetHandler) interceptorChain.pluginAll(resultSetHandler);
  return resultSetHandler;
}

public StatementHandler newStatementHandler(Executor executor, MappedStatement mappedStatement, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql) {
  StatementHandler statementHandler = new RoutingStatementHandler(executor, mappedStatement, parameterObject, rowBounds, resultHandler, boundSql);
  statementHandler = (StatementHandler) interceptorChain.pluginAll(statementHandler);
  return statementHandler;
}

而newStatementHandler等方法则用在执行器上，所以一路就通了。
SimpleExecutor

@Override
 public int doUpdate(MappedStatement ms, Object parameter) throws SQLException {
   Statement stmt = null;
   try {
     Configuration configuration = ms.getConfiguration();
     StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);
     stmt = prepareStatement(handler, ms.getStatementLog());
     return handler.update(stmt);
   } finally {
     closeStatement(stmt);
   }
 }

插件后源码如下：
sequencediagram2

6.缓存

先看下缓存Cache 类接口

实现类：

一级缓存

我们知道每个sqlSession都有自己的BaseExecutor，每个BaseExecutor都有自己的Local Cache。所以一级缓存是基于sqlSession级别的

一级缓存的配置，共有两个选项，SESSION或者STATEMENT，默认是SESSION级别。

<setting name="localCacheScope" value="SESSION"/>

//BaseExecutor 的query 构建CacheKey
CacheKey key = createCacheKey(ms, parameter, rowBounds, boundSql);

//可以认为key是Statement Id + Offset + Limmit + Sql + Params
@Override
  public CacheKey createCacheKey(MappedStatement ms, Object parameterObject, RowBounds rowBounds, BoundSql boundSql) {
    if (closed) {
      throw new ExecutorException("Executor was closed.");
    }
    CacheKey cacheKey = new CacheKey();
    cacheKey.update(ms.getId());
    cacheKey.update(rowBounds.getOffset());
    cacheKey.update(rowBounds.getLimit());
    cacheKey.update(boundSql.getSql());
    List<ParameterMapping> parameterMappings = boundSql.getParameterMappings();
    TypeHandlerRegistry typeHandlerRegistry = ms.getConfiguration().getTypeHandlerRegistry();
    // mimic DefaultParameterHandler logic
    for (ParameterMapping parameterMapping : parameterMappings) {
      if (parameterMapping.getMode() != ParameterMode.OUT) {
        Object value;
        String propertyName = parameterMapping.getProperty();
        if (boundSql.hasAdditionalParameter(propertyName)) {
          value = boundSql.getAdditionalParameter(propertyName);
        } else if (parameterObject == null) {
          value = null;
        } else if (typeHandlerRegistry.hasTypeHandler(parameterObject.getClass())) {
          value = parameterObject;
        } else {
          MetaObject metaObject = configuration.newMetaObject(parameterObject);
          value = metaObject.getValue(propertyName);
        }
        cacheKey.update(value);
      }
    }
    if (configuration.getEnvironment() != null) {
      // issue #176
      cacheKey.update(configuration.getEnvironment().getId());
    }
    return cacheKey;
  }

//CacheKey 
  public void update(Object object) {
    int baseHashCode = object == null ? 1 : ArrayUtil.hashCode(object); 

    count++;
    checksum += baseHashCode;
    baseHashCode *= count;

    hashcode = multiplier * hashcode + baseHashCode;

    updateList.add(object);
  }

查询的时候先生成cacheKey，然后使用CacheKey先去缓存查找

@SuppressWarnings("unchecked")
  @Override
  public <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey key, BoundSql boundSql) throws SQLException {
...
    if (queryStack == 0 && ms.isFlushCacheRequired()) {
      clearLocalCache();
    }
    List<E> list;
    try {
      queryStack++;
//这里获取
      list = resultHandler == null ? (List<E>) localCache.getObject(key) : null;
      if (list != null) {
        handleLocallyCachedOutputParameters(ms, key, parameter, boundSql);
      } else {
//从数据库取，同时写入localCache
        list = queryFromDatabase(ms, parameter, rowBounds, resultHandler, key, boundSql);
      }
    } finally {
      queryStack--;
    }
    if (queryStack == 0) {
      for (DeferredLoad deferredLoad : deferredLoads) {
        deferredLoad.load();
      }
      // issue #601
      deferredLoads.clear();
      if (configuration.getLocalCacheScope() == LocalCacheScope.STATEMENT) {
        // issue #482
        clearLocalCache();
      }
    }
    return list;
  }


private void handleLocallyCachedOutputParameters(MappedStatement ms, CacheKey key, Object parameter, BoundSql boundSql) {
    if (ms.getStatementType() == StatementType.CALLABLE) {
      final Object cachedParameter = localOutputParameterCache.getObject(key);
      if (cachedParameter != null && parameter != null) {
        final MetaObject metaCachedParameter = configuration.newMetaObject(cachedParameter);
        final MetaObject metaParameter = configuration.newMetaObject(parameter);
        for (ParameterMapping parameterMapping : boundSql.getParameterMappings()) {
          if (parameterMapping.getMode() != ParameterMode.IN) {
            final String parameterName = parameterMapping.getProperty();
            final Object cachedValue = metaCachedParameter.getValue(parameterName);
            metaParameter.setValue(parameterName, cachedValue);
          }
        }
      }
    }
  }

  private <E> List<E> queryFromDatabase(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey key, BoundSql boundSql) throws SQLException {
    List<E> list;
    localCache.putObject(key, EXECUTION_PLACEHOLDER);
    try {
      list = doQuery(ms, parameter, rowBounds, resultHandler, boundSql);
    } finally {
      localCache.removeObject(key);
    }
    localCache.putObject(key, list);
    if (ms.getStatementType() == StatementType.CALLABLE) {
      localOutputParameterCache.putObject(key, parameter);
    }
    return list;
  }

我们看到BaseExecutor类里有

  protected ConcurrentLinkedQueue<DeferredLoad> deferredLoads;

private static class DeferredLoad {

    private final MetaObject resultObject;
    private final String property;
    private final Class<?> targetType;
    private final CacheKey key;
    private final PerpetualCache localCache;
    private final ObjectFactory objectFactory;
    private final ResultExtractor resultExtractor;

    // issue #781
    public DeferredLoad(MetaObject resultObject,
                        String property,
                        CacheKey key,
                        PerpetualCache localCache,
                        Configuration configuration,
                        Class<?> targetType) {
      this.resultObject = resultObject;
      this.property = property;
      this.key = key;
      this.localCache = localCache;
      this.objectFactory = configuration.getObjectFactory();
      this.resultExtractor = new ResultExtractor(configuration, objectFactory);
      this.targetType = targetType;
    }

    public boolean canLoad() {
      return localCache.getObject(key) != null && localCache.getObject(key) != EXECUTION_PLACEHOLDER;
    }

    public void load() {
      @SuppressWarnings( "unchecked" )
      // we suppose we get back a List
      List<Object> list = (List<Object>) localCache.getObject(key);
      Object value = resultExtractor.extractObjectFromList(list, targetType);
      resultObject.setValue(property, value);
    }

  }

https://tech.meituan.com/mybatis_cache.html

7.结果集处理器ResultSetHandler

public ResultSetHandler newResultSetHandler(Executor executor, MappedStatement mappedStatement, RowBounds rowBounds, ParameterHandler parameterHandler,
    ResultHandler resultHandler, BoundSql boundSql) {
  ResultSetHandler resultSetHandler = new DefaultResultSetHandler(executor, mappedStatement, parameterHandler, resultHandler, boundSql, rowBounds);
  resultSetHandler = (ResultSetHandler) interceptorChain.pluginAll(resultSetHandler);
  return resultSetHandler;
}

最后一般是

@Override
public <E> List<E> query(Statement statement, ResultHandler resultHandler) throws SQLException {
  String sql = boundSql.getSql();
  statement.execute(sql);
  return resultSetHandler.<E>handleResultSets(statement);
}

@Override
  public List<Object> handleResultSets(Statement stmt) throws SQLException {
    ErrorContext.instance().activity("handling results").object(mappedStatement.getId());

    final List<Object> multipleResults = new ArrayList<Object>();

    int resultSetCount = 0;
//ResultSet rs = stmt.getResultSet(); 然后转化为new ResultSetWrapper(rs, configuration)
    ResultSetWrapper rsw = getFirstResultSet(stmt);
//配置文件中读取ResultMap
    List<ResultMap> resultMaps = mappedStatement.getResultMaps();
    int resultMapCount = resultMaps.size();
    validateResultMapsCount(rsw, resultMapCount);
    while (rsw != null && resultMapCount > resultSetCount) {
      ResultMap resultMap = resultMaps.get(resultSetCount);
//通过TypeHandler找到对应映射器，映射后放入multipleResults
      handleResultSet(rsw, resultMap, multipleResults, null);
      rsw = getNextResultSet(stmt);
      cleanUpAfterHandlingResultSet();
      resultSetCount++;
    }

    String[] resultSets = mappedStatement.getResultSets();
    if (resultSets != null) {
      while (rsw != null && resultSetCount < resultSets.length) {
        ResultMapping parentMapping = nextResultMaps.get(resultSets[resultSetCount]);
        if (parentMapping != null) {
          String nestedResultMapId = parentMapping.getNestedResultMapId();
          ResultMap resultMap = configuration.getResultMap(nestedResultMapId);
          handleResultSet(rsw, resultMap, null, parentMapping);
        }
        rsw = getNextResultSet(stmt);
        cleanUpAfterHandlingResultSet();
        resultSetCount++;
      }
    }
//如果只有一个映射，返回第一个  
    return collapseSingleResultList(multipleResults);
  }

Java字节码操纵

Posted on 2017-12-23 Edited on 2022-01-28 In 基础

什么是字节码？

机器码机器码(machine code)是CPU可直接解读的指令。机器码与硬件等有关，不同的CPU架构支持的硬件码也不相同。
字节码字节码（bytecode）是一种包含执行程序、由一序列 op 代码/数据对组成的二进制文件。字节码是一种中间码，它比机器码更抽象，需要直译器转译后才能成为机器码的中间代码。通常情况下它是已经经过编译，但与特定机器码无关。字节码主要为了实现特定软件运行和软件环境、与硬件环境无关。字节码的实现方式是通过编译器和虚拟机器。编译器将源码编译成字节码，特定平台上的虚拟机器将字节码转译为可以直接执行的指令。而在Java里，通过类加载器把字节码读入加载并转换成 java.lang.Class类的一个实例。

类文件结构

一个编译后的类文件包含下面的结构：

ClassFile {
    u4            magic;
    u2            minor_version;
    u2            major_version;
    u2            constant_pool_count;
    cp_info        contant_pool[constant_pool_count – 1];
    u2            access_flags;
    u2            this_class;
    u2            super_class;
    u2            interfaces_count;
    u2            interfaces[interfaces_count];
    u2            fields_count;
    field_info        fields[fields_count];
    u2            methods_count;
    method_info        methods[methods_count];
    u2            attributes_count;
    attribute_info    attributes[attributes_count];
}

字符	描述
magic, minor_version, major_version	类文件的版本信息和用于编译这个类的 JDK 版本。
constant_pool	类似于符号表，尽管它包含更多数据。下面有更多的详细描述。
access_flags	提供这个类的描述符列表。
this_class	提供这个类全名的常量池(constant_pool)索引，比如org/jamesdbloom/foo/Bar。
super_class	提供这个类的父类符号引用的常量池索引。
interfaces	指向常量池的索引数组，提供那些被实现的接口的符号引用。
fields	提供每个字段完整描述的常量池索引数组。
methods	指向constant_pool的索引数组，用于表示每个方法签名的完整描述。如果这个方法不是抽象方法也不是 native 方法，那么就会显示这个函数的字节码。
attributes	不同值的数组，表示这个类的附加信息，包括 RetentionPolicy.CLASS 和 RetentionPolicy.RUNTIME 注解。

举个简单的例子（例1.1）：

public class HelloWorld {
    public static void main(String[] args) {
        System.out.println("Hello World!");
    }
}

对应的字节码如下：

public class com.souche.rick.HelloWorld
  minor version: 0/*Java 类文件的版本信息*/
  major version: 52
  flags: ACC_PUBLIC, ACC_SUPER
Constant pool:/*该项存放了类中各种文字字符串、类名、方法名和接口名称、final 变量以及对外部类的引用信息等常量。虚拟机必须为每一个被装载的类维护一个常量池，常量池中存储了相应类型所用到的所有类型、字段和方法的符号引用。*/
   #1 = Methodref          #6.#20         // java/lang/Object."<init>":()V
   #2 = Fieldref           #21.#22        // java/lang/System.out:Ljava/io/PrintStream;
   #3 = String             #23            // Hello World!
   #4 = Methodref          #24.#25        // java/io/PrintStream.println:(Ljava/lang/String;)V
   #5 = Class              #26            // com/souche/rick/HelloWorld
   #6 = Class              #27            // java/lang/Object
   #7 = Utf8               <init>
   #8 = Utf8               ()V
   #9 = Utf8               Code
  #10 = Utf8               LineNumberTable/*为调试器提供源码中的每一行对应的字节码信息*/
  #11 = Utf8               LocalVariableTable /*列出了所有栈帧中的局部变量。这里唯一的局部变量就是 this。*/
  #12 = Utf8               this
  #13 = Utf8               Lcom/souche/rick/HelloWorld;
  #14 = Utf8               main
  #15 = Utf8               ([Ljava/lang/String;)V
  #16 = Utf8               args
  #17 = Utf8               [Ljava/lang/String;
  #18 = Utf8               SourceFile
  #19 = Utf8               HelloWorld.java
  #20 = NameAndType        #7:#8          // "<init>":()V
  #21 = Class              #28            // java/lang/System
  #22 = NameAndType        #29:#30        // out:Ljava/io/PrintStream;
  #23 = Utf8               Hello World!
  #24 = Class              #31            // java/io/PrintStream
  #25 = NameAndType        #32:#33        // println:(Ljava/lang/String;)V
  #26 = Utf8               com/souche/rick/HelloWorld
  #27 = Utf8               java/lang/Object
  #28 = Utf8               java/lang/System
  #29 = Utf8               out
  #30 = Utf8               Ljava/io/PrintStream;
  #31 = Utf8               java/io/PrintStream
  #32 = Utf8               println
  #33 = Utf8               (Ljava/lang/String;)V
{
  public com.souche.rick.HelloWorld();
    descriptor: ()V
    flags: ACC_PUBLIC
    Code:
      stack=1, locals=1, args_size=1
         0: aload_0
         1: invokespecial #1                  // Method java/lang/Object."<init>":()V
         4: return
      LineNumberTable:
        line 7: 0
      LocalVariableTable:
        Start  Length  Slot  Name   Signature
            0       5     0  this   Lcom/souche/rick/HelloWorld;

  public static void main(java.lang.String[]);
    descriptor: ([Ljava/lang/String;)V
    flags: ACC_PUBLIC, ACC_STATIC
    Code:
      stack=2, locals=1, args_size=1
         0: getstatic     #2                  // Field java/lang/System.out:Ljava/io/PrintStream; /*获取指定类的静态域，并将其值压入栈顶*/
         3: ldc           #3                  // String Hello World! /*将"Hello World!"从常量池中推送至栈顶*/
         5: invokevirtual #4                  // Method java/io/PrintStream.println:(Ljava/lang/String;)V /*调用实例方法println*/
         8: return /*从当前方法返回void*/
      LineNumberTable:
        line 9: 0
        line 10: 8
      LocalVariableTable:
        Start  Length  Slot  Name   Signature
            0       9     0  args   [Ljava/lang/String;
}

注：编译的时候选用-g参数，否则LineNumberTable参数默认是不显示的，maven编译的时候默认是使用-g参数所以不用担心

描述符标识字符含义

标识字符	含义	标识字符	含义
B	基本类型byte	J	基本类型long
C	基本类型char	S	基本类型short
D	基本类型double	Z	基本类型boolean
F	基本类型float	V	特殊类型void
I	基本类型int	L	对象类型，如Ljava/lang/Object

对于数组类型，每一唯独使用一个前置的“[”字符描述，如一个定义为“java.lang.String[][]”类型的二维数组，将被记录为“[[Ljava/lang/String”。

当描述符描述方法时，按照先参数列表后返回值的顺序描述，参数列表按照参数的严格顺序放在一组小括号“()”之内。如int getResult()方法的描述符为“()I”。

Method declaration in source file	Method descriptor
void m(int i, float f)	(IF)V
int m(Object o)	(Ljava/lang/Object;)I
int[] m(int i, String s)	(ILjava/lang/String;)[I
Object m(int[] i)	([I)Ljava/lang/Object;

关于助记符https://www.cnblogs.com/anbylau2130/p/6078427.html

简单了解了一些基本概念以后，我们正式进入主题：字节码操纵

原理

用一些字节码操纵框架修改字节码
自定义ClassLoader来加载修改后的字节码

其实还有另外一种形式，就是直接换掉原来的字节码。一种是在JVM加载用户的Class时，拦截，返回修改后的字节码。另外一种在运行时，使用Instrumentation.redefineClasses方法来替换掉原来的字节码，和这个类相关的实例立即生效。不过这种形式需要采用agent形式启动。有兴趣的同学可以自行搜索下相关内容。

操纵字节码的场景

操纵字节码可以在字节码被载入类加载器之前修改字节码二进制文件，从而做到一些技术上难以实现的功能，如：AOP增强、性能分析、调试跟踪、日志记录、bug定位、混淆代码，甚至连Scala、Groovy和Grails等JVM语言都用到大量的操纵字节码技术。

字节码操纵框架

ASM使用

ASM主要通过树这种数据结构来表示复杂的字节码结构，通过Visitor设计模式来实现(详细见手册)

生成一个类
上面的例1.1代码可以用ASM生成：

package com.souche.rick.asm;

import org.objectweb.asm.ClassWriter;
import org.objectweb.asm.Label;
import org.objectweb.asm.MethodVisitor;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import static org.objectweb.asm.Opcodes.*;

public class HelloWorld {
    public static void main(String[] args) throws IOException {
        byte[] bytes = builder();
        File file=new File("com"+File.separator+"souche"+File.separator+"rick", "HelloWorld.class");
        file.getParentFile().mkdirs();
        FileOutputStream fileOutputStream = new FileOutputStream(file);
        fileOutputStream.write(bytes);
        fileOutputStream.close();

    }

    public static byte[] builder(){
        ClassWriter cw = new ClassWriter(0);
        cw.visit(V1_8, ACC_PUBLIC  + ACC_SUPER,
                "com/souche/rick/HelloWorld", null, "java/lang/Object",
                null);
        MethodVisitor classInitMv = cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null);
        classInitMv.visitCode();
        Label l0 = new Label();
        classInitMv.visitLabel(l0);
        classInitMv.visitLineNumber(9, l0);
        classInitMv.visitVarInsn(ALOAD,0);
        classInitMv.visitMethodInsn(INVOKESPECIAL, "java/lang/Object", "<init>", "()V", false);
        classInitMv.visitInsn(RETURN);
        Label l1 = new Label();
        classInitMv.visitLabel(l1);
        classInitMv.visitLocalVariable("this", "Lcom/souche/rick/HelloWorld;", null, l0, l1, 0);
        classInitMv.visitMaxs(1, 1);
        classInitMv.visitEnd();

        MethodVisitor mainMv = cw.visitMethod(ACC_PUBLIC + ACC_STATIC, "main",
                "([Ljava/lang/String;)V", null, null);
        mainMv.visitCode();
        Label l2 = new Label();
        mainMv.visitLabel(l2);
        mainMv.visitLineNumber(13, l2);
        mainMv.visitFieldInsn(GETSTATIC, "java/lang/System", "out",
                "Ljava/io/PrintStream;");
        mainMv.visitLdcInsn("Hello World!");

        mainMv.visitMethodInsn(INVOKEVIRTUAL, "java/io/PrintStream", "println", "(Ljava/lang/String;)V", false);
        Label l3 = new Label();
        mainMv.visitLabel(l3);
        mainMv.visitLineNumber(15,l3);
        mainMv.visitInsn(RETURN);
        mainMv.visitLocalVariable("args", "[Ljava/lang/String;", null, l2, l3, 0);
        mainMv.visitMaxs(2, 1);
        mainMv.visitEnd();
        cw.visitEnd();
        byte[] b = cw.toByteArray();
        return b;
    }
}

效果：

public class com.souche.rick.HelloWorld
  minor version: 0
  major version: 52
  flags: ACC_PUBLIC, ACC_SUPER
Constant pool:
   #1 = Utf8               com/souche/rick/HelloWorld
   #2 = Class              #1             // com/souche/rick/HelloWorld
   #3 = Utf8               java/lang/Object
   #4 = Class              #3             // java/lang/Object
   #5 = Utf8               <init>
   #6 = Utf8               ()V
   #7 = NameAndType        #5:#6          // "<init>":()V
   #8 = Methodref          #4.#7          // java/lang/Object."<init>":()V
   #9 = Utf8               this
  #10 = Utf8               Lcom/souche/rick/HelloWorld;
  #11 = Utf8               main
  #12 = Utf8               ([Ljava/lang/String;)V
  #13 = Utf8               java/lang/System
  #14 = Class              #13            // java/lang/System
  #15 = Utf8               out
  #16 = Utf8               Ljava/io/PrintStream;
  #17 = NameAndType        #15:#16        // out:Ljava/io/PrintStream;
  #18 = Fieldref           #14.#17        // java/lang/System.out:Ljava/io/PrintStream;
  #19 = Utf8               Hello World!
  #20 = String             #19            // Hello World!
  #21 = Utf8               java/io/PrintStream
  #22 = Class              #21            // java/io/PrintStream
  #23 = Utf8               println
  #24 = Utf8               (Ljava/lang/String;)V
  #25 = NameAndType        #23:#24        // println:(Ljava/lang/String;)V
  #26 = Methodref          #22.#25        // java/io/PrintStream.println:(Ljava/lang/String;)V
  #27 = Utf8               args
  #28 = Utf8               [Ljava/lang/String;
  #29 = Utf8               Code
  #30 = Utf8               LocalVariableTable
  #31 = Utf8               LineNumberTable
{
  public com.souche.rick.HelloWorld();
    descriptor: ()V
    flags: ACC_PUBLIC
    Code:
      stack=1, locals=1, args_size=1
         0: aload_0
         1: invokespecial #8                  // Method java/lang/Object."<init>":()V
         4: return
      LocalVariableTable:
        Start  Length  Slot  Name   Signature
            0       5     0  this   Lcom/souche/rick/HelloWorld;
      LineNumberTable:
        line 9: 0

  public static void main(java.lang.String[]);
    descriptor: ([Ljava/lang/String;)V
    flags: ACC_PUBLIC, ACC_STATIC
    Code:
      stack=2, locals=1, args_size=1
         0: getstatic     #18                 // Field java/lang/System.out:Ljava/io/PrintStream;
         3: ldc           #20                 // String Hello World!
         5: invokevirtual #26                 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
         8: return
      LocalVariableTable:
        Start  Length  Slot  Name   Signature
            0       8     0  args   [Ljava/lang/String;
      LineNumberTable:
        line 13: 0
        line 15: 8
}

解析一个类
接下来是官方的一个例子，用来打印出这个类

public class ClassPrinter extends ClassVisitor {
    public ClassPrinter() {
        super(ASM5);
    }

    @Override
    public void visit(int version, int access, String name,
                      String signature, String superName, String[] interfaces) {
        System.out.println(name + " extends " + superName + " {");
    }

    @Override
    public void visitSource(String source, String debug) {
    }

    @Override
    public void visitOuterClass(String owner, String name, String desc) {
    }

    @Override
    public AnnotationVisitor visitAnnotation(String desc,
                                             boolean visible) {
        return null;
    }
    @Override
    public void visitAttribute(Attribute attr) {
    }


    @Override
    public void visitInnerClass(String name, String outerName,
                                String innerName, int access) {
    }


    @Override
    public FieldVisitor visitField(int access, String name, String desc,
                                   String signature, Object value) {
        System.out.println(" " + desc + " " + name);
        return null;
    }

    @Override
    public MethodVisitor visitMethod(int access, String name,
                                     String desc, String signature, String[] exceptions) {
        System.out.println(" " + name + desc);
        return null;
    }

    @Override
    public void visitEnd() {
        System.out.println("}");
    }

    public static void main(String[] args) throws IOException {
        ClassPrinter cp = new ClassPrinter();
        ClassReader cr = new ClassReader("java.lang.Runnable");
        cr.accept(cp, 0);
    }
}

3.转化一个类

spring里面有一个类叫LocalVariableTableParameterNameDiscoverer，它可以用来获取参数名列表，其实内部就是使用了ASM，

	private static class ParameterNameDiscoveringVisitor extends ClassVisitor {

		private static final String STATIC_CLASS_INIT = "<clinit>";/*实例初始化方法（init）。类实例化（clinit）*/

		private final Class<?> clazz;

		private final Map<Member, String[]> memberMap;

		public ParameterNameDiscoveringVisitor(Class<?> clazz, Map<Member, String[]> memberMap) {
			super(SpringAsmInfo.ASM_VERSION);
			this.clazz = clazz;
			this.memberMap = memberMap;
		}

		@Override
		public MethodVisitor visitMethod(int access, String name, String desc, String signature, String[] exceptions) {
			// 排除 synthetic + bridged && 静态类实例化
			if (!isSyntheticOrBridged(access) && !STATIC_CLASS_INIT.equals(name)) {
//desc 指的是Method descriptors（方法描述符）
				return new LocalVariableTableVisitor(clazz, memberMap, name, desc, isStatic(access));
			}
			return null;
		}
 /* ACC_SYNTHETIC说明这个方法是由编译器生成，并且不会在源代码中出现。ACC_BRIDGE说明是编译生成的桥接方法*/
		private static boolean isSyntheticOrBridged(int access) {
			return (((access & Opcodes.ACC_SYNTHETIC) | (access & Opcodes.ACC_BRIDGE)) > 0);
		}

		private static boolean isStatic(int access) {
			return ((access & Opcodes.ACC_STATIC) > 0);
		}
	}

private static class LocalVariableTableVisitor extends MethodVisitor {

		private static final String CONSTRUCTOR = "<init>";

		private final Class<?> clazz;

		private final Map<Member, String[]> memberMap;

		private final String name;

		private final Type[] args;

		private final String[] parameterNames;

		private final boolean isStatic;

		private boolean hasLvtInfo = false;

		/*
		 * The nth entry contains the slot index of the LVT table entry holding the
		 * argument name for the nth parameter.
		 */
		private final int[] lvtSlotIndex;

		public LocalVariableTableVisitor(Class<?> clazz, Map<Member, String[]> map, String name, String desc, boolean isStatic) {
			super(SpringAsmInfo.ASM_VERSION);
			this.clazz = clazz;
			this.memberMap = map;
			this.name = name;
//方法描述符转化为参数
			this.args = Type.getArgumentTypes(desc);
			this.parameterNames = new String[this.args.length];
			this.isStatic = isStatic;
			this.lvtSlotIndex = computeLvtSlotIndices(isStatic, this.args);
		}

		@Override
		public void visitLocalVariable(String name, String description, String signature, Label start, Label end, int index) {
			this.hasLvtInfo = true;
			for (int i = 0; i < this.lvtSlotIndex.length; i++) {
				if (this.lvtSlotIndex[i] == index) {
					this.parameterNames[i] = name;
				}
			}
		}

		@Override
		public void visitEnd() {
			if (this.hasLvtInfo || (this.isStatic && this.parameterNames.length == 0)) {
				//存储下来
				this.memberMap.put(resolveMember(), this.parameterNames);
			}
		}

		private Member resolveMember() {
...
		}

		private static int[] computeLvtSlotIndices(boolean isStatic, Type[] paramTypes) {
			int[] lvtIndex = new int[paramTypes.length];
//实例方法，前面第0个位置还有个this
			int nextIndex = (isStatic ? 0 : 1);
			for (int i = 0; i < paramTypes.length; i++) {
				lvtIndex[i] = nextIndex;
//如果是long和double需要2个连续的局部变量表来保存
				if (isWideType(paramTypes[i])) {
					nextIndex += 2;
				}
				else {
					nextIndex++;
				}
			}
			return lvtIndex;
		}
	}

注：一个局部变量表的占用了32位的存储空间（一个存储单位称之为slot，槽），所以可以存储一个boolean、byte、char、short、float、int、refrence和returnAdress数据，long和double需要2个连续的局部变量表来保存，通过较小位置的索引来获取。如果被调用的是实例方法，那么第0个位置存储“this”关键字代表当前实例对象的引用。

里面重点关注visitLocalVariable方法，其他的忽略，是不是就觉得简单多了？

Javassist使用

Javassist和其他的类似库不同的是，Javassist并不要求开发者对字节码方面具有多么深入的了解，同样的，它也允许开发者忽略被修改的类本身的细节和结构。相对其他如ASM显得更为简单，当然性能也较之更低下。

ClassPool 跟踪和控制所操作的类，支持JVM 搜索路径中装载的、自定义路径、字节数组、流中装载二进制类、从头开始创建新类
CtClass 装载到类池中的类
CtField 字段
CtMethod 方法
CtConstructor 构造函数

ClassPool

ClassPool.getDefault() 只是搜索JVM的同路径下的class
new ClassPool(true) 当ClassPool没被引用的时候，JVM的垃圾收集会收集该类

如果系统使用多个类装载器，getDefault()只能搜索当前jvm的路径，可能加载不到对象

1
2
3

ClassPool#insertClassPath(ClassPath)
  ClassPool#appendClassPath(ClassPath)
  ClassPool#removeClassPath(ClassPath)

CtClass ct = mPool.get(name) 通过类池获取类 CtClass ct = mPool.makeClass(mClassName)创建一个类

CtClass

getName获取类名、getSimpleName获取简要类名、getSuperclass获取父类、getInterfaces获取接口、cc.getMethods获取方法
CtMethod m = cc.getDeclaredMethod("say") 获取方法
Class c = cc.toClass() 转化为Class byte[] b = cc.toBytecode() 转化为字节码二进制

还是那个例子，不过Javassist是不是简单多了

public class HelloWorld {
    public static void main(String[] args) throws IOException, CannotCompileException {
        byte[] bytes = builder();
        File file=new File("com"+File.separator+"souche"+File.separator+"rick", "HelloWorld.class");
        file.getParentFile().mkdirs();
        FileOutputStream fileOutputStream = new FileOutputStream(file);
        fileOutputStream.write(bytes);
        fileOutputStream.close();
    }

    public static byte[] builder() throws IOException, CannotCompileException {
        ClassPool classPool = ClassPool.getDefault();
        CtClass ctClass = classPool.makeClass("com.souche.rick.HelloWorld");
        CtMethod ctMethod =  CtNewMethod.make(
                "public static void main(String[] args) { System.out.println(\"Hello World!\"); }",
                ctClass);
        ctClass.addMethod(ctMethod);
        return ctClass.toBytecode();
    }
}

内省

参数	含义
$0, $1, $2, …	$0为this ，其他为参数
$args	数组参数Object[]
$$	所有参数m($$) 相当于m($1,$2,…) （除了this）
$cflow(…)	cflow 变量
$r	结果类型
$w	包装类型
$_	结果值
$sig	java.lang.Class数组对象，用来表示每个参数的类型
$type	java.lang.Class 对象表示结果类型
$class	A java.lang.Class 表示现在这个类的类型

例子：

class Point {
    int x, y;
    void move(int dx, int dy) { x += dx; y += dy; }
}
ClassPool pool = ClassPool.getDefault();
CtClass cc = pool.get("Point");
CtMethod m = cc.getDeclaredMethod("move");
m.insertBefore("{ System.out.println($1); System.out.println($2); }");
cc.writeFile();

结果

class Point {
    int x, y;
    void move(int dx, int dy) {
        { System.out.println(dx); System.out.println(dy); }
        x += dx; y += dy;
    }
}

dubbo的ReferenceBean(dubbo调用方)里面创建代理就运用了Javassist技术来获取接口生成代理类:

ttd
只看部分关键代码：

ClassGenerator ccp = null, ccm = null;
		try
		{
			ccp = ClassGenerator.newInstance(cl);/*通过传入ClassLoader来使用ClassPool获取封装的ClassGenerator（里面封装了Javassist操作的几个对象）*/

			Set<String> worked = new HashSet<String>();
			List<Method> methods = new ArrayList<Method>();

			for(int i=0;i<ics.length;i++)
			{
				if( !Modifier.isPublic(ics[i].getModifiers()) )
				{
					String npkg = ics[i].getPackage().getName();
					if( pkg == null )
					{
						pkg = npkg;
					}
					else
					{
						if( !pkg.equals(npkg)  )
							throw new IllegalArgumentException("non-public interfaces from different packages");
					}
				}
				ccp.addInterface(ics[i]);

				for( Method method : ics[i].getMethods() )
				{
					String desc = ReflectUtils.getDesc(method);
					if( worked.contains(desc) )
						continue;
					worked.add(desc);

					int ix = methods.size();
					Class<?> rt = method.getReturnType();
					Class<?>[] pts = method.getParameterTypes();

					StringBuilder code = new StringBuilder("Object[] args = new Object[").append(pts.length).append("];");
					for(int j=0;j<pts.length;j++)
						code.append(" args[").append(j).append("] = ($w)$").append(j+1).append(";");
					code.append(" Object ret = handler.invoke(this, methods[" + ix + "], args);");
					if( !Void.TYPE.equals(rt) )
						code.append(" return ").append(asArgument(rt, "ret")).append(";");

					methods.add(method);
					ccp.addMethod(method.getName(), method.getModifiers(), rt, pts, method.getExceptionTypes(), code.toString());
				}
			}

			if( pkg == null )
				pkg = PACKAGE_NAME;

			// create ProxyInstance class.
			String pcn = pkg + ".proxy" + id;/*类名为  pkg + ".proxy" +id = 包名 + “.poxy” +自增数值*/
			ccp.setClassName(pcn);
			ccp.addField("public static java.lang.reflect.Method[] methods;");
			ccp.addField("private " + InvocationHandler.class.getName() + " handler;");
			ccp.addConstructor(Modifier.PUBLIC, new Class<?>[]{ InvocationHandler.class }, new Class<?>[0], "handler=$1;");
            ccp.addDefaultConstructor();
			Class<?> clazz = ccp.toClass();
			clazz.getField("methods").set(null, methods.toArray(new Method[0]));

			// create Proxy class.
			String fcn = Proxy.class.getName() + id;/*对象名“Proxy”+id*/
			ccm = ClassGenerator.newInstance(cl);
			ccm.setClassName(fcn);
			ccm.addDefaultConstructor();/*设置默认构造方法*/
			ccm.setSuperClass(Proxy.class);/*继承于Proxy*/
/*添加方法public Object newInstance(InvocationHandler h){ return new " + 包名 + “.poxy” +自增数值 + "($1);}*/
			ccm.addMethod("public Object newInstance(" + InvocationHandler.class.getName() + " h){ return new " + pcn + "($1); }");
			Class<?> pc = ccm.toClass();
			proxy = (Proxy)pc.newInstance();

如果有一个接口,被dubbo:reference引用，假设包名是com.souche.rick

public interface AuthService {
    User register(UserRegisterParam userToAdd);
    String login(String username, String password);
}

1.委托类

public class com.souche.rick.proxy1 implements AuthService{
/*这里的methods通过反射依次被放入register、login*/
      public static java.lang.reflect.Method[] methods ;

      private InvocationHandler handler;

      public com.souche.rick.proxy1(InvocationHandler handler){
         this.handler = handler;
      }

      public User register(UserRegisterParam userToAdd){
         Object[] args = new Object[1];
         args[0] = (UserRegisterParam)userToAdd;
         Object ret = handler.invoke(this, methods[0], args);
         return (User)ret;
      }

      public  String login(String username, String password){
         Object[] args = new Object[2];
         args[0] = (String)username;
         args[0] = (String)password;
         Object ret = handler.invoke(this, methods[1], args);
         return (String)ret;
      }
}

生成一个继承于Proxy的代理子类，

public class Proxy + id extends Proxy{
	public Object newInstance(InvocationHandler h){ 
               return new  com.souche.rick.proxy1(h);
        }
}

此外还有非常多的工具使用了字节码操纵技术，比如fastjson(ASM)、hibernate(cglib、javaassist)、zorka(ASM)、Btrace(ASM)

参考：
https://www.cnblogs.com/qiumingcheng/p/5400265.html

https://www.ibm.com/developerworks/cn/java/j-lo-classloader/index.html

http://www.importnew.com/17770.html

Java跟踪利器

Posted on 2017-12-21 Edited on 2022-01-28 In 基础

Java跟踪利器—BTrace

地址：https://github.com/btraceio/btrace

BTrace 是基于动态字节码修改技术(Hotswap)来实现运行时 java 程序的跟踪和替换。大体的原理可以用下面的公式描述：Client(Java compile api + attach api) + Agent（脚本解析引擎 + ASM + JDK6 Instumentation） + Socket其实 BTrace 就是使用了 java attach api 附加 agent.jar ，然后使用脚本解析引擎+asm来重写指定类的字节码，再使用 instrument 实现对原有类的替换。

Btrace可以做什么？

接口性能变慢，分析每个方法的耗时情况；
当在Map中插入大量数据，分析其扩容情况；
分析哪个方法调用了System.gc()，调用栈如何；
执行某个方法抛出异常时，分析运行时参数；
….

BTrace重要概念与局限性

虽然BTrace很强大,但Btrace脚本就是一个普通的用@Btrace注解的Java类,其中包含一个或多个public static void修饰的方法,为了保证对目标程序不造成影响,Btrace脚本对其可以执行的动作做了很多限制

不能创建对象
不能抛出或者捕获异常
不能用synchronized关键字
不能对目标程序中的instace或者static变量
不能调用目标程序的instance或者static方法
脚本的field、method都必须是static的
脚本不能包括outer,inner,nested class
脚本中不能有循环,不能继承任何类,任何接口与assert语句

1.安装

https://github.com/btraceio/btrace/releases/tag/v1.3.10.1 下载到包解压

2.配置环境

export JAVA_HOME=/home/wenwei/jdk1.8.0_111
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib 
export PATH=${JAVA_HOME}/bin:$PATH
export BTRACE_HOME=/home/wenwei/btrace
export PATH=$PATH:$BTRACE_HOME/bin

3.预编译：

执行之前可以用预编译命令检查脚本的正确性，预编译命令为 btracec，它是一个 javac-like 命令

btracec JStack.java

4.使用BTrace

1.
btrace [-I <include-path>] [-p <port>] [-cp <classpath>] <pid> <btrace-script> [<args>]

-I:没有这个表明跳过预编译
include-path:指定用来编译脚本的头文件路径(关于预编译可参考例子ThreadBean.java)
port:btrace agent端口,默认是2020
classpath:编译所需类路径,一般是指btrace-client.jar等类所在路径
pid:java进程id
btrace-script:btrace脚本可以是.java文件,也可以是.class文件
args:传递给btrace脚本的参数, 在脚本中可以通过$(), $length()来获取这些参数(定义在BTraceUtils中)

例如：
btrace -cp lib/servlet-api.jar -p 2021 53523 JStack.java

2.
java -javaagent:btrace-agent.jar=[<agent-arg>[,<agent-arg>]*]? <launch-args>

参数：

noServer - don’t start the socket server
bootClassPath - boot classpath to be used
systemClassPath - system classpath to be used
debug - turns on verbose debug messages (true/false)
unsafe - do not check for btrace restrictions violations (true/false)
dumpClasses - dump the transformed bytecode to files (true/false)
dumpDir - specifies the folder where the transformed classes will be dumped to
stdout - redirect the btrace output to stdout instead of writing it to an arbitrary file (true/false)
probeDescPath - the path to search for probe descriptor XMLs
startupRetransform - enable retransform of all the loaded classes at attach (true/false)
scriptdir - the path to a directory containing scripts to be run at the agent startup
scriptOutputFile - the path to a file the btrace agent will store its output
script - colon separated list of tracing scripts to be run at the agent startup

5.脚本

package com.sun.btrace.samples;

import com.sun.btrace.annotations.*;
import static com.sun.btrace.BTraceUtils.Sys.*;
import static com.sun.btrace.BTraceUtils.Threads.*;

/*
 * A simple sample prints stack traces and exits. This
 * BTrace program mimics the jstack command line tool in JDK.
 */
@BTrace
public class JStack {
    static {
        deadlocks(false);
        jstackAll();
        exit(0);
    }
}

6.参数说明

@OnMethod:指定使用当前注解的方法应该在什么情况下触发,claszz属性指定要匹配的类的全限定类名,可以用正则表达式:/类名的Pattern/匹配,用”+类名”匹配所有子类,用”@某某注解”匹配用该注解注解过的类method属性指定要匹配的方法名称,可以用正则表达式:/方法名称的Pattern/匹配type属性:void(java.lang.String)可以用于匹配:public void funcName(String param) throws Exception,location属性用@Location来表明,匹配了clazz,method情况,在方法执行的何时去执行脚本(前,后,异常,行,某个方法调用)
@OnTimer:指定一个定时任务
@OnExit:当脚本运行Sys.exit(code)时触发
@OnError:当脚本运行抛出异常时触发
@OnEvent:脚本运行时Ctrl+C可以发送事件
@OnLowMemory:让你指定一个阀值,内存低于阀值触发
@OnProbe:可以用一个xml文件来描述你想在什么时候触发该方法

@Self:目标对象本身
@Retrun:目标程序方法返回值(Kind.RETURN)
@ProbeClassName:目标类名
@ProbeMethodName:目标方法名
@targetInstance:@Location指定的clazz,method的目标(Kind.CALL)
@targetMethodOrField:@Location指定的clazz,method的目标的方法或字段(Kind.CALL)
@Duration:目标方法执行时间,单位是纳秒,需要与 Kind.RETURN 或者 Kind.ERROR 一起使用

正则表达式定位

可以用表达式，批量定义需要监控的类与方法。正则表达式需要写在两个 “/“ 中间。

下例监控javax.swing下的所有类的所有方法….可能会非常慢，建议范围还是窄些。

@OnMethod(clazz="/javax\\.swing\\..*/", method="/.*/")
public static void swingMethods( @ProbeClassName String probeClass, @ProbeMethodName String probeMethod) {
   print("entered " + probeClass + "."  + probeMethod);
}

通过在拦截函数的定义里注入@ProbeClassName String probeClass, @ProbeMethodName String probeMethod 参数，告诉脚本实际匹配到的类和方法名。

另一个例子，监控Statement的executeUpdate(), executeQuery() 和 executeBatch() 三个方法，见JdbcQueries.java

按接口，父类，Annotation定位

比如我想匹配所有的Filter类，在接口或基类的名称前面，加个+ 就行
@OnMethod(clazz="+com.vip.demo.Filter", method="doFilter")

也可以按类或方法上的annotaiton匹配，前面加上@就行
@OnMethod(clazz="@javax.jws.WebService", method="@javax.jws.WebMethod")

其他

构造函数的名字是
@OnMethod(clazz="java.net.ServerSocket", method="<init>")
静态内部类的写法，是在类与内部类之间加上”$”
@OnMethod(clazz="com.vip.MyServer$MyInnerClass", method="hello")
如果有多个同名的函数，想区分开来，可以在拦截函数上定义不同的参数列表（见4.1）。

拦截时机

可以为同一个函数的不同的Location，分别定义多个拦截函数。

Kind.Entry与Kind.Return

@OnMethod( clazz="java.net.ServerSocket", method="bind" )
不写Location，默认就是刚进入函数的时候(Kind.ENTRY)。

但如果你想获得函数的返回结果或执行时间，则必须把切入点定在返回(Kind.RETURN)时。

1
2
3

OnMethod(clazz = "java.net.ServerSocket", method = "getLocalPort", location = @Location(Kind.RETURN))

public static void onGetPort(@Return int port, @Duration long duration)

duration的单位是纳秒，要除以 1,000,000 才是毫秒。

Kind.Error, Kind.Throw和 Kind.Catch

异常抛出(Throw)，异常被捕获(Catch)，异常没被捕获被抛出函数之外(Error)，主要用于对某些异常情况的跟踪。

在拦截函数的参数定义里注入一个Throwable的参数，代表异常。

@OnMethod(clazz = "java.net.ServerSocket", method = "bind", location = @Location(Kind.ERROR))

public static void onBind(Throwable exception, @Duration long duration)

Kind.Call与Kind.Line

下例定义监控bind()函数里调用的所有其他函数：

1
2
3

@OnMethod(clazz = "java.net.ServerSocket", method = "bind", location = @Location(value = Kind.CALL, clazz = "/.*/", method = "/.*/", where = Where.AFTER))

public static void onBind(@Self Object self, @TargetInstance Object instance, @TargetMethodOrField String method, @Duration long duration)

所调用的类及方法名所注入到@TargetInstance与 @TargetMethodOrField中。

静态函数中，instance的值为空。如果想获得执行时间，必须把Where定义成AFTER。
如果想获得执行时间，必须把Where定义成AFTER。

注意这里，一定不要像下面这样大范围的匹配，否则这性能是神仙也没法救了：

@OnMethod(clazz = "/javax\\.swing\\..*/", method = "/.*/", location = @Location(value = Kind.CALL, clazz = "/.*/", method = "/.*/"))
下例监控代码是否到达了Socket类的第363行。

@OnMethod(clazz = "java.net.ServerSocket", location = @Location(value = Kind.LINE, line = 363))
public static void onBind4() {
   println("socket bind reach line:363");
}

line还可以为-1，然后每行都会打印出来，加参数int line 获得的当前行数。此时会显示函数里完整的执行路径，但肯定又非常慢。

打印this，参数与返回值

定义注入

1
2
3

import com.sun.btrace.AnyType;
@OnMethod(clazz = "java.io.File", method = "createTempFile", location = @Location(value = Kind.RETURN))
public static void o(@Self Object self, String prefix, String suffix, @Return AnyType result)

如果想打印它们，首先按顺序定义用@Self 注释的this，完整的参数列表，以及用@Return 注释的返回值。

需要打印哪个就定义哪个，不需要的就不要定义。但定义一定要按顺序，比如参数列表不能跑到返回值的后面。

Self：
如果是静态函数， self为空。

前面提到，如果上述使用了非JDK的类，命令行里要指定classpath。不过，如前所述，因为BTrace里不允许调用类的方法，所以定义具体类很多时候也没意思，所以self定义为Object就够了。

参数：
参数数列表要么不要定义，要定义就要定义完整，否则BTrace无法处理不同参数的同名函数。

如果有些参数你实在不想引入非JDK类，又不会造成同名函数不可区分，可以用AnyType来定义（不能用Object）。

如果拦截点用正则表达式中匹配了多个函数，函数之间的参数个数不一样，你又还是想把参数打印出来时，可以用AnyType[] args来定义。

但不知道是不是当前版本的bug，AnyType[] args 不能和 location＝Kind.RETURN 同用，否则会进入一种奇怪的静默状态，只要有一个函数定义错了，整个Btrace就什么都打印不出来。

结果：
同理，结果也可以用AnyType来定义，特别是用正则表达式匹配多个函数的时候，连void都可以表示。

打印

再次强调，为了保证性能不受影响，Btrace不允许调用任何实例方法。
比如不能调用getter方法（怕在getter里有复杂的计算），只会通过直接反射来读取属性名。
又比如，除了JDK类，其他类toString时只会打印其类名＋System.IdentityHashCode。
println, printArray，都按上面的规律进行，所以只能打打基本类型。

如果想打印一个Object的属性，用printFields()来反射。

如果只想反射某个属性，参照下面打印Port属性的写法。从性能考虑，应把field用静态变量缓存起来。

注意JDK类与非JDK类的区别：

import java.lang.reflect.Field;

//JDK的类这样写就行
private static Field fdFiled = field("java.io,FileInputStream", "fd");

//非JDK的类，要给出ClassLoader，否则ClassNotFound
private static Field portField = field(classForName("com.vip.demo.MyObject", contextClassLoader()), "port");

public static void onChannelRead(@Self Object self) {
    println("port:" + getInt(portField, self));
}

TLS，拦截函数间的通信机制

如果要多个拦截函数之间要通信，可以使用@TLS定义 ThreadLocal的变量来共享

@TLS
private static int port = -1;

@OnMethod(clazz = "java.net.ServerSocket", method = "<init>")
public static void onServerSocket(int p){
    port = p;
}

@OnMethod(clazz = "java.net.ServerSocket", method = "bind")
public static void onBind(){
  println("server socket at " + port);
}

典型场景

打印慢调用

下例打印所有用时超过1毫秒的filter。

@OnMethod(clazz = "+com.vip.demo.Filter", method = "doFilter", location = @Location(Kind.RETURN))
public static void onDoFilter2(@ProbeClassName String pcn,  @Duration long duration) {
    if (duration > 1000000) {
        println(pcn + ",duration:" + (duration / 100000));
    }
}

最好能抽取了打印耗时的函数，减少代码重复度。
定位到某一个Filter慢了之后，可以直接用Location(Kind.CALL)，进一步找出它里面的哪一步慢了。

谁调用了这个函数

比如，谁调用了System.gc() ?

@OnMethod(clazz = "java.lang.System", method = "gc")
public static void onSystemGC() {
    println("entered System.gc()");
    jstack();
}

捕捉异常，或进入了某个特定代码行时，this对象及参数的值

按之前的提示，自己组合一下即可。

打印函数的调用/慢调用的统计信息

如果你已经看到了这里，那基本也不用我再啰嗦了，自己看Samples的Histogram.java, HistoOnEvent.java
可以用AtomicInteger构造计数器，然后定时(@OnTimer)，或根据事件(@OnEvent)输出结果(ctrl+c后选择发送事件)。

HashMap源码之红黑树(jdk1.8)

Posted on 2017-12-18 Edited on 2022-01-28 In 源码

TreeNode类似一颗红黑树
看下普通treeMap
https://juejin.im/post/5a0658f76fb9a04523415a8d

基本结构:

    static final class TreeNode<K,V> extends LinkedHashMap.Entry<K,V> {
//父节点
        TreeNode<K,V> parent;  // red-black tree links
//左右
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;//红还是黑
        TreeNode(int hash, K key, V val, Node<K,V> next) {
            super(hash, key, val, next);
        }

//返回根结点
        final TreeNode<K,V> root() {
            for (TreeNode<K,V> r = this, p;;) {
                if ((p = r.parent) == null)
                    return r;
                r = p;
            }
        }

static <K,V> void moveRootToFront(Node<K,V>[] tab, TreeNode<K,V> root) {
    int n;
    if (root != null && tab != null && (n = tab.length) > 0) {
        int index = (n - 1) & root.hash;
        TreeNode<K,V> first = (TreeNode<K,V>)tab[index];
        if (root != first) {
            Node<K,V> rn;
            tab[index] = root;
            TreeNode<K,V> rp = root.prev;
            if ((rn = root.next) != null)
                ((TreeNode<K,V>)rn).prev = rp;
            if (rp != null)
                rp.next = rn;
            if (first != null)
                first.prev = root;
            root.next = first;
            root.prev = null;
        }
        assert checkInvariants(root);
    }
}

final TreeNode<K,V> find(int h, Object k, Class<?> kc) {
    TreeNode<K,V> p = this;
    do {
        int ph, dir; K pk;
        TreeNode<K,V> pl = p.left, pr = p.right, q;
        if ((ph = p.hash) > h)
            p = pl;
        else if (ph < h)
            p = pr;
        else if ((pk = p.key) == k || (k != null && k.equals(pk)))
            return p;
        else if (pl == null)
            p = pr;
        else if (pr == null)
            p = pl;
        else if ((kc != null ||
                  (kc = comparableClassFor(k)) != null) &&
                 (dir = compareComparables(kc, k, pk)) != 0)
            p = (dir < 0) ? pl : pr;
        else if ((q = pr.find(h, k, kc)) != null)
            return q;
        else
            p = pl;
    } while (p != null);
    return null;
}

1
2
3

final TreeNode<K,V> getTreeNode(int h, Object k) {
    return ((parent != null) ? root() : this).find(h, k, null);
}

static int tieBreakOrder(Object a, Object b) {
    int d;
    if (a == null || b == null ||
        (d = a.getClass().getName().
         compareTo(b.getClass().getName())) == 0)
        d = (System.identityHashCode(a) <= System.identityHashCode(b) ?
             -1 : 1);
    return d;
}

hd.treeify(tab);

        final void treeify(Node<K,V>[] tab) {
            TreeNode<K,V> root = null;
            for (TreeNode<K,V> x = this, next; x != null; x = next) {
                next = (TreeNode<K,V>)x.next;
                x.left = x.right = null;
//如果是第一个结点则设置为根结点
                if (root == null) {
                    x.parent = null;
                    x.red = false;
                    root = x;
                }
//非第一个结点
                else {
                    K k = x.key;
                    int h = x.hash;
                    Class<?> kc = null;
                    for (TreeNode<K,V> p = root;;) {
                        int dir, ph;
                        K pk = p.key;
//根据hash值大小判断放在左子结点还是右
                        if ((ph = p.hash) > h)
                            dir = -1;
                        else if (ph < h)
                            dir = 1;
//hash值一样
                        else if ((kc == null &&
                                  (kc = comparableClassFor(k)) == null) ||
                                 (dir = compareComparables(kc, k, pk)) == 0)
                            dir = tieBreakOrder(k, pk);

                        TreeNode<K,V> xp = p;
                        if ((p = (dir <= 0) ? p.left : p.right) == null) {
                            x.parent = xp;
//设置
                            if (dir <= 0)
                                xp.left = x;
                            else
                                xp.right = x;
                            root = balanceInsertion(root, x);
                            break;
                        }
                    }
                }
            }
            moveRootToFront(tab, root);
        }

static <K,V> TreeNode<K,V> balanceInsertion(TreeNode<K,V> root,
                                            TreeNode<K,V> x) {
    x.red = true;
    for (TreeNode<K,V> xp, xpp, xppl, xppr;;) {
        if ((xp = x.parent) == null) {
            x.red = false;
            return x;
        }
        else if (!xp.red || (xpp = xp.parent) == null)
            return root;
        if (xp == (xppl = xpp.left)) {
            if ((xppr = xpp.right) != null && xppr.red) {
                xppr.red = false;
                xp.red = false;
                xpp.red = true;
                x = xpp;
            }
            else {
                if (x == xp.right) {
                    root = rotateLeft(root, x = xp);
                    xpp = (xp = x.parent) == null ? null : xp.parent;
                }
                if (xp != null) {
                    xp.red = false;
                    if (xpp != null) {
                        xpp.red = true;
                        root = rotateRight(root, xpp);
                    }
                }
            }
        }
        else {
            if (xppl != null && xppl.red) {
                xppl.red = false;
                xp.red = false;
                xpp.red = true;
                x = xpp;
            }
            else {
                if (x == xp.left) {
                    root = rotateRight(root, x = xp);
                    xpp = (xp = x.parent) == null ? null : xp.parent;
                }
                if (xp != null) {
                    xp.red = false;
                    if (xpp != null) {
                        xpp.red = true;
                        root = rotateLeft(root, xpp);
                    }
                }
            }
        }
    }
}

final Node<K,V> untreeify(HashMap<K,V> map) {
    Node<K,V> hd = null, tl = null;
    for (Node<K,V> q = this; q != null; q = q.next) {
        Node<K,V> p = map.replacementNode(q, null);
        if (tl == null)
            hd = p;
        else
            tl.next = p;
        tl = p;
    }
    return hd;
}

final TreeNode<K,V> putTreeVal(HashMap<K,V> map, Node<K,V>[] tab,
                               int h, K k, V v) {
    Class<?> kc = null;
    boolean searched = false;
    TreeNode<K,V> root = (parent != null) ? root() : this;
    for (TreeNode<K,V> p = root;;) {
        int dir, ph; K pk;
        if ((ph = p.hash) > h)
            dir = -1;
        else if (ph < h)
            dir = 1;
        else if ((pk = p.key) == k || (k != null && k.equals(pk)))
            return p;
        else if ((kc == null &&
                  (kc = comparableClassFor(k)) == null) ||
                 (dir = compareComparables(kc, k, pk)) == 0) {
            if (!searched) {
                TreeNode<K,V> q, ch;
                searched = true;
                if (((ch = p.left) != null &&
                     (q = ch.find(h, k, kc)) != null) ||
                    ((ch = p.right) != null &&
                     (q = ch.find(h, k, kc)) != null))
                    return q;
            }
            dir = tieBreakOrder(k, pk);
        }

        TreeNode<K,V> xp = p;
        if ((p = (dir <= 0) ? p.left : p.right) == null) {
            Node<K,V> xpn = xp.next;
            TreeNode<K,V> x = map.newTreeNode(h, k, v, xpn);
            if (dir <= 0)
                xp.left = x;
            else
                xp.right = x;
            xp.next = x;
            x.parent = x.prev = xp;
            if (xpn != null)
                ((TreeNode<K,V>)xpn).prev = x;
            moveRootToFront(tab, balanceInsertion(root, x));
            return null;
        }
    }
}

final void removeTreeNode(HashMap<K,V> map, Node<K,V>[] tab,
                          boolean movable) {
    int n;
    if (tab == null || (n = tab.length) == 0)
        return;
    int index = (n - 1) & hash;
    TreeNode<K,V> first = (TreeNode<K,V>)tab[index], root = first, rl;
    TreeNode<K,V> succ = (TreeNode<K,V>)next, pred = prev;
    if (pred == null)
        tab[index] = first = succ;
    else
        pred.next = succ;
    if (succ != null)
        succ.prev = pred;
    if (first == null)
        return;
    if (root.parent != null)
        root = root.root();
    if (root == null || root.right == null ||
        (rl = root.left) == null || rl.left == null) {
        tab[index] = first.untreeify(map);  // too small
        return;
    }
    TreeNode<K,V> p = this, pl = left, pr = right, replacement;
    if (pl != null && pr != null) {
        TreeNode<K,V> s = pr, sl;
        while ((sl = s.left) != null) // find successor
            s = sl;
        boolean c = s.red; s.red = p.red; p.red = c; // swap colors
        TreeNode<K,V> sr = s.right;
        TreeNode<K,V> pp = p.parent;
        if (s == pr) { // p was s's direct parent
            p.parent = s;
            s.right = p;
        }
        else {
            TreeNode<K,V> sp = s.parent;
            if ((p.parent = sp) != null) {
                if (s == sp.left)
                    sp.left = p;
                else
                    sp.right = p;
            }
            if ((s.right = pr) != null)
                pr.parent = s;
        }
        p.left = null;
        if ((p.right = sr) != null)
            sr.parent = p;
        if ((s.left = pl) != null)
            pl.parent = s;
        if ((s.parent = pp) == null)
            root = s;
        else if (p == pp.left)
            pp.left = s;
        else
            pp.right = s;
        if (sr != null)
            replacement = sr;
        else
            replacement = p;
    }
    else if (pl != null)
        replacement = pl;
    else if (pr != null)
        replacement = pr;
    else
        replacement = p;
    if (replacement != p) {
        TreeNode<K,V> pp = replacement.parent = p.parent;
        if (pp == null)
            root = replacement;
        else if (p == pp.left)
            pp.left = replacement;
        else
            pp.right = replacement;
        p.left = p.right = p.parent = null;
    }

    TreeNode<K,V> r = p.red ? root : balanceDeletion(root, replacement);

    if (replacement == p) {  // detach
        TreeNode<K,V> pp = p.parent;
        p.parent = null;
        if (pp != null) {
            if (p == pp.left)
                pp.left = null;
            else if (p == pp.right)
                pp.right = null;
        }
    }
    if (movable)
        moveRootToFront(tab, r);
}

当扩容的时候，节点使用split，((TreeNode<K,V>)e).split(this, newTab, j, oldCap);

final void split(HashMap<K,V> map, Node<K,V>[] tab, int index, int bit) {
    TreeNode<K,V> b = this;
    // Relink into lo and hi lists, preserving order
    TreeNode<K,V> loHead = null, loTail = null;
    TreeNode<K,V> hiHead = null, hiTail = null;
    int lc = 0, hc = 0;
    for (TreeNode<K,V> e = b, next; e != null; e = next) {
        next = (TreeNode<K,V>)e.next;
        e.next = null;
        if ((e.hash & bit) == 0) {
            if ((e.prev = loTail) == null)
                loHead = e;
            else
                loTail.next = e;
            loTail = e;
            ++lc;
        }
        else {
            if ((e.prev = hiTail) == null)
                hiHead = e;
            else
                hiTail.next = e;
            hiTail = e;
            ++hc;
        }
    }

    if (loHead != null) {
        if (lc <= UNTREEIFY_THRESHOLD)
            tab[index] = loHead.untreeify(map);
        else {
            tab[index] = loHead;
            if (hiHead != null) // (else is already treeified)
                loHead.treeify(tab);
        }
    }
    if (hiHead != null) {
        if (hc <= UNTREEIFY_THRESHOLD)
            tab[index + bit] = hiHead.untreeify(map);
        else {
            tab[index + bit] = hiHead;
            if (loHead != null)
                hiHead.treeify(tab);
        }
    }
}

static <K,V> TreeNode<K,V> rotateLeft(TreeNode<K,V> root,
                                      TreeNode<K,V> p) {
    TreeNode<K,V> r, pp, rl;
    if (p != null && (r = p.right) != null) {
        if ((rl = p.right = r.left) != null)
            rl.parent = p;
        if ((pp = r.parent = p.parent) == null)
            (root = r).red = false;
        else if (pp.left == p)
            pp.left = r;
        else
            pp.right = r;
        r.left = p;
        p.parent = r;
    }
    return root;
}

static <K,V> TreeNode<K,V> rotateRight(TreeNode<K,V> root,
                                       TreeNode<K,V> p) {
    TreeNode<K,V> l, pp, lr;
    if (p != null && (l = p.left) != null) {
        if ((lr = p.left = l.right) != null)
            lr.parent = p;
        if ((pp = l.parent = p.parent) == null)
            (root = l).red = false;
        else if (pp.right == p)
            pp.right = l;
        else
            pp.left = l;
        l.right = p;
        p.parent = l;
    }
    return root;
}


static <K,V> TreeNode<K,V> balanceDeletion(TreeNode<K,V> root,
                                           TreeNode<K,V> x) {
    for (TreeNode<K,V> xp, xpl, xpr;;)  {
        if (x == null || x == root)
            return root;
        else if ((xp = x.parent) == null) {
            x.red = false;
            return x;
        }
        else if (x.red) {
            x.red = false;
            return root;
        }
        else if ((xpl = xp.left) == x) {
            if ((xpr = xp.right) != null && xpr.red) {
                xpr.red = false;
                xp.red = true;
                root = rotateLeft(root, xp);
                xpr = (xp = x.parent) == null ? null : xp.right;
            }
            if (xpr == null)
                x = xp;
            else {
                TreeNode<K,V> sl = xpr.left, sr = xpr.right;
                if ((sr == null || !sr.red) &&
                    (sl == null || !sl.red)) {
                    xpr.red = true;
                    x = xp;
                }
                else {
                    if (sr == null || !sr.red) {
                        if (sl != null)
                            sl.red = false;
                        xpr.red = true;
                        root = rotateRight(root, xpr);
                        xpr = (xp = x.parent) == null ?
                            null : xp.right;
                    }
                    if (xpr != null) {
                        xpr.red = (xp == null) ? false : xp.red;
                        if ((sr = xpr.right) != null)
                            sr.red = false;
                    }
                    if (xp != null) {
                        xp.red = false;
                        root = rotateLeft(root, xp);
                    }
                    x = root;
                }
            }
        }
        else { // symmetric
            if (xpl != null && xpl.red) {
                xpl.red = false;
                xp.red = true;
                root = rotateRight(root, xp);
                xpl = (xp = x.parent) == null ? null : xp.left;
            }
            if (xpl == null)
                x = xp;
            else {
                TreeNode<K,V> sl = xpl.left, sr = xpl.right;
                if ((sl == null || !sl.red) &&
                    (sr == null || !sr.red)) {
                    xpl.red = true;
                    x = xp;
                }
                else {
                    if (sl == null || !sl.red) {
                        if (sr != null)
                            sr.red = false;
                        xpl.red = true;
                        root = rotateLeft(root, xpl);
                        xpl = (xp = x.parent) == null ?
                            null : xp.left;
                    }
                    if (xpl != null) {
                        xpl.red = (xp == null) ? false : xp.red;
                        if ((sl = xpl.left) != null)
                            sl.red = false;
                    }
                    if (xp != null) {
                        xp.red = false;
                        root = rotateRight(root, xp);
                    }
                    x = root;
                }
            }
        }
    }
}

     static <K,V> boolean checkInvariants(TreeNode<K,V> t) {
        TreeNode<K,V> tp = t.parent, tl = t.left, tr = t.right,
            tb = t.prev, tn = (TreeNode<K,V>)t.next;
        if (tb != null && tb.next != t)
            return false;
        if (tn != null && tn.prev != t)
            return false;
        if (tp != null && t != tp.left && t != tp.right)
            return false;
        if (tl != null && (tl.parent != t || tl.hash > t.hash))
            return false;
        if (tr != null && (tr.parent != t || tr.hash < t.hash))
            return false;
        if (t.red && tl != null && tl.red && tr != null && tr.red)
            return false;
        if (tl != null && !checkInvariants(tl))
            return false;
        if (tr != null && !checkInvariants(tr))
            return false;
        return true;
    }
}

treeifyBin

/**
     * Replaces all linked nodes in bin at index for given hash unless
     * table is too small, in which case resizes instead.
     */
    final void treeifyBin(Node<K,V>[] tab, int hash) {
        int n, index; Node<K,V> e;
 //如果当前哈希表为空，或者哈希表中元素的个数小于 进行树形化的阈值(默认为 64)，就去新建/扩容
        if (tab == null || (n = tab.length) < MIN_TREEIFY_CAPACITY)
            resize();
//(n - 1) & hash：确保索引在数组范围内
        else if ((e = tab[index = (n - 1) & hash]) != null) {
            TreeNode<K,V> hd = null, tl = null;
            do {
//    // For treeifyBin
//    TreeNode<K,V> replacementTreeNode(Node<K,V> p, Node<K,V> next) {
 //       return new TreeNode<>(p.hash, p.key, p.value, next);
 //   }
                TreeNode<K,V> p = replacementTreeNode(e, null);
                if (tl == null)
                    hd = p;
                else {
                    p.prev = tl;
                    tl.next = p;
                }
                tl = p;
            } while ((e = e.next) != null);
 //让桶的第一个元素指向新建的红黑树头结点，以后这个桶里的元素就是红黑树而不是链表了
            if ((tab[index] = hd) != null)
                hd.treeify(tab);
        }
    }

https://www.cnblogs.com/QH-Jimmy/p/7810644.html

RocketMQ 源码分析(一) ----入门使用、Namesrv分析

Posted on 2017-11-16 Edited on 2022-01-28 In 中间件 , 源码

入门使用：http://rocketmq.apache.org/docs/quick-start/

下载

unzip rocketmq-all-4.2.0-source-release.zip
cd rocketmq-all-4.2.0/
mvn -Prelease-all -DskipTests clean install -U
cd distribution/target/apache-rocketmq

Start Name Server

1 2	nohup sh bin/mqnamesrv & tail -f ~/logs/rocketmqlogs/namesrv.log

The Name Server boot success…

Start Broker

1 2	nohup sh bin/mqbroker -n localhost:9876 & tail -f ~/logs/rocketmqlogs/broker.log

The broker[%s, 172.30.30.233:10911] boot success…

发送消息可靠同步的

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
            DefaultMQProducer("please_rename_unique_group_name");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " +
                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

可靠的异步

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
        //Launch the instance.
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        for (int i = 0; i < 100; i++) {
                final int index = i;
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

消费者

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
        
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("TopicTest", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

单向传播

弱可靠

public class OnewayProducer {
    public static void main(String[] args) throws Exception{
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " +
                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            producer.sendOneway(msg);

        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

广播

给所有订阅者广播

public class BroadcastProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.start();

        for (int i = 0; i < 100; i++){
            Message msg = new Message("TopicTest",
                "TagA",
                "OrderID188",
                "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        producer.shutdown();
    }
}

public class BroadcastConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //set to broadcast mode
        consumer.setMessageModel(MessageModel.BROADCASTING);

        consumer.subscribe("TopicTest", "TagA || TagC || TagD");

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.printf("Broadcast Consumer Started.%n");
    }
}

定时消息

1.订阅等待消息

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
   
public class ScheduledMessageConsumer {
    public static void main(String[] args) throws Exception {
        // Instantiate message consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
        // Subscribe topics
        consumer.subscribe("TestTopic", "*");
        // Register message listener
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    // Print approximate delay time period
                    System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // Launch consumer
        consumer.start();
    }
}

2.发送

import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.common.message.Message;
    
 public class ScheduledMessageProducer {
    
     public static void main(String[] args) throws Exception {
         // Instantiate a producer to send scheduled messages
         DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
         // Launch producer
         producer.start();
         int totalMessagesToSend = 100;
         for (int i = 0; i < totalMessagesToSend; i++) {
             Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
             // This message will be delivered to consumer 10 seconds later.
             message.setDelayTimeLevel(3);
             // Send the message
             producer.send(message);
         }
         // Shutdown producer after use.
         producer.shutdown();
     } 
 }

批量消息

String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
try {
    producer.send(messages);
} catch (Exception e) {
    e.printStackTrace();
    //handle the error
}

！！因为最大只能发4M消息，最好每个不要超过1M

消息切分

public class ListSplitter implements Iterator<List<Message>> {
    private final int SIZE_LIMIT = 1000 * 1000;
    private final List<Message> messages;
    private int currIndex;
    public ListSplitter(List<Message> messages) {
            this.messages = messages;
    }
    @Override public boolean hasNext() {
        return currIndex < messages.size();
    }
    @Override public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; //for log overhead
            if (tmpSize > SIZE_LIMIT) {
                //it is unexpected that single message exceeds the SIZE_LIMIT
                //here just let it go, otherwise it will block the splitting process
                if (nextIndex - currIndex == 0) {
                   //if the next sublist has no element, add this one and then break, otherwise just break
                   nextIndex++;  
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
    
        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
}
//then you could split the large list into small ones:
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
   try {
       List<Message>  listItem = splitter.next();
       producer.send(listItem);
   } catch (Exception e) {
       e.printStackTrace();
       //handle the error
   }
}

过滤消息

1 2	DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE"); consumer.subscribe("TOPIC", "TAGA \|\| TAGB \|\| TAGC");

DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();

Message msg = new Message("TopicTest",
    tag,
    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// Set some properties.
msg.putUserProperty("a", String.valueOf(i));

SendResult sendResult = producer.send(msg);
   
producer.shutdown();

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");

// only subsribe messages have property a, also a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

事务

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        producer.setCheckThreadPoolMinSize(2);
        producer.setCheckThreadPoolMaxSize(2);
        producer.setCheckRequestHoldMax(2000);
        producer.setTransactionCheckListener(transactionCheckListener);
        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
        for (int i = 0; i < 100; i++) {
            try {
                Message msg =
                    new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
                System.out.printf("%s%n", sendResult);

                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

1
2
3

public interface TransactionCheckListener {
    LocalTransactionState checkLocalTransactionState(final MessageExt msg);
}

public class TransactionCheckListenerImpl implements TransactionCheckListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    @Override
    public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
        System.out.printf("server checking TrMsg %s%n", msg);

        int value = transactionIndex.getAndIncrement();
        if ((value % 6) == 0) {
            throw new RuntimeException("Could not find db");
        } else if ((value % 5) == 0) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else if ((value % 4) == 0) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }

        return LocalTransactionState.UNKNOW;
    }
}

log配置http://rocketmq.apache.org/docs/logappender-example/

OpenMessaging

http://rocketmq.apache.org/docs/openmessaging-example/

Message Reliablity

影响消息可靠性的几种情况：

Broker正常关闭
Broker异常Crash
OS Crash
机器掉电，但是能立即恢复供电情况。
机器无法开机（可能是cpu、主板、内存等关键设备损坏）
磁盘设备损坏。

(1)、(2)、(3)、(4)四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。

(5)、(6)属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。RocketMQ在这两种情况下，通过异步复制，可保证99%的消息不丢，但是仍然会有极少量的消息可能丢失。通过同步双写技术可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场合，例如与Money相关的应用。

RocketMQ从3.0版本开始支持同步双写。

架构

NameServer Cluster

命名服务器提供了轻量级服务发现和路由。每台命名服务器记录了完整的路由信息。提供了一致性读写服务，支持快速存储扩展。

Broker Cluster

Brokers专注于消息存储，提供轻量级的TOPIC和QUEUE机制。支持“推”和“拉”模式,包含容错机制(2或3份副本)，并提供强大的峰值填充和按原始时间顺序累积数千亿条消息的能力。此外，Brokers还提供灾难恢复，丰富的指标统计和警报机制。

源码分析

代码分层

脚本源码分析

mqnamesrv.sh

#!/bin/sh
if [ -z "$ROCKETMQ_HOME" ] ; then
  ## resolve links - $0 may be a link to maven's home
  PRG="$0"

  # need this for relative symlinks
  while [ -h "$PRG" ] ; do
    ls=`ls -ld "$PRG"`
    link=`expr "$ls" : '.*-> \(.*\)$'`
    if expr "$link" : '/.*' > /dev/null; then
      PRG="$link"
    else
      PRG="`dirname "$PRG"`/$link"
    fi
  done

  saveddir=`pwd`

  ROCKETMQ_HOME=`dirname "$PRG"`/..

  # make it fully qualified
  ROCKETMQ_HOME=`cd "$ROCKETMQ_HOME" && pwd`

  cd "$saveddir"
fi

export ROCKETMQ_HOME

sh ${ROCKETMQ_HOME}/bin/runserver.sh org.apache.rocketmq.namesrv.NamesrvStartup $@

runserver.sh

#!/bin/sh
error_exit ()
{
    echo "ERROR: $1 !!"
    exit 1
}

[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)!"

export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}

#===========================================================================================
# JVM Configuration
#===========================================================================================
JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:PermSize=128m -XX:MaxPermSize=320m"
JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:+DisableExplicitGC -XX:-UseParNewGC"
JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:/dev/shm/rmq_srv_gc.log -XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT}  -XX:-UseLargePages"
JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${BASE_DIR}/lib"
#JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"

$JAVA ${JAVA_OPT} $@

脚本最后就是通过 java org.apache.rocketmq.namesrv.NamesrvStartup 运行nameserver

$@指的脚本传入的参数

NamesrvStartup

public class NamesrvStartup {
    public static Properties properties = null;
    public static CommandLine commandLine = null;

    public static void main(String[] args) {
        main0(args);
    }

    public static NamesrvController main0(String[] args) {
//设置rocketmq.remoting.version （版本号）
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
//com.rocketmq.remoting.socket.sndbuf.size  
        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) {
            NettySystemConfig.socketSndbufSize = 4096;
        }
//com.rocketmq.remoting.socket.rcvbuf.size
        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) {
            NettySystemConfig.socketRcvbufSize = 4096;
        }

        try {
     //命令行Apache Commons CLI 见 https://my.oschina.net/cloudcoder/blog/363793
            Options options = ServerUtil.buildCommandlineOptions(new Options());
            commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
            if (null == commandLine) {
                System.exit(-1);
                return null;
            }
//
            final NamesrvConfig namesrvConfig = new NamesrvConfig();
            final NettyServerConfig nettyServerConfig = new NettyServerConfig();
            nettyServerConfig.setListenPort(9876);
            if (commandLine.hasOption('c')) {
                String file = commandLine.getOptionValue('c');
                if (file != null) {
                    InputStream in = new BufferedInputStream(new FileInputStream(file));
                    properties = new Properties();
                    properties.load(in);
                    MixAll.properties2Object(properties, namesrvConfig);
                    MixAll.properties2Object(properties, nettyServerConfig);
                    namesrvConfig.setConfigStorePath(file);
                    System.out.printf("load config properties file OK, " + file + "%n");
                    in.close();
                }
            }

            if (commandLine.hasOption('p')) {
                MixAll.printObjectProperties(null, namesrvConfig);
                MixAll.printObjectProperties(null, nettyServerConfig);
                System.exit(0);
            }

            MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

            if (null == namesrvConfig.getRocketmqHome()) {
                System.out.printf("Please set the " + MixAll.ROCKETMQ_HOME_ENV + " variable in your environment to match the location of the RocketMQ installation%n");
                System.exit(-2);
            }
//logback配置 见http://www.iteye.com/topic/345924
            LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
            JoranConfigurator configurator = new JoranConfigurator();
            configurator.setContext(lc);
            lc.reset();
            configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
            final Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
//打印配置
            MixAll.printObjectProperties(log, namesrvConfig);
            MixAll.printObjectProperties(log, nettyServerConfig);
//构造NamesrvController
            final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

            // remember all configs to prevent discard
            controller.getConfiguration().registerConfig(properties);
//初始化
            boolean initResult = controller.initialize();
            if (!initResult) {
                controller.shutdown();
                System.exit(-3);
            }
//优雅退出
            Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    controller.shutdown();
                    return null;
                }
            }));
//启动
            controller.start();

            String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
            log.info(tip);
            System.out.printf(tip + "%n");

            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }

        return null;
    }
}

NamesrvController

namesrvConfig：nameServer的配置
nettyServerConfig：NameServer的netty配置
remotingServer：NameServer 的netty服务器
scheduledExecutorService：routeInfoManager和kvConfigManager使用的定时线程池
remotingExecutor：netty使用的线程池
brokerHosekeppingService：
kvConfigManager：kv配置管理
routeInfoManager：包含broker的ip和对应的队列信息，说明producer可以往哪一个broker发送消息，consumer从哪一个broker pull消息

public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
    this.namesrvConfig = namesrvConfig;
    this.nettyServerConfig = nettyServerConfig;
    this.kvConfigManager = new KVConfigManager(this);
    this.routeInfoManager = new RouteInfoManager();
    this.brokerHousekeepingService = new BrokerHousekeepingService(this);
    this.configuration = new Configuration(
        log,
        this.namesrvConfig, this.nettyServerConfig
    );
    this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
}

initialize

 public boolean initialize() {
     //从"user.home"+/namesrv/kvConfig.json   获取json配置解析到内存中->configTable
        this.kvConfigManager.load();
//构造NettyRemotingServer
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
//默认8
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
//注册requestProcessor
        this.registerProcessor();
//“NSScheduledThread” 每10秒扫描下，如果broker超过2分钟，关闭channel，打印The broker channel expired,
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);

//每10分钟打印下kvConfigManager的配置
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);

        return true;
    }


    private void registerProcessor() {
//集群
        if (namesrvConfig.isClusterTest()) {
            this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
                this.remotingExecutor);
        } else {
            this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
        }
    }

start && shutdown

public void start() throws Exception {
    this.remotingServer.start();
}

public void shutdown() {
    this.remotingServer.shutdown();
    this.remotingExecutor.shutdown();
    this.scheduledExecutorService.shutdown();
}

NettyRemotingServer

public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
        final ChannelEventListener channelEventListener) {
        super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
        this.serverBootstrap = new ServerBootstrap();
        this.nettyServerConfig = nettyServerConfig;
        this.channelEventListener = channelEventListener;

        int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
        if (publicThreadNums <= 0) {
            publicThreadNums = 4;
        }

        this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyServerPublicExecutor_" + this.threadIndex.incrementAndGet());
            }
        });
//boss线程组
        this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyBoss_%d", this.threadIndex.incrementAndGet()));
            }
        });

    //1、是否是linux平台2、useEpollNativeSelector配置是否为true3、epoll是否可用
        if (useEpoll()) {
            //serverSelectorThreads 默认为3
            this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                private int threadTotal = nettyServerConfig.getServerSelectorThreads();

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
                }
            });
        } else {
            this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                private int threadTotal = nettyServerConfig.getServerSelectorThreads();

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyServerNIOSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
                }
            });
        }
    }

start

@Override
public void start() {
    this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
        //默认8
        nettyServerConfig.getServerWorkerThreads(),
        new ThreadFactory() {

            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
            }
        });

    ServerBootstrap childHandler =
        this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
            .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .option(ChannelOption.SO_REUSEADDR, true)
        //
            .option(ChannelOption.SO_KEEPALIVE, false)
        //禁用了Nagle算法，允许小包的发送
            .childOption(ChannelOption.TCP_NODELAY, true)
        //发送缓冲区大小，默认65535
            .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
        //接收缓冲区大小，默认65535
            .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
        //绑定端口，默认8888
            .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline()
                        .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
                            new HandshakeHandler(TlsSystemConfig.tlsMode))
                        .addLast(defaultEventExecutorGroup,
                            new NettyEncoder(),
                            new NettyDecoder(),
                            new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                            new NettyConnectManageHandler(),
                            new NettyServerHandler()
                        );
                }
            });

    if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
        childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    }

    try {
        ChannelFuture sync = this.serverBootstrap.bind().sync();
        InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
        this.port = addr.getPort();
    } catch (InterruptedException e1) {
        throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
    }

    if (this.channelEventListener != null) {
        this.nettyEventExecutor.start();
    }

    this.timer.scheduleAtFixedRate(new TimerTask() {

        @Override
        public void run() {
            try {
                NettyRemotingServer.this.scanResponseTable();
            } catch (Throwable e) {
                log.error("scanResponseTable exception", e);
            }
        }
    }, 1000 * 3, 1000);
}

zookeeper

Posted on 2017-11-11 Edited on 2022-01-28 In 一致性

CAP

CAP由Eric Brewer在2000年PODC会议上提出[1][2]，是Eric Brewer在Inktomi[3]期间研发搜索引擎、分布式web缓存时得出的关于数据一致性(consistency)、服务可用性(availability)、分区容错性(partition-tolerance)的猜想：

数据一致性(consistency)：如果系统对一个写操作返回成功，那么之后的读请求都必须读到这个新数据；如果返回失败，那么所有读操作都不能读到这个数据，对调用者而言数据具有强一致性(strong consistency) (又叫原子性 atomic、线性一致性 linearizable consistency)[5]
服务可用性(availability)：所有读写请求在一定时间内得到响应，可终止、不会一直等待
分区容错性(partition-tolerance)：在网络分区的情况下，被分隔的节点仍能正常对外服务

在某时刻如果满足AP，分隔的节点同时对外服务但不能相互通信，将导致状态不一致，即不能满足C；如果满足CP，网络分区的情况下为达成C，请求只能一直等待，即不满足A；如果要满足CA，在一定时间内要达到节点状态一致，要求不能出现网络分区，则不能满足P。

C、A、P三者最多只能满足其中两个，和FLP定理一样，CAP定理也指示了一个不可达的结果(impossibility result)。

在满足分区容错的前提下，没有算法能同时满足数据一致性和服务可用性[11]：

P 是必选项，那3选2的选择题不就变成数据一致性(consistency)、服务可用性(availability) 2选1？工程实践中一致性有不同程度，可用性也有不同等级，在保证分区容错性的前提下，放宽约束后可以兼顾一致性和可用性，两者不是非此即彼[12]。

从源码了解spring MVC

Posted on 2017-11-01 Edited on 2022-01-28 In 源码

一般要让springmvc生效，我们都在web.xml配上这么一段

<servlet>
      <servlet-name>dispatcher</servlet-name>
      <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
      <load-on-startup>1</load-on-startup>
  </servlet>

  <servlet-mapping>
      <servlet-name>dispatcher</servlet-name>
      <url-pattern>/</url-pattern>
  </servlet-mapping>

记得以前很早的时候刚开始学java web的时候都是直接继承HttpServlet然后写的，这个当然也不例外，

sequencediagram1

Servlet接口包括 public void init(ServletConfig config) throws ServletException;、 public void service(ServletRequest req, ServletResponse res) throws ServletException, IOException;、public void destroy();

在DispatcherServlet父类FrameworkServlet

@Override
protected void service(HttpServletRequest request, HttpServletResponse response)
		throws ServletException, IOException {

	HttpMethod httpMethod = HttpMethod.resolve(request.getMethod());
	if (HttpMethod.PATCH == httpMethod || httpMethod == null) {//因为HttpServlet没有实现对PATCH的支持
		processRequest(request, response);
	}
	else {
		super.service(request, response);
	}
}

最后根据请求调用的又是FrameworkServlet的

@Override
protected final void doGet(HttpServletRequest request, HttpServletResponse response)
		throws ServletException, IOException {
	processRequest(request, response);
}

接下来

protected final void processRequest(HttpServletRequest request, HttpServletResponse response)
			throws ServletException, IOException {

		long startTime = System.currentTimeMillis();
		Throwable failureCause = null;

		LocaleContext previousLocaleContext = LocaleContextHolder.getLocaleContext();
		LocaleContext localeContext = buildLocaleContext(request);

		RequestAttributes previousAttributes = RequestContextHolder.getRequestAttributes();
		ServletRequestAttributes requestAttributes = buildRequestAttributes(request, response, previousAttributes);

		WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
		asyncManager.registerCallableInterceptor(FrameworkServlet.class.getName(), new RequestBindingInterceptor());

		initContextHolders(request, localeContext, requestAttributes);

		try {
			doService(request, response);
		}
		catch (ServletException ex) {
			failureCause = ex;
			throw ex;
		}
		catch (IOException ex) {
			failureCause = ex;
			throw ex;
		}
		catch (Throwable ex) {
			failureCause = ex;
			throw new NestedServletException("Request processing failed", ex);
		}

		finally {
			resetContextHolders(request, previousLocaleContext, previousAttributes);
			if (requestAttributes != null) {
				requestAttributes.requestCompleted();
			}

			if (logger.isDebugEnabled()) {
				if (failureCause != null) {
					this.logger.debug("Could not complete request", failureCause);
				}
				else {
					if (asyncManager.isConcurrentHandlingStarted()) {
						logger.debug("Leaving response open for concurrent processing");
					}
					else {
						this.logger.debug("Successfully completed request");
					}
				}
			}

			publishRequestHandledEvent(request, response, startTime, failureCause);
		}
	}

DispatcherServlet

@Override
	protected void doService(HttpServletRequest request, HttpServletResponse response) throws Exception {
		if (logger.isDebugEnabled()) {
			String resumed = WebAsyncUtils.getAsyncManager(request).hasConcurrentResult() ? " resumed" : "";
			logger.debug("DispatcherServlet with name '" + getServletName() + "'" + resumed +
					" processing " + request.getMethod() + " request for [" + getRequestUri(request) + "]");
		}

		// Keep a snapshot of the request attributes in case of an include,
		// to be able to restore the original attributes after the include.
		Map<String, Object> attributesSnapshot = null;
		if (WebUtils.isIncludeRequest(request)) {//javax.servlet.include.request_uri 
			attributesSnapshot = new HashMap<String, Object>();
			Enumeration<?> attrNames = request.getAttributeNames();
			while (attrNames.hasMoreElements()) {
				String attrName = (String) attrNames.nextElement();
				if (this.cleanupAfterInclude || attrName.startsWith("org.springframework.web.servlet")) {
					attributesSnapshot.put(attrName, request.getAttribute(attrName));
				}
			}
		}

		// Make framework objects available to handlers and view objects.
		request.setAttribute(WEB_APPLICATION_CONTEXT_ATTRIBUTE, getWebApplicationContext());
		request.setAttribute(LOCALE_RESOLVER_ATTRIBUTE, this.localeResolver);
		request.setAttribute(THEME_RESOLVER_ATTRIBUTE, this.themeResolver);
		request.setAttribute(THEME_SOURCE_ATTRIBUTE, getThemeSource());
//转发 request是不变的，重定向会生成新的request，那传递参数就不能直接用request进行传递。通过FlashMap存储一个请求的输出，当进入另一个请求时作为该请求的输入，典型场景如重定向（POST-REDIRECT-GET模式，1、POST时将下一次需要的数据放在FlashMap；2、重定向；3、通过GET访问重定向的地址，此时FlashMap会把1放到FlashMap的数据取出放到请求中，并从FlashMap中删除；从而支持在两次请求之间保存数据并防止了重复表单提交）。
		FlashMap inputFlashMap = this.flashMapManager.retrieveAndUpdate(request, response);
		if (inputFlashMap != null) {
			request.setAttribute(INPUT_FLASH_MAP_ATTRIBUTE, Collections.unmodifiableMap(inputFlashMap));
		}
		request.setAttribute(OUTPUT_FLASH_MAP_ATTRIBUTE, new FlashMap());
		request.setAttribute(FLASH_MAP_MANAGER_ATTRIBUTE, this.flashMapManager);

		try {
			doDispatch(request, response);
		}
		finally {
			if (!WebAsyncUtils.getAsyncManager(request).isConcurrentHandlingStarted()) {
				// Restore the original attribute snapshot, in case of an include.
				if (attributesSnapshot != null) {
					restoreAttributesAfterInclude(request, attributesSnapshot);
				}
			}
		}
	}

doDispatch

	protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
		HttpServletRequest processedRequest = request;
		HandlerExecutionChain mappedHandler = null;
		boolean multipartRequestParsed = false;

		WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);

		try {
			ModelAndView mv = null;
			Exception dispatchException = null;

			try {
				processedRequest = checkMultipart(request);//文件上传,返回MultipartHttpServletRequest对象，如果不是则返回原始对象
				multipartRequestParsed = (processedRequest != request);

				// Determine handler for the current request.
				mappedHandler = getHandler(processedRequest);//HandlerMapping找到对应handler
//BeanNameUrlHandlerMapping ：通过对比url和bean的name找到对应的对象 
//SimpleUrlHandlerMapping ：也是直接配置url和对应bean,比BeanNameUrlHandlerMapping功能更多 
//DefaultAnnotationHandlerMapping : 主要是针对注解配置@RequestMapping的，已过时 
//RequestMappingHandlerMapping ：取代了上面一个 
				if (mappedHandler == null || mappedHandler.getHandler() == null) {
					noHandlerFound(processedRequest, response);
					return;
				}

				// Determine handler adapter for the current request.
//HttpRequestHandlerAdapter ： 要求handler实现HttpRequestHandler接口，该接口的方法为                                                             void handleRequest(HttpServletRequest request, HttpServletResponse response)也就是  handler必须有一个handleRequest方法 
//SimpleControllerHandlerAdapter：要求handler实现Controller接口，该接口的方法为ModelAndView handleRequest(HttpServletRequest request, HttpServletResponse response)，也就是本工程采用的 
//AnnotationMethodHandlerAdapter ：和上面的DefaultAnnotationHandlerMapping配对使用的，也已过时 
//RequestMappingHandlerAdapter : 和上面的RequestMappingHandlerMapping配对使用，针对@RequestMapping 
				HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());//由handler 找到对应的HandlerAdapter

				// Process last-modified header, if supported by the handler.
				String method = request.getMethod();
				boolean isGet = "GET".equals(method);
				if (isGet || "HEAD".equals(method)) {
					long lastModified = ha.getLastModified(request, mappedHandler.getHandler());
					if (logger.isDebugEnabled()) {
						logger.debug("Last-Modified value for [" + getRequestUri(request) + "] is: " + lastModified);
					}
					if (new ServletWebRequest(request, response).checkNotModified(lastModified) && isGet) {
						return;
					}
				}

                               //HandlerInterceptor的preHandle
				if (!mappedHandler.applyPreHandle(processedRequest, response)) {
					return;
				}

				// 执行 handler.
				mv = ha.handle(processedRequest, response, mappedHandler.getHandler());

				if (asyncManager.isConcurrentHandlingStarted()) {
					return;
				}

                               //设置一个默认的视图名，用在某些直接返回model的
				applyDefaultViewName(processedRequest, mv);

                                //HandlerInterceptor postHandle
				mappedHandler.applyPostHandle(processedRequest, response, mv);
			}
			catch (Exception ex) {
				dispatchException = ex;
			}
			catch (Throwable err) {
				// As of 4.3, we're processing Errors thrown from handler methods as well,
				// making them available for @ExceptionHandler methods and other scenarios.
				dispatchException = new NestedServletException("Handler dispatch failed", err);
			}
			processDispatchResult(processedRequest, response, mappedHandler, mv, dispatchException);
		}
		catch (Exception ex) {
			triggerAfterCompletion(processedRequest, response, mappedHandler, ex);
		}
		catch (Throwable err) {
			triggerAfterCompletion(processedRequest, response, mappedHandler,
					new NestedServletException("Handler processing failed", err));
		}
		finally {
			if (asyncManager.isConcurrentHandlingStarted()) {
				// Instead of postHandle and afterCompletion
				if (mappedHandler != null) {
					mappedHandler.applyAfterConcurrentHandlingStarted(processedRequest, response);
				}
			}
			else {
				// Clean up any resources used by a multipart request.
				if (multipartRequestParsed) {
					cleanupMultipart(processedRequest);
				}
			}
		}
	}

getHandler

protected HandlerExecutionChain getHandler(HttpServletRequest request) throws Exception {
	for (HandlerMapping hm : this.handlerMappings) {
		if (logger.isTraceEnabled()) {
			logger.trace(
					"Testing handler map [" + hm + "] in DispatcherServlet with name '" + getServletName() + "'");
		}
		HandlerExecutionChain handler = hm.getHandler(request);
		if (handler != null) {
			return handler;
		}
	}
	return null;
}

getHandlerAdapter

	protected HandlerAdapter getHandlerAdapter(Object handler) throws ServletException {
		for (HandlerAdapter ha : this.handlerAdapters) {
			if (logger.isTraceEnabled()) {
				logger.trace("Testing handler adapter [" + ha + "]");
			}
//如果适配，就返回HandlerAdapter
			if (ha.supports(handler)) {
				return ha;
			}
		}
		throw new ServletException("No adapter for handler [" + handler +
				"]: The DispatcherServlet configuration needs to include a HandlerAdapter that supports this handler");
	}

processDispatchResult

private void processDispatchResult(HttpServletRequest request, HttpServletResponse response,
			HandlerExecutionChain mappedHandler, ModelAndView mv, Exception exception) throws Exception {

		boolean errorView = false;

		if (exception != null) {
			if (exception instanceof ModelAndViewDefiningException) {
				logger.debug("ModelAndViewDefiningException encountered", exception);
				mv = ((ModelAndViewDefiningException) exception).getModelAndView();
			}
			else {
				Object handler = (mappedHandler != null ? mappedHandler.getHandler() : null);
				mv = processHandlerException(request, response, handler, exception);
				errorView = (mv != null);
			}
		}

		// Did the handler return a view to render?
		if (mv != null && !mv.wasCleared()) {
			render(mv, request, response);
			if (errorView) {
				WebUtils.clearErrorRequestAttributes(request);
			}
		}
		else {
			if (logger.isDebugEnabled()) {
				logger.debug("Null ModelAndView returned to DispatcherServlet with name '" + getServletName() +
						"': assuming HandlerAdapter completed request handling");
			}
		}

		if (WebAsyncUtils.getAsyncManager(request).isConcurrentHandlingStarted()) {
			// Concurrent handling started during a forward
			return;
		}

		if (mappedHandler != null) {
//HandlerInterceptor的afterCompletion
			mappedHandler.triggerAfterCompletion(request, response, null);
		}
	}



protected void render(ModelAndView mv, HttpServletRequest request, HttpServletResponse response) throws Exception {
		// Determine locale for request and apply it to the response.
		Locale locale = this.localeResolver.resolveLocale(request);
		response.setLocale(locale);

		View view;
		if (mv.isReference()) {
			// 解析视图名
			view = resolveViewName(mv.getViewName(), mv.getModelInternal(), locale, request);
			if (view == null) {
				throw new ServletException("Could not resolve view with name '" + mv.getViewName() +
						"' in servlet with name '" + getServletName() + "'");
			}
		}
		else {
			// No need to lookup: the ModelAndView object contains the actual View object.
			view = mv.getView();
			if (view == null) {
				throw new ServletException("ModelAndView [" + mv + "] neither contains a view name nor a " +
						"View object in servlet with name '" + getServletName() + "'");
			}
		}

		// Delegate to the View object for rendering.
		if (logger.isDebugEnabled()) {
			logger.debug("Rendering view [" + view + "] in DispatcherServlet with name '" + getServletName() + "'");
		}
		try {
			if (mv.getStatus() != null) {
				response.setStatus(mv.getStatus().value());
			}
//这里决定究竟是转发还是重定向,或者说变成其他视图
			view.render(mv.getModelInternal(), request, response);
		}
		catch (Exception ex) {
			if (logger.isDebugEnabled()) {
				logger.debug("Error rendering view [" + view + "] in DispatcherServlet with name '" +
						getServletName() + "'", ex);
			}
			throw ex;
		}
	}


	@Override
	public void render(Map<String, ?> model, HttpServletRequest request, HttpServletResponse response) throws Exception {
		if (logger.isTraceEnabled()) {
			logger.trace("Rendering view with name '" + this.beanName + "' with model " + model +
				" and static attributes " + this.staticAttributes);
		}

		Map<String, Object> mergedModel = createMergedOutputModel(model, request, response);
		prepareResponse(request, response);
		renderMergedOutputModel(mergedModel, getRequestToExpose(request), response);
	}


@Override
    protected void renderMergedOutputModel(Map<String, Object> model, //
                                           HttpServletRequest request, //
                                           HttpServletResponse response) throws Exception {
        Object value = filterModel(model);

        ByteArrayOutputStream outnew = new ByteArrayOutputStream();

        int len = JSON.writeJSONString(outnew, //
                fastJsonConfig.getCharset(), //
                value, //
                fastJsonConfig.getSerializeConfig(), //
                fastJsonConfig.getSerializeFilters(), //
                fastJsonConfig.getDateFormat(), //
                JSON.DEFAULT_GENERATE_FEATURE, //
                fastJsonConfig.getSerializerFeatures());

        if (this.updateContentLength) {
            // Write content length (determined via byte array).
            response.setContentLength(len);
        }

        // Flush byte array to servlet output stream.
        ServletOutputStream out = response.getOutputStream();
        outnew.writeTo(out);
        outnew.close();
        out.flush();
    }

HandlerExecutionChain

1请求对应的控制器
2拦截器

boolean applyPreHandle(HttpServletRequest request, HttpServletResponse response) throws Exception {
	HandlerInterceptor[] interceptors = getInterceptors();
	if (!ObjectUtils.isEmpty(interceptors)) {
		for (int i = 0; i < interceptors.length; i++) {
			HandlerInterceptor interceptor = interceptors[i];
			if (!interceptor.preHandle(request, response, this.handler)) {
				triggerAfterCompletion(request, response, null);
				return false;
			}
			this.interceptorIndex = i;
		}
	}
	return true;
}


void applyPostHandle(HttpServletRequest request, HttpServletResponse response, ModelAndView mv) throws Exception {
	HandlerInterceptor[] interceptors = getInterceptors();
	if (!ObjectUtils.isEmpty(interceptors)) {
		for (int i = interceptors.length - 1; i >= 0; i--) {
			HandlerInterceptor interceptor = interceptors[i];
			interceptor.postHandle(request, response, this.handler, mv);
		}
	}
}


void triggerAfterCompletion(HttpServletRequest request, HttpServletResponse response, Exception ex)
		throws Exception {

	HandlerInterceptor[] interceptors = getInterceptors();
	if (!ObjectUtils.isEmpty(interceptors)) {
		for (int i = this.interceptorIndex; i >= 0; i--) {
			HandlerInterceptor interceptor = interceptors[i];
			try {
				interceptor.afterCompletion(request, response, this.handler, ex);
			}
			catch (Throwable ex2) {
				logger.error("HandlerInterceptor.afterCompletion threw exception", ex2);
			}
		}
	}
}

HandlerAdapter

初始化：

private void initHandlerAdapters(ApplicationContext context) {
		this.handlerAdapters = null;

		if (this.detectAllHandlerAdapters) {
			// Find all HandlerAdapters in the ApplicationContext, including ancestor contexts.
			Map<String, HandlerAdapter> matchingBeans =
					BeanFactoryUtils.beansOfTypeIncludingAncestors(context, HandlerAdapter.class, true, false);
			if (!matchingBeans.isEmpty()) {
				this.handlerAdapters = new ArrayList<HandlerAdapter>(matchingBeans.values());
				// We keep HandlerAdapters in sorted order.
				AnnotationAwareOrderComparator.sort(this.handlerAdapters);
			}
		}
		else {
			try {
				HandlerAdapter ha = context.getBean(HANDLER_ADAPTER_BEAN_NAME, HandlerAdapter.class);
				this.handlerAdapters = Collections.singletonList(ha);
			}
			catch (NoSuchBeanDefinitionException ex) {
				// Ignore, we'll add a default HandlerAdapter later.
			}
		}

		// Ensure we have at least some HandlerAdapters, by registering
		// default HandlerAdapters if no other adapters are found.
		if (this.handlerAdapters == null) {
			this.handlerAdapters = getDefaultStrategies(context, HandlerAdapter.class);
			if (logger.isDebugEnabled()) {
				logger.debug("No HandlerAdapters found in servlet '" + getServletName() + "': using default");
			}
		}
	}

AbstractHandlerMethodAdapter#handle

	@Override
	public final ModelAndView handle(HttpServletRequest request, HttpServletResponse response, Object handler)
			throws Exception {

		return handleInternal(request, response, (HandlerMethod) handler);
	}

@Override
	protected ModelAndView handleInternal(HttpServletRequest request,
			HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {

		ModelAndView mav;
//检查request是支持的method和是否需要session
		checkRequest(request);

		//如果session级别，则同步
		if (this.synchronizeOnSession) {
			HttpSession session = request.getSession(false);
			if (session != null) {
				Object mutex = WebUtils.getSessionMutex(session);
				synchronized (mutex) {
					mav = invokeHandlerMethod(request, response, handlerMethod);
				}
			}
			else {
				// No HttpSession available -> no mutex necessary
				mav = invokeHandlerMethod(request, response, handlerMethod);
			}
		}
		else {
			// No synchronization on session demanded at all...
			mav = invokeHandlerMethod(request, response, handlerMethod);
		}

		if (!response.containsHeader(HEADER_CACHE_CONTROL)) {
			if (getSessionAttributesHandler(handlerMethod).hasSessionAttributes()) {
				applyCacheSeconds(response, this.cacheSecondsForSessionAttributeHandlers);
			}
			else {
				prepareResponse(response);
			}
		}

		return mav;
	}



protected ModelAndView invokeHandlerMethod(HttpServletRequest request,
			HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {

		ServletWebRequest webRequest = new ServletWebRequest(request, response);
		try {
			WebDataBinderFactory binderFactory = getDataBinderFactory(handlerMethod);
			ModelFactory modelFactory = getModelFactory(handlerMethod, binderFactory);

			ServletInvocableHandlerMethod invocableMethod = createInvocableHandlerMethod(handlerMethod);
			invocableMethod.setHandlerMethodArgumentResolvers(this.argumentResolvers);
			invocableMethod.setHandlerMethodReturnValueHandlers(this.returnValueHandlers);
			invocableMethod.setDataBinderFactory(binderFactory);
			invocableMethod.setParameterNameDiscoverer(this.parameterNameDiscoverer);

			ModelAndViewContainer mavContainer = new ModelAndViewContainer();
			mavContainer.addAllAttributes(RequestContextUtils.getInputFlashMap(request));
			modelFactory.initModel(webRequest, mavContainer, invocableMethod);
			mavContainer.setIgnoreDefaultModelOnRedirect(this.ignoreDefaultModelOnRedirect);

			AsyncWebRequest asyncWebRequest = WebAsyncUtils.createAsyncWebRequest(request, response);
			asyncWebRequest.setTimeout(this.asyncRequestTimeout);

			WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
			asyncManager.setTaskExecutor(this.taskExecutor);
			asyncManager.setAsyncWebRequest(asyncWebRequest);
			asyncManager.registerCallableInterceptors(this.callableInterceptors);
			asyncManager.registerDeferredResultInterceptors(this.deferredResultInterceptors);

			if (asyncManager.hasConcurrentResult()) {
				Object result = asyncManager.getConcurrentResult();
				mavContainer = (ModelAndViewContainer) asyncManager.getConcurrentResultContext()[0];
				asyncManager.clearConcurrentResult();
				if (logger.isDebugEnabled()) {
					logger.debug("Found concurrent result value [" + result + "]");
				}
				invocableMethod = invocableMethod.wrapConcurrentResult(result);
			}
//最重要的地方，执行
			invocableMethod.invokeAndHandle(webRequest, mavContainer);
			if (asyncManager.isConcurrentHandlingStarted()) {
				return null;
			}

			return getModelAndView(mavContainer, modelFactory, webRequest);
		}
		finally {
			webRequest.requestCompleted();
		}
	}

ServletInvocableHandlerMethod

public void invokeAndHandle(ServletWebRequest webRequest,
			ModelAndViewContainer mavContainer, Object... providedArgs) throws Exception {

		Object returnValue = invokeForRequest(webRequest, mavContainer, providedArgs);
		setResponseStatus(webRequest);

		if (returnValue == null) {
			if (isRequestNotModified(webRequest) || hasResponseStatus() || mavContainer.isRequestHandled()) {
				mavContainer.setRequestHandled(true);
				return;
			}
		}
		else if (StringUtils.hasText(this.responseReason)) {
			mavContainer.setRequestHandled(true);
			return;
		}

		mavContainer.setRequestHandled(false);
		try {
			this.returnValueHandlers.handleReturnValue(
					returnValue, getReturnValueType(returnValue), mavContainer, webRequest);
		}
		catch (Exception ex) {
			if (logger.isTraceEnabled()) {
				logger.trace(getReturnValueHandlingErrorMessage("Error handling return value", returnValue), ex);
			}
			throw ex;
		}
	}

public Object invokeForRequest(NativeWebRequest request, ModelAndViewContainer mavContainer,
			Object... providedArgs) throws Exception {

		Object[] args = getMethodArgumentValues(request, mavContainer, providedArgs);
		if (logger.isTraceEnabled()) {
			StringBuilder sb = new StringBuilder("Invoking [");
			sb.append(getBeanType().getSimpleName()).append(".");
			sb.append(getMethod().getName()).append("] method with arguments ");
			sb.append(Arrays.asList(args));
			logger.trace(sb.toString());
		}
		Object returnValue = doInvoke(args);
		if (logger.isTraceEnabled()) {
			logger.trace("Method [" + getMethod().getName() + "] returned [" + returnValue + "]");
		}
		return returnValue;
	}


protected Object doInvoke(Object... args) throws Exception {
		ReflectionUtils.makeAccessible(getBridgedMethod());
		try {
//通过反射
			return getBridgedMethod().invoke(getBean(), args);
		}
		catch (IllegalArgumentException ex) {
			assertTargetBean(getBridgedMethod(), getBean(), args);
			String message = (ex.getMessage() != null ? ex.getMessage() : "Illegal argument");
			throw new IllegalStateException(getInvocationErrorMessage(message, args), ex);
		}
		catch (InvocationTargetException ex) {
			// Unwrap for HandlerExceptionResolvers ...
			Throwable targetException = ex.getTargetException();
			if (targetException instanceof RuntimeException) {
				throw (RuntimeException) targetException;
			}
			else if (targetException instanceof Error) {
				throw (Error) targetException;
			}
			else if (targetException instanceof Exception) {
				throw (Exception) targetException;
			}
			else {
				String msg = getInvocationErrorMessage("Failed to invoke controller method", args);
				throw new IllegalStateException(msg, targetException);
			}
		}
	}

HandlerMapping

题外，这个是如何初始化的：

private void initHandlerMappings(ApplicationContext context) {
		this.handlerMappings = null;

		if (this.detectAllHandlerMappings) {//判断是否默认添加所有的HandlerMappings,初始值是默认添加的
			// Find all HandlerMappings in the ApplicationContext, including ancestor contexts.
			Map<String, HandlerMapping> matchingBeans =
					BeanFactoryUtils.beansOfTypeIncludingAncestors(context, HandlerMapping.class, true, false);
			if (!matchingBeans.isEmpty()) {
				this.handlerMappings = new ArrayList<HandlerMapping>(matchingBeans.values());
				// 通过@order注解去排序
				AnnotationAwareOrderComparator.sort(this.handlerMappings);
			}
		}
//如果不是默认添加所有的，那么就去context中找一个声明的bean
		else {
			try {
				HandlerMapping hm = context.getBean(HANDLER_MAPPING_BEAN_NAME, HandlerMapping.class);
				this.handlerMappings = Collections.singletonList(hm);
			}
			catch (NoSuchBeanDefinitionException ex) {
				// Ignore, we'll add a default HandlerMapping later.
			}
		}

		// Ensure we have at least one HandlerMapping, by registering
		// a default HandlerMapping if no other mappings are found.
//如果上面两步没有找到可以使用的handlerMapping，那么就采用默认的handlerMapping
        //默认的HandlerMapping都定义在了DispatcherServlet.properties中，大致定义了如下两个
		if (this.handlerMappings == null) {
			this.handlerMappings = getDefaultStrategies(context, HandlerMapping.class);
			if (logger.isDebugEnabled()) {
				logger.debug("No HandlerMappings found in servlet '" + getServletName() + "': using default");
			}
		}
	}

redis设计与实现阅读笔记

Posted on 2017-09-28 Edited on 2022-01-28 In 数据库

1.redis数据结构

简单字符串

redis专门封装了一个叫SDS的数据结构

struct sdshdr {
    // 记录 buf 数组中已使用字节的数量
    // 等于 SDS 所保存字符串的长度
    int len;

    // 记录 buf 数组中未使用字节的数量
    int free;

    // 字节数组，用于保存字符串
    char buf[];
};

free 属性的值为 0 ，表示这个 SDS 没有分配任何未使用空间。
len 属性的值为 5 ，表示这个 SDS 保存了一个五字节长的字符串。（strlen的复杂度就从O(n)变为O(1)）
buf 属性是一个 char 类型的数组，数组的前五个字节分别保存了 ‘R’ 、 ‘e’ 、 ‘d’ 、 ‘i’ 、 ‘s’ 五个字符，而最后一个字节则保存了空字符 ‘\0’ 。

2015-09-13_55f50d86a66ae

优点：

常数复杂度获取字符串长度
杜绝缓冲区溢出
举个例子， <string.h>/strcat 函数可以将 src 字符串中的内容拼接到 dest 字符串的末尾：

char *strcat(char *dest, const char *src);

因为 C 字符串不记录自身的长度，所以 strcat 假定用户在执行这个函数时，已经为 dest 分配了足够多的内存，可以容纳 src 字符串中的所有内容，而一旦这个假定不成立时，就会产生缓冲区溢出。
举个例子，假设程序里有两个在内存中紧邻着的 C 字符串 s1 和 s2 ，其中 s1 保存了字符串 “Redis” ，而 s2 则保存了字符串 “MongoDB”，如图 2-7 所示。
2015-09-13_55f50e28c1620

strcat(s1, " Cluster");

将 s1 的内容修改为 “Redis Cluster” ，但粗心的他却忘了在执行 strcat 之前为 s1 分配足够的空间，那么在 strcat 函数执行之后， s1 的数据将溢出到 s2 所在的空间中，导致 s2 保存的内容被意外地修改，如图 2-8 所示。
2015-09-13_55f50e29e04a2

与 C 字符串不同， SDS 的空间分配策略完全杜绝了发生缓冲区溢出的可能性：当 SDS API 需要对 SDS 进行修改时， API 会先检查 SDS 的空间是否满足修改所需的要求，如果不满足的话， API 会自动将 SDS 的空间扩展至执行修改所需的大小，然后才执行实际的修改操作，所以使用 SDS 既不需要手动修改 SDS 的空间大小，也不会出现前面所说的缓冲区溢出问题。

举个例子， SDS 的 API 里面也有一个用于执行拼接操作的 sdscat 函数，它可以将一个 C 字符串拼接到给定 SDS 所保存的字符串的后面，但是在执行拼接操作之前， sdscat 会先检查给定 SDS 的空间是否足够，如果不够的话， sdscat 就会先扩展 SDS 的空间，然后才执行拼接操作。

减少修改字符串时带来的内存重分配次数
因为 C 字符串并不记录自身的长度，所以对于一个包含了 N 个字符的 C 字符串来说，这个 C 字符串的底层实现总是一个 N+1 个字符长的数组（额外的一个字符空间用于保存空字符）。
因为 C 字符串的长度和底层数组的长度之间存在着这种关联性，所以每次增长或者缩短一个 C 字符串，程序都总要对保存这个 C 字符串的数组进行一次内存重分配操作：

如果程序执行的是增长字符串的操作，比如拼接操作（append），那么在执行这个操作之前，程序需要先通过内存重分配来扩展底层数组的空间大小 —— 如果忘了这一步就会产生缓冲区溢出。
如果程序执行的是缩短字符串的操作，比如截断操作（trim），那么在执行这个操作之后，程序需要通过内存重分配来释放字符串不再使用的那部分空间 —— 如果忘了这一步就会产生内存泄漏。
举个例子，如果我们持有一个值为 “Redis” 的 C 字符串 s ，那么为了将 s 的值改为 “Redis Cluster” ，在执行：
strcat(s, " Cluster");

为了避免 C 字符串的这种缺陷， SDS 通过未使用空间解除了字符串长度和底层数组长度之间的关联：在 SDS 中， buf 数组的长度不一定就是字符数量加一，数组里面可以包含未使用的字节，而这些字节的数量就由 SDS 的 free 属性记录。
通过未使用空间， SDS 实现了空间预分配和惰性空间释放两种优化策略。

空间预分配
空间预分配用于优化 SDS 的字符串增长操作：当 SDS 的 API 对一个 SDS 进行修改，并且需要对 SDS 进行空间扩展的时候，程序不仅会为 SDS 分配修改所必须要的空间，还会为 SDS 分配额外的未使用空间。

其中，额外分配的未使用空间数量由以下公式决定：

如果对 SDS 进行修改之后， SDS 的长度（也即是 len 属性的值）将小于 1 MB ，那么程序分配和 len 属性同样大小的未使用空间，这时 SDS len 属性的值将和 free 属性的值相同。举个例子，如果进行修改之后， SDS 的 len 将变成 13 字节，那么程序也会分配13 字节的未使用空间， SDS 的 buf 数组的实际长度将变成 13 + 13 + 1 = 27 字节（额外的一字节用于保存空字符）。

如果对 SDS 进行修改之后， SDS 的长度将大于等于 1 MB ，那么程序会分配 1 MB 的未使用空间。举个例子，如果进行修改之后， SDS 的 len 将变成 30 MB ，那么程序会分配 1 MB 的未使用空间， SDS 的 buf 数组的实际长度将为 30 MB + 1 MB + 1 byte 。

sdscat(s, " Cluster");

那么 sdscat 将执行一次内存重分配操作，将 SDS 的长度修改为 13 字节，并将 SDS 的未使用空间同样修改为 13 字节，如图 2-12 所示。

2015-09-13_55f50e329713a

2015-09-13_55f50e33eb3a4

如果这时，我们再次对 s 执行：
sdscat(s, " Tutorial");
那么这次 sdscat 将不需要执行内存重分配：因为未使用空间里面的 13 字节足以保存 9 字节的 “ Tutorial” ，执行 sdscat 之后的 SDS 如图 2-13 所示。

在扩展 SDS 空间之前， SDS API 会先检查未使用空间是否足够，如果足够的话， API 就会直接使用未使用空间，而无须执行内存重分配。

通过这种预分配策略， SDS 将连续增长 N 次字符串所需的内存重分配次数从必定 N 次降低为最多 N 次。

惰性空间释放

惰性空间释放用于优化 SDS 的字符串缩短操作：当 SDS 的 API 需要缩短 SDS 保存的字符串时，程序并不立即使用内存重分配来回收缩短后多出来的字节，而是使用 free 属性将这些字节的数量记录起来，并等待将来使用。
举个例子， sdstrim 函数接受一个 SDS 和一个 C 字符串作为参数，从 SDS 左右两端分别移除所有在 C 字符串中出现过的字符。

sdstrim(s, “XY”); // 移除 SDS 字符串中的所有 ‘X’ 和 ‘Y’

会将 SDS 修改成图 2-15 所示的样子。

注意执行 sdstrim 之后的 SDS 并没有释放多出来的 8 字节空间，而是将这 8 字节空间作为未使用空间保留在了 SDS 里面，如果将来要对 SDS 进行增长操作的话，这些未使用空间就可能会派上用场。

二进制安全
SDS 的 API 都是二进制安全的（binary-safe）：所有 SDS API 都会以处理二进制的方式来处理 SDS 存放在 buf 数组里的数据，程序不会对其中的数据做任何限制、过滤、或者假设 —— 数据在写入时是什么样的，它被读取时就是什么样。

兼容部分 C 字符串函数

链表

redis> LLEN integers
(integer) 1024

redis> LRANGE integers 0 10
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
6) "6"
7) "7"
8) "8"
9) "9"
10) "10"
11) "11"

typedef struct listNode {

    // 前置节点
    struct listNode *prev;

    // 后置节点
    struct listNode *next;

    // 节点的值
    void *value;

} listNode;

多个 listNode 可以通过 prev 和 next 指针组成双端链表，如图 3-1 所示。

2015-09-13_55f50fad082e3

typedef struct list {

    // 表头节点
    listNode *head;

    // 表尾节点
    listNode *tail;

    // 链表所包含的节点数量
    unsigned long len;

    // 节点值复制函数
    void *(*dup)(void *ptr);

    // 节点值释放函数
    void (*free)(void *ptr);

    // 节点值对比函数
    int (*match)(void *ptr, void *key);

} list;

list 结构为链表提供了表头指针 head 、表尾指针 tail ，以及链表长度计数器 len ，而 dup 、 free 和 match 成员则是用于实现多态链表所需的类型特定函数：

dup 函数用于复制链表节点所保存的值；
free 函数用于释放链表节点所保存的值；
match 函数则用于对比链表节点所保存的值和另一个输入值是否相等。

2015-09-13_55f50fb39b6cb

Redis 的链表实现的特性可以总结如下：

双端：链表节点带有 prev 和 next 指针，获取某个节点的前置节点和后置节点的复杂度都是 O(1) 。
无环：表头节点的 prev 指针和表尾节点的 next 指针都指向 NULL ，对链表的访问以 NULL 为终点。
带表头指针和表尾指针：通过 list 结构的 head 指针和 tail 指针，程序获取链表的表头节点和表尾节点的复杂度为 O(1) 。
带链表长度计数器：程序使用 list 结构的 len 属性来对 list 持有的链表节点进行计数，程序获取链表中节点数量的复杂度为 O(1)。
多态：链表节点使用 void* 指针来保存节点值，并且可以通过 list 结构的 dup 、 free 、 match 三个属性为节点值设置类型特定函数，所以链表可以用于保存各种不同类型的值。

字典

字典，又称符号表（symbol table）、关联数组（associative array）或者映射（map），是一种用于保存键值对（key-value pair）的抽象数据结构。

redis> HLEN website
(integer) 10086

redis> HGETALL website
1) "Redis"
2) "Redis.io"
3) "MariaDB"
4) "MariaDB.org"
5) "MongoDB"
6) "MongoDB.org"
# ...

其中一个键值对的键为 “Redis” ，值为 “Redis.io” 。
另一个键值对的键为 “MariaDB” ，值为 “MariaDB.org” ；
还有一个键值对的键为 “MongoDB” ，值为 “MongoDB.org” ；

typedef struct dictht {

    // 哈希表数组
    dictEntry **table;

    // 哈希表大小
    unsigned long size;

    // 哈希表大小掩码，用于计算索引值
    // 总是等于 size - 1
    unsigned long sizemask;

    // 该哈希表已有节点的数量
    unsigned long used;

} dictht;

table 属性是一个数组，数组中的每个元素都是一个指向 dict.h/dictEntry 结构的指针，每个 dictEntry 结构保存着一个键值对。
size 属性记录了哈希表的大小，也即是 table 数组的大小，而 used 属性则记录了哈希表目前已有节点（键值对）的数量。
sizemask 属性的值总是等于 size - 1 ，这个属性和哈希值一起决定一个键应该被放到 table 数组的哪个索引上面。

2015-09-13_55f511fc9428c

哈希表节点

typedef struct dictEntry {

    // 键
    void *key;

    // 值
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
    } v;

    // 指向下个哈希表节点，形成链表
    struct dictEntry *next;

} dictEntry;

key 属性保存着键值对中的键，而 v 属性则保存着键值对中的值，其中键值对的值可以是一个指针，或者是一个 uint64_t 整数，又或者是一个 int64_t 整数。

next 属性是指向另一个哈希表节点的指针，这个指针可以将多个哈希值相同的键值对连接在一次，以此来解决键冲突（collision）的问题。

举个例子，图 4-2 就展示了如何通过 next 指针，将两个索引值相同的键 k1 和 k0 连接在一起。

2015-09-13_55f51205335f9

字典

typedef struct dict {

    // 类型特定函数
    dictType *type;

    // 私有数据
    void *privdata;

    // 哈希表
    dictht ht[2];

    // rehash 索引
    // 当 rehash 不在进行时，值为 -1
    int rehashidx; /* rehashing not in progress if rehashidx == -1 */

} dict;

type 属性和 privdata 属性是针对不同类型的键值对，为创建多态字典而设置的：

type 属性是一个指向 dictType 结构的指针，每个 dictType 结构保存了一簇用于操作特定类型键值对的函数， Redis 会为用途不同的字典设置不同的类型特定函数。

而 privdata 属性则保存了需要传给那些类型特定函数的可选参数。

typedef struct dictType {

    // 计算哈希值的函数
    unsigned int (*hashFunction)(const void *key);

    // 复制键的函数
    void *(*keyDup)(void *privdata, const void *key);

    // 复制值的函数
    void *(*valDup)(void *privdata, const void *obj);

    // 对比键的函数
    int (*keyCompare)(void *privdata, const void *key1, const void *key2);

    // 销毁键的函数
    void (*keyDestructor)(void *privdata, void *key);

    // 销毁值的函数
    void (*valDestructor)(void *privdata, void *obj);

} dictType;

ht 属性是一个包含两个项的数组，数组中的每个项都是一个 dictht 哈希表，一般情况下，字典只使用 ht[0] 哈希表， ht[1] 哈希表只会在对 ht[0] 哈希表进行 rehash 时使用。

除了 ht[1] 之外，另一个和 rehash 有关的属性就是 rehashidx ：它记录了 rehash 目前的进度，如果目前没有在进行 rehash ，那么它的值为 -1 。

图 4-3 展示了一个普通状态下（没有进行 rehash）的字典：

2015-09-13_55f5120772706

哈希算法

当要将一个新的键值对添加到字典里面时，程序需要先根据键值对的键计算出哈希值和索引值，然后再根据索引值，将包含新键值对的哈希表节点放到哈希表数组的指定索引上面。

Redis 计算哈希值和索引值的方法如下：

# 使用字典设置的哈希函数，计算键 key 的哈希值
hash = dict->type->hashFunction(key);

# 使用哈希表的 sizemask 属性和哈希值，计算出索引值
# 根据情况不同， ht[x] 可以是 ht[0] 或者 ht[1]
index = hash & dict->ht[x].sizemask;

2015-09-13_55f5128c90054

举个例子，对于图 4-4 所示的字典来说，如果我们要将一个键值对 k0 和 v0 添加到字典里面，那么程序会先使用语句：
hash = dict->type->hashFunction(k0);
计算键 k0 的哈希值。
假设计算得出的哈希值为 8 ，那么程序会继续使用语句：
index = hash & dict->ht[0].sizemask = 8 & 3 = 0;
计算出键 k0 的索引值 0 ，这表示包含键值对 k0 和 v0 的节点应该被放置到哈希表数组的索引 0 位置上，如图 4-5 所示。
2015-09-13_55f51293b4642

当字典被用作数据库的底层实现，或者哈希键的底层实现时， Redis 使用 MurmurHash2 算法来计算键的哈希值。

MurmurHash 算法最初由 Austin Appleby 于 2008 年发明，这种算法的优点在于，即使输入的键是有规律的，算法仍能给出一个很好的随机分布性，并且算法的计算速度也非常快。

MurmurHash 算法目前的最新版本为 MurmurHash3 ，而 Redis 使用的是 MurmurHash2 ，关于 MurmurHash 算法的更多信息可以参考该算法的主页： http://code.google.com/p/smhasher/

解决键冲突

当有两个或以上数量的键被分配到了哈希表数组的同一个索引上面时，我们称这些键发生了冲突（collision）。

Redis 的哈希表使用链地址法（separate chaining）来解决键冲突：每个哈希表节点都有一个 next 指针，多个哈希表节点可以用 next 指针构成一个单向链表，被分配到同一个索引上的多个节点可以用这个单向链表连接起来，这就解决了键冲突的问题。
举个例子，假设程序要将键值对 k2 和 v2 添加到图 4-6 所示的哈希表里面，并且计算得出 k2 的索引值为 2 ，那么键 k1 和 k2 将产生冲突，而解决冲突的办法就是使用 next 指针将键 k2 和 k1 所在的节点连接起来，如图 4-7 所示。

2015-09-13_55f512c4d7f99

2015-09-13_55f512c2524e5

因为 dictEntry 节点组成的链表没有指向链表表尾的指针，所以为了速度考虑，程序总是将新节点添加到链表的表头位置（复杂度为 O(1)），排在其他已有节点的前面。

rehash
随着操作的不断执行，哈希表保存的键值对会逐渐地增多或者减少，为了让哈希表的负载因子（load factor）维持在一个合理的范围之内，当哈希表保存的键值对数量太多或者太少时，程序需要对哈希表的大小进行相应的扩展或者收缩。

扩展和收缩哈希表的工作可以通过执行 rehash （重新散列）操作来完成， Redis 对字典的哈希表执行 rehash 的步骤如下：

为字典的 ht[1] 哈希表分配空间，这个哈希表的空间大小取决于要执行的操作，以及 ht[0] 当前包含的键值对数量（也即是ht[0].used 属性的值）：
如果执行的是扩展操作，那么 ht[1] 的大小为第一个大于等于 ht[0].used * 2 的 2^n （2 的 n 次方幂）；
如果执行的是收缩操作，那么 ht[1] 的大小为第一个大于等于 ht[0].used 的 2^n 。
将保存在 ht[0] 中的所有键值对 rehash 到 ht[1] 上面： rehash 指的是重新计算键的哈希值和索引值，然后将键值对放置到 ht[1] 哈希表的指定位置上。

当 ht[0] 包含的所有键值对都迁移到了 ht[1] 之后（ht[0] 变为空表），释放 ht[0] ，将 ht[1] 设置为 ht[0] ，并在 ht[1] 新创建一个空白哈希表，为下一次 rehash 做准备。

举个例子，假设程序要对图 4-8 所示字典的 ht[0] 进行扩展操作，那么程序将执行以下步骤：
ht[0].used 当前的值为 4 ， 4 * 2 = 8 ，而 8 （2^3）恰好是第一个大于等于 4 的 2 的 n 次方，所以程序会将 ht[1] 哈希表的大小设置为 8 。图 4-9 展示了 ht[1] 在分配空间之后，字典的样子。
将 ht[0] 包含的四个键值对都 rehash 到 ht[1] ，如图 4-10 所示。
释放 ht[0] ，并将 ht[1] 设置为 ht[0] ，然后为 ht[1] 分配一个空白哈希表，如图 4-11 所示。
至此，对哈希表的扩展操作执行完毕，程序成功将哈希表的大小从原来的 4 改为了现在的 8 。

2015-09-13_55f5130162f2d

2015-09-13_55f51302b6785

2015-09-13_55f51309b4775

2015-09-13_55f5130b2ec57

哈希表的扩展与收缩

当以下条件中的任意一个被满足时，程序会自动开始对哈希表执行扩展操作：

服务器目前没有在执行 BGSAVE 命令或者 BGREWRITEAOF 命令，并且哈希表的负载因子大于等于 1 ；
服务器目前正在执行 BGSAVE 命令或者 BGREWRITEAOF 命令，并且哈希表的负载因子大于等于 5 ；
其中哈希表的负载因子可以通过公式：

1 2	# 负载因子 = 哈希表已保存节点数量 / 哈希表大小 load_factor = ht[0].used / ht[0].size

计算得出。

比如说，对于一个大小为 4 ，包含 4 个键值对的哈希表来说，这个哈希表的负载因子为：
load_factor = 4 / 4 = 1

又比如说，对于一个大小为 512 ，包含 256 个键值对的哈希表来说，这个哈希表的负载因子为：
load_factor = 256 / 512 = 0.5

根据 BGSAVE 命令或 BGREWRITEAOF 命令是否正在执行，服务器执行扩展操作所需的负载因子并不相同，这是因为在执行 BGSAVE 命令或BGREWRITEAOF 命令的过程中， Redis 需要创建当前服务器进程的子进程，而大多数操作系统都采用写时复制（copy-on-write）技术来优化子进程的使用效率，所以在子进程存在期间，服务器会提高执行扩展操作所需的负载因子，从而尽可能地避免在子进程存在期间进行哈希表扩展操作，这可以避免不必要的内存写入操作，最大限度地节约内存。
另一方面，当哈希表的负载因子小于 0.1 时，程序自动开始对哈希表执行收缩操作。

渐进式 rehash
上一节说过，扩展或收缩哈希表需要将 ht[0] 里面的所有键值对 rehash 到 ht[1] 里面，但是，这个 rehash 动作并不是一次性、集中式地完成的，而是分多次、渐进式地完成的。

这样做的原因在于，如果 ht[0] 里只保存着四个键值对，那么服务器可以在瞬间就将这些键值对全部 rehash 到 ht[1] ；但是，如果哈希表里保存的键值对数量不是四个，而是四百万、四千万甚至四亿个键值对，那么要一次性将这些键值对全部 rehash 到 ht[1] 的话，庞大的计算量可能会导致服务器在一段时间内停止服务。

因此，为了避免 rehash 对服务器性能造成影响，服务器不是一次性将 ht[0] 里面的所有键值对全部 rehash 到 ht[1] ，而是分多次、渐进式地将 ht[0] 里面的键值对慢慢地 rehash 到 ht[1] 。
以下是哈希表渐进式 rehash 的详细步骤：

为 ht[1] 分配空间，让字典同时持有 ht[0] 和 ht[1] 两个哈希表。
在字典中维持一个索引计数器变量 rehashidx ，并将它的值设置为 0 ，表示 rehash 工作正式开始。
在 rehash 进行期间，每次对字典执行添加、删除、查找或者更新操作时，程序除了执行指定的操作以外，还会顺带将 ht[0] 哈希表在 rehashidx 索引上的所有键值对 rehash 到 ht[1] ，当 rehash 工作完成之后，程序将 rehashidx 属性的值增一。
随着字典操作的不断执行，最终在某个时间点上， ht[0] 的所有键值对都会被 rehash 至 ht[1] ，这时程序将 rehashidx 属性的值设为 -1 ，表示 rehash 操作已完成。
渐进式 rehash 的好处在于它采取分而治之的方式，将 rehash 键值对所需的计算工作均滩到对字典的每个添加、删除、查找和更新操作上，从而避免了集中式 rehash 而带来的庞大计算量。

https://www.kancloud.cn/kancloud/redisbook/63842

跳跃表

跳跃表（skiplist）是一种有序数据结构，它通过在每个节点中维持多个指向其他节点的指针，从而达到快速访问节点的目的。

跳跃表支持平均 O(\log N) 最坏 O(N) 复杂度的节点查找，还可以通过顺序性操作来批量处理节点。
在大部分情况下，跳跃表的效率可以和平衡树相媲美，并且因为跳跃表的实现比平衡树要来得更为简单，所以有不少程序都使用跳跃表来代替平衡树。

Redis 使用跳跃表作为有序集合键的底层实现之一：如果一个有序集合包含的元素数量比较多，又或者有序集合中元素的成员（member）是比较长的字符串时， Redis 就会使用跳跃表来作为有序集合键的底层实现。

举个例子， fruit-price 是一个有序集合键，这个有序集合以水果名为成员，水果价钱为分值，保存了 130 款水果的价钱：

redis> ZRANGE fruit-price 0 2 WITHSCORES
1) "banana"
2) "5"
3) "cherry"
4) "6.5"
5) "apple"
6) "8"

redis> ZCARD fruit-price
(integer) 130

fruit-price 有序集合的所有数据都保存在一个跳跃表里面，其中每个跳跃表节点（node）都保存了一款水果的价钱信息，所有水果按价钱的高低从低到高在跳跃表里面排序：

和链表、字典等数据结构被广泛地应用在 Redis 内部不同， Redis 只在两个地方用到了跳跃表，一个是实现有序集合键，另一个是在集群节点中用作内部数据结构，除此之外，跳跃表在 Redis 里面没有其他用途。

Redis 的跳跃表由 redis.h/zskiplistNode 和 redis.h/zskiplist 两个结构定义，其中 zskiplistNode 结构用于表示跳跃表节点，而 zskiplist结构则用于保存跳跃表节点的相关信息，比如节点的数量，以及指向表头节点和表尾节点的指针，等等。

2015-09-13_55f51478611a6

图 5-1 展示了一个跳跃表示例，位于图片最左边的是 zskiplist 结构，该结构包含以下属性：
header ：指向跳跃表的表头节点。
tail ：指向跳跃表的表尾节点。
level ：记录目前跳跃表内，层数最大的那个节点的层数（表头节点的层数不计算在内）。
length ：记录跳跃表的长度，也即是，跳跃表目前包含节点的数量（表头节点不计算在内）。
位于 zskiplist 结构右方的是四个 zskiplistNode 结构，该结构包含以下属性：
层（level）：节点中用 L1 、 L2 、 L3 等字样标记节点的各个层， L1 代表第一层， L2 代表第二层，以此类推。每个层都带有两个属性：前进指针和跨度。前进指针用于访问位于表尾方向的其他节点，而跨度则记录了前进指针所指向节点和当前节点的距离。在上面的图片中，连线上带有数字的箭头就代表前进指针，而那个数字就是跨度。当程序从表头向表尾进行遍历时，访问会沿着层的前进指针进行。
后退（backward）指针：节点中用 BW 字样标记节点的后退指针，它指向位于当前节点的前一个节点。后退指针在程序从表尾向表头遍历时使用。
分值（score）：各个节点中的 1.0 、 2.0 和 3.0 是节点所保存的分值。在跳跃表中，节点按各自所保存的分值从小到大排列。
成员对象（obj）：各个节点中的 o1 、 o2 和 o3 是节点所保存的成员对象。
注意表头节点和其他节点的构造是一样的：表头节点也有后退指针、分值和成员对象，不过表头节点的这些属性都不会被用到，所以图中省略了这些部分，只显示了表头节点的各个层。

typedef struct zskiplistNode {

    // 后退指针
    struct zskiplistNode *backward;

    // 分值
    double score;

    // 成员对象
    robj *obj;

    // 层
    struct zskiplistLevel {

        // 前进指针
        struct zskiplistNode *forward;

        // 跨度
        unsigned int span;

    } level[];

} zskiplistNode;

层
跳跃表节点的 level 数组可以包含多个元素，每个元素都包含一个指向其他节点的指针，程序可以通过这些层来加快访问其他节点的速度，一般来说，层的数量越多，访问其他节点的速度就越快。

每次创建一个新跳跃表节点的时候，程序都根据幂次定律（power law，越大的数出现的概率越小）随机生成一个介于 1 和 32 之间的值作为 level 数组的大小，这个大小就是层的“高度”。

前进指针

每个层都有一个指向表尾方向的前进指针（level[i].forward 属性），用于从表头向表尾方向访问节点。
图 5-3 用虚线表示出了程序从表头向表尾方向，遍历跳跃表中所有节点的路径：

1.迭代程序首先访问跳跃表的第一个节点（表头），然后从第四层的前进指针移动到表中的第二个节点。
2.在第二个节点时，程序沿着第二层的前进指针移动到表中的第三个节点。
3.在第三个节点时，程序同样沿着第二层的前进指针移动到表中的第四个节点。
4.当程序再次沿着第四个节点的前进指针移动时，它碰到一个 NULL ，程序知道这时已经到达了跳跃表的表尾，于是结束这次遍历。

跨度
层的跨度（level[i].span 属性）用于记录两个节点之间的距离：
两个节点之间的跨度越大，它们相距得就越远。
指向 NULL 的所有前进指针的跨度都为 0 ，因为它们没有连向任何节点。
初看上去，很容易以为跨度和遍历操作有关，但实际上并不是这样 —— 遍历操作只使用前进指针就可以完成了，跨度实际上是用来计算排位（rank）的：在查找某个节点的过程中，将沿途访问过的所有层的跨度累计起来，得到的结果就是目标节点在跳跃表中的排位。

举个例子，图 5-4 用虚线标记了在跳跃表中查找分值为 3.0 、成员对象为 o3 的节点时，沿途经历的层：查找的过程只经过了一个层，并且层的跨度为 3 ，所以目标节点在跳跃表中的排位为 3 。

后退指针

节点的后退指针（backward 属性）用于从表尾向表头方向访问节点：跟可以一次跳过多个节点的前进指针不同，因为每个节点只有一个后退指针，所以每次只能后退至前一个节点。
图 5-6 用虚线展示了如果从表尾向表头遍历跳跃表中的所有节点：程序首先通过跳跃表的 tail 指针访问表尾节点，然后通过后退指针访问倒数第二个节点，之后再沿着后退指针访问倒数第三个节点，再之后遇到指向 NULL 的后退指针，于是访问结束。

分值和成员
节点的分值（score 属性）是一个 double 类型的浮点数，跳跃表中的所有节点都按分值从小到大来排序。
节点的成员对象（obj 属性）是一个指针，它指向一个字符串对象，而字符串对象则保存着一个 SDS 值。
在同一个跳跃表中，各个节点保存的成员对象必须是唯一的，但是多个节点保存的分值却可以是相同的：分值相同的节点将按照成员对象在字典序中的大小来进行排序，成员对象较小的节点会排在前面（靠近表头的方向），而成员对象较大的节点则会排在后面（靠近表尾的方向）。

举个例子，在图 5-7 所示的跳跃表中，三个跳跃表节点都保存了相同的分值 10086.0 ，但保存成员对象 o1 的节点却排在保存成员对象 o2和 o3 的节点之前，而保存成员对象 o2 的节点又排在保存成员对象 o3 的节点之前，由此可见， o1 、 o2 、 o3 三个成员对象在字典中的排序为 o1 <= o2 <= o3 。

跳跃表

typedef struct zskiplist {

    // 表头节点和表尾节点
    struct zskiplistNode *header, *tail;

    // 表中节点的数量
    unsigned long length;

    // 表中层数最大的节点的层数
    int level;

} zskiplist;

整数集合

整数集合（intset）是集合键的底层实现之一：当一个集合只包含整数值元素，并且这个集合的元素数量不多时， Redis 就会使用整数集合作为集合键的底层实现。

redis> SADD numbers 1 3 5 7 9
(integer) 5

redis> OBJECT ENCODING numbers
"intset"

typedef struct intset {

    // 编码方式
    uint32_t encoding;

    // 集合包含的元素数量
    uint32_t length;

    // 保存元素的数组
    int8_t contents[];

} intset;

contents 数组是整数集合的底层实现：整数集合的每个元素都是 contents 数组的一个数组项（item），各个项在数组中按值的大小从小到大有序地排列，并且数组中不包含任何重复项。

length 属性记录了整数集合包含的元素数量，也即是 contents 数组的长度。

虽然 intset 结构将 contents 属性声明为 int8_t 类型的数组，但实际上 contents 数组并不保存任何 int8_t 类型的值 —— contents 数组的真正类型取决于 encoding 属性的值：

如果 encoding 属性的值为 INTSET_ENC_INT16 ，那么 contents 就是一个 int16_t 类型的数组，数组里的每个项都是一个 int16_t 类型的整数值（最小值为 -32,768 ，最大值为 32,767 ）。
如果 encoding 属性的值为 INTSET_ENC_INT32 ，那么 contents 就是一个 int32_t 类型的数组，数组里的每个项都是一个 int32_t 类型的整数值（最小值为 -2,147,483,648 ，最大值为 2,147,483,647 ）。
如果 encoding 属性的值为 INTSET_ENC_INT64 ，那么 contents 就是一个 int64_t 类型的数组，数组里的每个项都是一个 int64_t 类型的整数值（最小值为 -9,223,372,036,854,775,808 ，最大值为 9,223,372,036,854,775,807 ）。
encoding 属性的值为 INTSET_ENC_INT16 ，表示整数集合的底层实现为 int16_t 类型的数组，而集合保存的都是 int16_t 类型的整数值。
length 属性的值为 5 ，表示整数集合包含五个元素。
contents 数组按从小到大的顺序保存着集合中的五个元素。
因为每个集合元素都是 int16_t 类型的整数值，所以 contents 数组的大小等于 sizeof(int16_t) * 5 = 16 * 5 = 80 位。

2015-09-13_55f51a1132a4b

encoding 属性的值为 INTSET_ENC_INT64 ，表示整数集合的底层实现为 int64_t 类型的数组，而数组中保存的都是 int64_t 类型的整数值。
length 属性的值为 4 ，表示整数集合包含四个元素。
contents 数组按从小到大的顺序保存着集合中的四个元素。
因为每个集合元素都是 int64_t 类型的整数值，所以 contents 数组的大小为 sizeof(int64_t) * 4 = 64 * 4 = 256 位。

2015-09-13_55f51a139a148

虽然 contents 数组保存的四个整数值中，只有 -2675256175807981027 是真正需要用 int64_t 类型来保存的，而其他的 1 、 3 、 5 三个值都可以用 int16_t 类型来保存，不过根据整数集合的升级规则，当向一个底层为 int16_t 数组的整数集合添加一个 int64_t 类型的整数值时，整数集合已有的所有元素都会被转换成 int64_t 类型，所以 contents 数组保存的四个整数值都是 int64_t 类型的，不仅仅是-2675256175807981027 。

升级
每当我们要将一个新元素添加到整数集合里面，并且新元素的类型比整数集合现有所有元素的类型都要长时，整数集合需要先进行升级（upgrade），然后才能将新元素添加到整数集合里面。

升级整数集合并添加新元素共分为三步进行：

根据新元素的类型，扩展整数集合底层数组的空间大小，并为新元素分配空间。
将底层数组现有的所有元素都转换成与新元素相同的类型，并将类型转换后的元素放置到正确的位上，而且在放置元素的过程中，需要继续维持底层数组的有序性质不变。
将新元素添加到底层数组里面。

举个例子，假设现在有一个 INTSET_ENC_INT16 编码的整数集合，集合中包含三个 int16_t 类型的元素，如图 6-3 所示。

因为每个元素都占用 16 位空间，所以整数集合底层数组的大小为 3 * 16 = 48 位，图 6-4 展示了整数集合的三个元素在这 48 位里的位置。

2015-09-13_55f51ab6dbe72

现在，假设我们要将类型为 int32_t 的整数值 65535 添加到整数集合里面，因为 65535 的类型 int32_t 比整数集合当前所有元素的类型都要长，所以在将 65535 添加到整数集合之前，程序需要先对整数集合进行升级。
升级首先要做的是，根据新类型的长度，以及集合元素的数量（包括要添加的新元素在内），对底层数组进行空间重分配。

整数集合目前有三个元素，再加上新元素 65535 ，整数集合需要分配四个元素的空间，因为每个 int32_t 整数值需要占用 32 位空间，所以在空间重分配之后，底层数组的大小将是 32 * 4 = 128 位，如图 6-5 所示

2015-09-13_55f51abfde71b

虽然程序对底层数组进行了空间重分配，但数组原有的三个元素 1 、 2 、 3 仍然是 int16_t 类型，这些元素还保存在数组的前 48 位里面，所以程序接下来要做的就是将这三个元素转换成 int32_t 类型，并将转换后的元素放置到正确的位上面，而且在放置元素的过程中，需要维持底层数组的有序性质不变。
首先，因为元素 3 在 1 、 2 、 3 、 65535 四个元素中排名第三，所以它将被移动到 contents 数组的索引 2 位置上，也即是数组 64 位至 95 位的空间内，如图 6-6 所示。

2015-09-13_55f51ac25b6a0

接着，因为元素 2 在 1 、 2 、 3 、 65535 四个元素中排名第二，所以它将被移动到 contents 数组的索引 1 位置上，也即是数组的 32位至 63 位的空间内，如图 6-7 所示。

2015-09-13_55f51ac353bfa

之后，因为元素 1 在 1 、 2 、 3 、 65535 四个元素中排名第一，所以它将被移动到 contents 数组的索引 0 位置上，也即是数组的 0 位至 31 位的空间内，如图 6-8 所示。
2015-09-13_55f51ac466154

然后，因为元素 65535 在 1 、 2 、 3 、 65535 四个元素中排名第四，所以它将被添加到 contents 数组的索引 3 位置上，也即是数组的96 位至 127 位的空间内，如图 6-9 所示。
2015-09-13_55f51acae92db

最后，程序将整数集合 encoding 属性的值从 INTSET_ENC_INT16 改为 INTSET_ENC_INT32 ，并将 length 属性的值从 3 改为 4 ，设置完成之后的整数集合如图 6-10 所示。
2015-09-13_55f51ae117896

因为每次向整数集合添加新元素都可能会引起升级，而每次升级都需要对底层数组中已有的所有元素进行类型转换，所以向整数集合添加新元素的时间复杂度为 O(N)

其他类型的升级操作，比如从 INTSET_ENC_INT16 编码升级为 INTSET_ENC_INT64 编码，或者从 INTSET_ENC_INT32 编码升级为 INTSET_ENC_INT64 编码，升级的过程都和上面展示的升级过程类似。

升级之后新元素的摆放位置

因为引发升级的新元素的长度总是比整数集合现有所有元素的长度都大，所以这个新元素的值要么就大于所有现有元素，要么就小于所有现有元素：

在新元素小于所有现有元素的情况下，新元素会被放置在底层数组的最开头（索引 0 ）；
在新元素大于所有现有元素的情况下，新元素会被放置在底层数组的最末尾（索引 length-1 ）。

降级
整数集合不支持降级操作，一旦对数组进行了升级，编码就会一直保持升级后的状态。

压缩列表

当一个列表键只包含少量列表项，并且每个列表项要么就是小整数值，要么就是长度比较短的字符串，那么 Redis 就会使用压缩列表来做列表键的底层实现。

redis> RPUSH lst 1 3 5 10086 "hello" "world"
(integer) 6

redis> OBJECT ENCODING lst
"ziplist"

另外，当一个哈希键只包含少量键值对，并且每个键值对的键和值要么就是小整数值，要么就是长度比较短的字符串，那么 Redis 就会使用压缩列表来做哈希键的底层实现。

举个例子，执行以下命令将创建一个压缩列表实现的哈希键：

redis> HMSET profile "name" "Jack" "age" 28 "job" "Programmer"
OK

redis> OBJECT ENCODING profile
"ziplist"

2015-09-13_55f51bfccbd83

列表 zlbytes 属性的值为 0x50 （十进制 80），表示压缩列表的总长为 80 字节。
列表 zltail 属性的值为 0x3c （十进制 60），这表示如果我们有一个指向压缩列表起始地址的指针 p ，那么只要用指针 p 加上偏移量 60 ，就可以计算出表尾节点 entry3 的地址。
列表 zllen 属性的值为 0x3 （十进制 3），表示压缩列表包含三个节点。

压缩列表节点的构成

每个压缩列表节点可以保存一个字节数组或者一个整数值，其中，字节数组可以是以下三种长度的其中一种：

长度小于等于 63 （2^{6}-1）字节的字节数组；
长度小于等于 16383 （2^{14}-1）字节的字节数组；
长度小于等于 4294967295 （2^{32}-1）字节的字节数组；

而整数值则可以是以下六种长度的其中一种：

4 位长，介于 0 至 12 之间的无符号整数；
1 字节长的有符号整数；
3 字节长的有符号整数；
int16_t 类型整数；
int32_t 类型整数；
int64_t 类型整数。

每个压缩列表节点都由 previous_entry_length 、 encoding 、 content 三个部分组成

previous_entry_length

节点的 previous_entry_length 属性以字节为单位，记录了压缩列表中前一个节点的长度。
previous_entry_length 属性的长度可以是 1 字节或者 5 字节：

如果前一节点的长度小于 254 字节，那么 previous_entry_length 属性的长度为 1 字节：前一节点的长度就保存在这一个字节里面。
如果前一节点的长度大于等于 254 字节，那么 previous_entry_length 属性的长度为 5 字节：其中属性的第一字节会被设置为 0xFE（十进制值 254），而之后的四个字节则用于保存前一节点的长度。

因为节点的 previous_entry_length 属性记录了前一个节点的长度，所以程序可以通过指针运算，根据当前节点的起始地址来计算出前一个节点的起始地址。

压缩列表的从表尾向表头遍历操作就是使用这一原理实现的：只要我们拥有了一个指向某个节点起始地址的指针，那么通过这个指针以及这个节点的 previous_entry_length 属性，程序就可以一直向前一个节点回溯，最终到达压缩列表的表头节点。
图 7-8 展示了一个从表尾节点向表头节点进行遍历的完整过程：

首先，我们拥有指向压缩列表表尾节点 entry4 起始地址的指针 p1 （指向表尾节点的指针可以通过指向压缩列表起始地址的指针加上zltail 属性的值得出）；
通过用 p1 减去 entry4 节点 previous_entry_length 属性的值，我们得到一个指向 entry4 前一节点 entry3 起始地址的指针 p2 ；
通过用 p2 减去 entry3 节点 previous_entry_length 属性的值，我们得到一个指向 entry3 前一节点 entry2 起始地址的指针 p3 ；
通过用 p3 减去 entry2 节点 previous_entry_length 属性的值，我们得到一个指向 entry2 前一节点 entry1 起始地址的指针 p4 ， entry1为压缩列表的表头节点；
最终，我们从表尾节点向表头节点遍历了整个列表。

encoding

节点的 encoding 属性记录了节点的 content 属性所保存数据的类型以及长度：

一字节、两字节或者五字节长，值的最高位为 00 、 01 或者 10 的是字节数组编码：这种编码表示节点的 content 属性保存着字节数组，数组的长度由编码除去最高两位之后的其他位记录；
一字节长，值的最高位以 11 开头的是整数编码：这种编码表示节点的 content 属性保存着整数值，整数值的类型和长度由编码除去最高两位之后的其他位记录；

content

编码的最高两位 00 表示节点保存的是一个字节数组；
编码的后六位 001011 记录了字节数组的长度 11 ；
content 属性保存着节点的值 “hello world” 。

编码 11000000 表示节点保存的是一个 int16_t 类型的整数值；
content 属性保存着节点的值 10086 。

连锁更新
https://www.kancloud.cn/kancloud/redisbook/63859

对象

typedef struct redisObject {

    // 类型
    unsigned type:4;

    // 编码
    unsigned encoding:4;

    // 指向底层实现数据结构的指针
    void *ptr;

    // ...

} robj;

类型

类型常量	对象的名称
REDIS_STRING	字符串对象
REDIS_LIST	列表对象
REDIS_HASH	哈希对象
REDIS_SET	集合对象
REDIS_ZSET	有序集合对象

# 键为字符串对象，值为字符串对象

redis> SET msg "hello world"
OK

redis> TYPE msg
string

# 键为字符串对象，值为列表对象

redis> RPUSH numbers 1 3 5
(integer) 6

redis> TYPE numbers
list

# 键为字符串对象，值为哈希对象

redis> HMSET profile name Tome age 25 career Programmer
OK

redis> TYPE profile
hash

# 键为字符串对象，值为集合对象

redis> SADD fruits apple banana cherry
(integer) 3

redis> TYPE fruits
set

# 键为字符串对象，值为有序集合对象

redis> ZADD price 8.5 apple 5.0 banana 6.0 cherry
(integer) 3

redis> TYPE price
zset

对象	对象 type 属性的值	TYPE 命令的输出
字符串对象	REDIS_STRING	“string”
列表对象	REDIS_LIST	“list”
哈希对象	REDIS_HASH	“hash”
集合对象	REDIS_SET	“set”
有序集合对象	REDIS_ZSET	“zset”

编码和底层实现

编码常量	编码所对应的底层数据结构
REDIS_ENCODING_INT	long 类型的整数
REDIS_ENCODING_EMBSTR	embstr 编码的简单动态字符串
REDIS_ENCODING_RAW	简单动态字符串
REDIS_ENCODING_HT	字典
REDIS_ENCODING_LINKEDLIST	双端链表
REDIS_ENCODING_ZIPLIST	压缩列表
REDIS_ENCODING_INTSET	整数集合
REDIS_ENCODING_SKIPLIST	跳跃表和字典

类型	编码	对象
REDIS_STRING	REDIS_ENCODING_INT	使用整数值实现的字符串对象。
REDIS_STRING	REDIS_ENCODING_EMBSTR	使用 embstr 编码的简单动态字符串实现的字符串对象。
REDIS_STRING	REDIS_ENCODING_RAW	使用简单动态字符串实现的字符串对象。
REDIS_LIST	REDIS_ENCODING_ZIPLIST	使用压缩列表实现的列表对象。
REDIS_LIST	REDIS_ENCODING_LINKEDLIST	使用双端链表实现的列表对象。
REDIS_HASH	REDIS_ENCODING_ZIPLIST	使用压缩列表实现的哈希对象。
REDIS_HASH	REDIS_ENCODING_HT	使用字典实现的哈希对象。
REDIS_SET	REDIS_ENCODING_INTSET	使用整数集合实现的集合对象。
REDIS_SET	REDIS_ENCODING_HT	使用字典实现的集合对象。
REDIS_ZSET	REDIS_ENCODING_ZIPLIST	使用压缩列表实现的有序集合对象。
REDIS_ZSET	REDIS_ENCODING_SKIPLIST	使用跳跃表和字典实现的有序集合对象。

redis> SET msg "hello wrold"
OK

redis> OBJECT ENCODING msg
"embstr"

redis> SET story "long long long long long long ago ..."
OK

redis> OBJECT ENCODING story
"raw"

redis> SADD numbers 1 3 5
(integer) 3

redis> OBJECT ENCODING numbers
"intset"

redis> SADD numbers "seven"
(integer) 1

redis> OBJECT ENCODING numbers
"hashtable"

对象所使用的底层数据结构	编码常量	OBJECT ENCODING 命令输出
整数	REDIS_ENCODING_INT	“int”
embstr 编码的简单动态字符串（SDS）	REDIS_ENCODING_EMBSTR	“embstr”
简单动态字符串	REDIS_ENCODING_RAW	“raw”
字典	REDIS_ENCODING_HT	“hashtable”
双端链表	REDIS_ENCODING_LINKEDLIST	“linkedlist”
压缩列表	REDIS_ENCODING_ZIPLIST	“ziplist”
整数集合	REDIS_ENCODING_INTSET	“intset”
跳跃表和字典	REDIS_ENCODING_SKIPLIST	“skiplist”

https://www.kancloud.cn/kancloud/redisbook/63862

内存回收
Redis 在自己的对象系统中构建了一个引用计数（reference counting）技术实现的内存回收机制

typedef struct redisObject {

    // ...

    // 引用计数
    int refcount;

    // ...

} robj;

在创建一个新对象时，引用计数的值会被初始化为 1 ；
当对象被一个新程序使用时，它的引用计数值会被增一；
当对象不再被一个程序使用时，它的引用计数值会被减一；
当对象的引用计数值变为 0 时，对象所占用的内存会被释放。

函数	作用
incrRefCount	将对象的引用计数值增一。
decrRefCount	将对象的引用计数值减一，当对象的引用计数值等于 0 时，释放对象。
resetRefCount	将对象的引用计数值设置为 0 ，但并不释放对象，这个函数通常在需要重新设置对象的引用计数值时使用。

对象共享
对象的引用计数属性还带有对象共享的作用

Redis 只对包含整数值的字符串对象进行共享。

对象的空转时长

除了前面介绍过的 type 、 encoding 、 ptr 和 refcount 四个属性之外， redisObject 结构包含的最后一个属性为 lru 属性，该属性记录了对象最后一次被命令程序访问的时间：

typedef struct redisObject {

    // ...

    unsigned lru:22;

    // ...

} robj;

单机数据库的实现

数据库

数据库键空间
Redis 是一个键值对（key-value pair）数据库服务器，服务器中的每个数据库都由一个 redis.h/redisDb 结构表示，其中， redisDb 结构的dict 字典保存了数据库中的所有键值对，我们将这个字典称为键空间（key space）：

typedef struct redisDb {

    // ...

    // 数据库键空间，保存着数据库中的所有键值对
    dict *dict;

    // ...

} redisDb;

添加新键
添加一个新键值对到数据库，实际上就是将一个新键值对添加到键空间字典里面，其中键为字符串对象，而值则为任意一种类型的 Redis 对象。

redis> SET message "hello world"
OK

redis> RPUSH alphabet "a" "b" "c"
(integer) 3

redis> HSET book name "Redis in Action"
(integer) 1

redis> HSET book author "Josiah L. Carlson"
(integer) 1

redis> HSET book publisher "Manning"
(integer) 1

1 2	redis> SET date "2013.12.1" OK

2015-09-13_55f5227cb91e3

删除键
删除数据库中的一个键，实际上就是在键空间里面删除键所对应的键值对对象。
更新键
对一个数据库键进行更新，实际上就是对键空间里面键所对应的值对象进行更新，根据值对象的类型不同，更新的具体方法也会有所不同。
对键取值

读写键空间时的维护操作
当使用 Redis 命令对数据库进行读写时，服务器不仅会对键空间执行指定的读写操作，还会执行一些额外的维护操作，其中包括：
在读取一个键之后（读操作和写操作都要对键进行读取），服务器会根据键是否存在，以此来更新服务器的键空间命中（hit）次数或键空间不命中（miss）次数，这两个值可以在 INFO stats 命令的 keyspace_hits 属性和 keyspace_misses 属性中查看。
在读取一个键之后，服务器会更新键的 LRU （最后一次使用）时间，这个值可以用于计算键的闲置时间，使用命令 OBJECT idletime 命令可以查看键 key 的闲置时间。
如果服务器在读取一个键时，发现该键已经过期，那么服务器会先删除这个过期键，然后才执行余下的其他操作，本章稍后对过期键的讨论会详细说明这一点。
如果有客户端使用 WATCH 命令监视了某个键，那么服务器在对被监视的键进行修改之后，会将这个键标记为脏（dirty），从而让事务程序注意到这个键已经被修改过，《事务》一章会详细说明这一点。
服务器每次修改一个键之后，都会对脏（dirty）键计数器的值增一，这个计数器会触发服务器的持久化以及复制操作执行，《RDB 持久化》、《AOF 持久化》和《复制》这三章都会说到这一点。
如果服务器开启了数据库通知功能，那么在对键进行修改之后，服务器将按配置发送相应的数据库通知，本章稍后讨论数据库通知功能的实现时会详细说明这一点。

Redis 服务器的所有数据库都保存在 redisServer.db 数组中，而数据库的数量则由 redisServer.dbnum 属性保存。
客户端通过修改目标数据库指针，让它指向 redisServer.db 数组中的不同元素来切换不同的数据库。
数据库主要由 dict 和 expires 两个字典构成，其中 dict 字典负责保存键值对，而 expires 字典则负责保存键的过期时间。
因为数据库由字典构成，所以对数据库的操作都是建立在字典操作之上的。
数据库的键总是一个字符串对象，而值则可以是任意一种 Redis 对象类型，包括字符串对象、哈希表对象、集合对象、列表对象和有序集合对象，分别对应字符串键、哈希表键、集合键、列表键和有序集合键。
expires 字典的键指向数据库中的某个键，而值则记录了数据库键的过期时间，过期时间是一个以毫秒为单位的 UNIX 时间戳。
Redis 使用惰性删除和定期删除两种策略来删除过期的键：惰性删除策略只在碰到过期键时才进行删除操作，定期删除策略则每隔一段时间，主动查找并删除过期键。
执行 SAVE 命令或者 BGSAVE 命令所产生的新 RDB 文件不会包含已经过期的键。
执行 BGREWRITEAOF 命令所产生的重写 AOF 文件不会包含已经过期的键。
当一个过期键被删除之后，服务器会追加一条 DEL 命令到现有 AOF 文件的末尾，显式地删除过期键。
当主服务器删除一个过期键之后，它会向所有从服务器发送一条 DEL 命令，显式地删除过期键。
从服务器即使发现过期键，也不会自作主张地删除它，而是等待主节点发来 DEL 命令，这种统一、中心化的过期键删除策略可以保证主从服务器数据的一致性。
当 Redis 命令对数据库进行修改之后，服务器会根据配置，向客户端发送数据库通知。

RDB 持久化

RDB 文件结构

RDB 文件的最开头是 REDIS 部分，这个部分的长度为 5 字节，保存着 “REDIS” 五个字符。通过这五个字符，程序可以在载入文件时，快速检查所载入的文件是否 RDB 文件。

db_version 长度为 4 字节，它的值是一个字符串表示的整数，这个整数记录了 RDB 文件的版本号，比如 “0006” 就代表 RDB 文件的版本为第六版。

databases 部分包含着零个或任意多个数据库，以及各个数据库中的键值对数据：

如果服务器的数据库状态为空（所有数据库都是空的），那么这个部分也为空，长度为 0 字节。
如果服务器的数据库状态为非空（有至少一个数据库非空），那么这个部分也为非空，根据数据库所保存键值对的数量、类型和内容不同，这个部分的长度也会有所不同。

EOF 常量的长度为 1 字节，这个常量标志着 RDB 文件正文内容的结束，当读入程序遇到这个值的时候，它知道所有数据库的所有键值对都已经载入完毕了

check_sum 是一个 8 字节长的无符号整数，保存着一个校验和，这个校验和是程序通过对 REDIS 、 db_version 、 databases 、 EOF 四个部分的内容进行计算得出的。服务器在载入 RDB 文件时，会将载入数据所计算出的校验和与 check_sum 所记录的校验和进行对比，以此来检查 RDB 文件是否有出错或者损坏的情况出现。

databases 部分
一个 RDB 文件的 databases 部分可以保存任意多个非空数据库。

每个非空数据库在 RDB 文件中都可以保存为 SELECTDB 、 db_number 、 key_value_pairs 三个部分

SELECTDB 常量的长度为 1 字节，当读入程序遇到这个值的时候，它知道接下来要读入的将是一个数据库号码。

db_number 保存着一个数据库号码，根据号码的大小不同，这个部分的长度可以是 1 字节、 2 字节或者 5 字节。当程序读入 db_number 部分之后，服务器会调用 SELECT 命令，根据读入的数据库号码进行数据库切换，使得之后读入的键值对可以载入到正确的数据库中。

key_value_pairs 部分保存了数据库中的所有键值对数据，如果键值对带有过期时间，那么过期时间也会和键值对保存在一起。根据键值对的数量、类型、内容、以及是否有过期时间等条件的不同， key_value_pairs 部分的长度也会有所不同。

key_value_pairs 部分
RDB 文件中的每个 key_value_pairs 部分都保存了一个或以上数量的键值对，如果键值对带有过期时间的话，那么键值对的过期时间也会被保存在内。

不带过期时间的键值对在 RDB 文件中对由 TYPE 、 key 、 value 三部分组成，如图 IMAGE_KEY_WITHOUT_EXPIRE_TIME 所示

REDIS_RDB_TYPE_STRING
REDIS_RDB_TYPE_LIST
REDIS_RDB_TYPE_SET
REDIS_RDB_TYPE_ZSET
REDIS_RDB_TYPE_HASH
REDIS_RDB_TYPE_LIST_ZIPLIST
REDIS_RDB_TYPE_SET_INTSET
REDIS_RDB_TYPE_ZSET_ZIPLIST
REDIS_RDB_TYPE_HASH_ZIPLIST

以上列出的每个 TYPE 常量都代表了一种对象类型或者底层编码，当服务器读入 RDB 文件中的键值对数据时，程序会根据 TYPE 的值来决定如何读入和解释 value 的数据。
key 和 value 分别保存了键值对的键对象和值对象：

其中 key 总是一个字符串对象，它的编码方式和 REDIS_RDB_TYPE_STRING 类型的 value 一样。根据内容长度的不同， key 的长度也会有所不同。
根据 TYPE 类型的不同，以及保存内容长度的不同，保存 value 的结构和长度也会有所不同，本节稍后会详细说明每种 TYPE 类型的value 结构保存方式。

https://www.kancloud.cn/kancloud/redisbook/63879

AOF 持久化

AOF 持久化功能的实现可以分为命令追加（append）、文件写入、文件同步（sync）三个步骤。

当 AOF 持久化功能处于打开状态时，服务器在执行完一个写命令之后，会以协议格式将被执行的写命令追加到服务器状态的 aof_buf 缓冲区的末尾：

struct redisServer {

    // ...

    // AOF 缓冲区
    sds aof_buf;

    // ...
};

AOF 文件的写入与同步
Redis 的服务器进程就是一个事件循环（loop），这个循环中的文件事件负责接收客户端的命令请求，以及向客户端发送命令回复，而时间事件则负责执行像 serverCron 函数这样需要定时运行的函数。

因为服务器在处理文件事件时可能会执行写命令，使得一些内容被追加到 aof_buf 缓冲区里面，所以在服务器每次结束一个事件循环之前，它都会调用 flushAppendOnlyFile 函数，考虑是否需要将 aof_buf 缓冲区中的内容写入和保存到 AOF 文件里面，这个过程可以用以下伪代码表示：

def eventLoop():

    while True:

        # 处理文件事件，接收命令请求以及发送命令回复
        # 处理命令请求时可能会有新内容被追加到 aof_buf 缓冲区中
        processFileEvents()

        # 处理时间事件
        processTimeEvents()

        # 考虑是否要将 aof_buf 中的内容写入和保存到 AOF 文件里面
        flushAppendOnlyFile()

appendfsync 选项的值	flushAppendOnlyFile 函数的行为
always	将 aof_buf 缓冲区中的所有内容写入并同步到 AOF 文件。
everysec	将 aof_buf 缓冲区中的所有内容写入到 AOF 文件，如果上次同步 AOF 文件的时间距离现在超过一秒钟，那么再次对 AOF 文件进行同步，并且这个同步操作是由一个线程专门负责执行的。
no	将 aof_buf 缓冲区中的所有内容写入到 AOF 文件，但并不对 AOF 文件进行同步，何时同步由操作系统来决定。

如果用户没有主动为 appendfsync 选项设置值，那么 appendfsync 选项的默认值为 everysec ，关于 appendfsync 选项的更多信息，请参考 Redis 项目附带的示例配置文件 redis.conf 。

为了提高文件的写入效率，在现代操作系统中，当用户调用 write 函数，将一些数据写入到文件的时候，操作系统通常会将写入数据暂时保存在一个内存缓冲区里面，等到缓冲区的空间被填满、或者超过了指定的时限之后，才真正地将缓冲区中的数据写入到磁盘里面。

这种做法虽然提高了效率，但也为写入数据带来了安全问题，因为如果计算机发生停机，那么保存在内存缓冲区里面的写入数据将会丢失。为此，系统提供了 fsync 和 fdatasync 两个同步函数，它们可以强制让操作系统立即将缓冲区中的数据写入到硬盘里面，从而确保写入数据的安全性。

如果这时 flushAppendOnlyFile 函数被调用，假设服务器当前 appendfsync 选项的值为 everysec ，并且根据 server.aof_last_fsync 属性显示，距离上次同步 AOF 文件已经超过一秒钟，那么服务器会先将 aof_buf 中的内容写入到 AOF 文件中，然后再对 AOF 文件进行同步。

当 appendfsync 的值为 always 时，服务器在每个事件循环都要将 aof_buf 缓冲区中的所有内容写入到 AOF 文件，并且同步 AOF 文件，所以 always 的效率是 appendfsync 选项三个值当中最慢的一个，但从安全性来说， always 也是最安全的，因为即使出现故障停机， AOF 持久化也只会丢失一个事件循环中所产生的命令数据。

当 appendfsync 的值为 everysec 时，服务器在每个事件循环都要将 aof_buf 缓冲区中的所有内容写入到 AOF 文件，并且每隔超过一秒就要在子线程中对 AOF 文件进行一次同步：从效率上来讲， everysec 模式足够快，并且就算出现故障停机，数据库也只丢失一秒钟的命令数据。

事件

Redis 基于 Reactor 模式开发了自己的网络事件处理器：这个处理器被称为文件事件处理器（file event handler）：

文件事件处理器使用 I/O 多路复用（multiplexing）程序来同时监听多个套接字，并根据套接字目前执行的任务来为套接字关联不同的事件处理器。
当被监听的套接字准备好执行连接应答（accept）、读取（read）、写入（write）、关闭（close）等操作时，与操作相对应的文件事件就会产生，这时文件事件处理器就会调用套接字之前关联好的事件处理器来处理这些事件。

虽然文件事件处理器以单线程方式运行，但通过使用 I/O 多路复用程序来监听多个套接字，文件事件处理器既实现了高性能的网络通信模型，又可以很好地与 Redis 服务器中其他同样以单线程方式运行的模块进行对接，这保持了 Redis 内部单线程设计的简单性。

文件事件处理器的构成
图 IMAGE_CONSTRUCT_OF_FILE_EVENT_HANDLER 展示了文件事件处理器的四个组成部分，它们分别是套接字、 I/O 多路复用程序、文件事件分派器（dispatcher）、以及事件处理器。

2015-09-13_55f524b50e6e6

文件事件是对套接字操作的抽象，每当一个套接字准备好执行连接应答（accept）、写入、读取、关闭等操作时，就会产生一个文件事件。因为一个服务器通常会连接多个套接字，所以多个文件事件有可能会并发地出现。

I/O 多路复用程序负责监听多个套接字，并向文件事件分派器传送那些产生了事件的套接字。

尽管多个文件事件可能会并发地出现，但 I/O 多路复用程序总是会将所有产生事件的套接字都入队到一个队列里面，然后通过这个队列，以有序（sequentially）、同步（synchronously）、每次一个套接字的方式向文件事件分派器传送套接字：当上一个套接字产生的事件被处理完毕之后（该套接字为事件所关联的事件处理器执行完毕）， I/O 多路复用程序才会继续向文件事件分派器传送下一个套接字，如图 IMAGE_DISPATCH_EVENT_VIA_QUEUE 。

2015-09-13_55f524bd8058c

文件事件分派器接收 I/O 多路复用程序传来的套接字，并根据套接字产生的事件的类型，调用相应的事件处理器。

服务器会为执行不同任务的套接字关联不同的事件处理器，这些处理器是一个个函数，它们定义了某个事件发生时，服务器应该执行的动作。

I/O 多路复用程序的实现
Redis 的 I/O 多路复用程序的所有功能都是通过包装常见的 select 、 epoll 、 evport 和 kqueue 这些 I/O 多路复用函数库来实现的，每个 I/O 多路复用函数库在 Redis 源码中都对应一个单独的文件，比如 ae_select.c 、 ae_epoll.c 、 ae_kqueue.c ，诸如此类。

2015-09-13_55f524bea64ce

Redis 在 I/O 多路复用程序的实现源码中用 #include 宏定义了相应的规则，程序会在编译时自动选择系统中性能最高的 I/O 多路复用函数库来作为 Redis 的 I/O 多路复用程序的底层实现：

https://www.kancloud.cn/kancloud/redisbook/63885

一次完整的客户端与服务器连接事件示例
让我们来追踪一次 Redis 客户端与服务器进行连接并发送命令的整个过程，看看在过程中会产生什么事件，而这些事件又是如何被处理的。

假设一个 Redis 服务器正在运作，那么这个服务器的监听套接字的 AE_READABLE 事件应该正处于监听状态之下，而该事件所对应的处理器为连接应答处理器。

如果这时有一个 Redis 客户端向服务器发起连接，那么监听套接字将产生 AE_READABLE 事件，触发连接应答处理器执行：处理器会对客户端的连接请求进行应答，然后创建客户端套接字，以及客户端状态，并将客户端套接字的 AE_READABLE 事件与命令请求处理器进行关联，使得客户端可以向主服务器发送命令请求。

之后，假设客户端向主服务器发送一个命令请求，那么客户端套接字将产生 AE_READABLE 事件，引发命令请求处理器执行，处理器读取客户端的命令内容，然后传给相关程序去执行。

执行命令将产生相应的命令回复，为了将这些命令回复传送回客户端，服务器会将客户端套接字的 AE_WRITABLE 事件与命令回复处理器进行关联：当客户端尝试读取命令回复的时候，客户端套接字将产生 AE_WRITABLE 事件，触发命令回复处理器执行，当命令回复处理器将命令回复全部写入到套接字之后，服务器就会解除客户端套接字的 AE_WRITABLE 事件与命令回复处理器之间的关联。

图 IMAGE_COMMAND_PROGRESS 总结了上面描述的整个通讯过程，以及通讯时用到的事件处理器。
2015-09-13_55f524c5218e4

客户端属性
客户端状态包含的属性可以分为两类：

一类是比较通用的属性，这些属性很少与特定功能相关，无论客户端执行的是什么工作，它们都要用到这些属性。
另外一类是和特定功能相关的属性，比如操作数据库时需要用到的 db 属性和 dictid 属性，执行事务时需要用到的 mstate 属性，以及执行 WATCH 命令时需要用到的 watched_keys 属性，等等。

套接字描述符

typedef struct redisClient {

    // ...

    int fd;

    // ...

} redisClient;

根据客户端类型的不同， fd 属性的值可以是 -1 或者是大于 -1 的整数：
伪客户端（fake client）的 fd 属性的值为 -1 ：伪客户端处理的命令请求来源于 AOF 文件或者 Lua 脚本，而不是网络，所以这种客户端不需要套接字连接，自然也不需要记录套接字描述符。目前 Redis 服务器会在两个地方用到伪客户端，一个用于载入 AOF 文件并还原数据库状态，而另一个则用于执行 Lua 脚本中包含的 Redis 命令。

普通客户端的 fd 属性的值为大于 -1 的整数：普通客户端使用套接字来与服务器进行通讯，所以服务器会用 fd 属性来记录客户端套接字的描述符。因为合法的套接字描述符不能是 -1 ，所以普通客户端的套接字描述符的值必然是大于 -1 的整数。

执行 CLIENT_LIST 命令可以列出目前所有连接到服务器的普通客户端，命令输出中的 fd 域显示了服务器连接客户端所使用的套接字描述符：

redis> CLIENT list

addr=127.0.0.1:53428 fd=6 name= age=1242 idle=0 ...
addr=127.0.0.1:53469 fd=7 name= age=4 idle=4 ...

使用 CLIENT_SETNAME 命令可以为客户端设置一个名字，让客户端的身份变得更清晰。

typedef struct redisClient {

    // ...

    robj *name;

    // ...

} redisClient;

标志
客户端的标志属性 flags 记录了客户端的角色（role），以及客户端目前所处的状态：

typedef struct redisClient {

    // ...

    int flags;

    // ...

} redisClient;

可以是多个标志的二进制或，比如：

flags = <flag1> | <flag2> | ...

# 客户端是一个主服务器
REDIS_MASTER

# 客户端正在被列表命令阻塞
REDIS_BLOCKED

# 客户端正在执行事务，但事务的安全性已被破坏
REDIS_MULTI | REDIS_DIRTY_CAS

# 客户端是一个从服务器，并且版本低于 Redis 2.8
REDIS_SLAVE | REDIS_PRE_PSYNC

# 这是专门用于执行 Lua 脚本包含的 Redis 命令的伪客户端
# 它强制服务器将当前执行的命令写入 AOF 文件，并复制给从服务器
REDIS_LUA_CLIENT | REDIS_FORCE_AOF | REDIS_FORCE_REPL

输入缓冲区

typedef struct redisClient {

    // ...

    sds querybuf;

    // ...

} redisClient;

输入缓冲区的大小会根据输入内容动态地缩小或者扩大，但它的最大大小不能超过 1 GB ，否则服务器将关闭这个客户端。

命令与命令参数

typedef struct redisClient {

    // ...

    robj **argv;

    int argc;

    // ...

} redisClient;

身份验证

typedef struct redisClient {

    // ...

    int authenticated;

    // ...

} redisClient;

如果 authenticated 的值为 0 ，那么表示客户端未通过身份验证；如果 authenticated 的值为 1 ，那么表示客户端已经通过了身份验证。

# authenticated 属性的值从 0 变为 1
redis> AUTH 123321
OK

redis> PING
PONG

redis> SET msg "hello world"
OK

时间

typedef struct redisClient {

    // ...

    time_t ctime;

    time_t lastinteraction;

    time_t obuf_soft_limit_reached_time;

    // ...

} redisClient;

服务端
命令请求的执行过程

那么从客户端发送 SET KEY VALUE 命令到获得回复 OK 期间，客户端和服务器共需要执行以下操作：

客户端向服务器发送命令请求 SET KEY VALUE 。
服务器接收并处理客户端发来的命令请求 SET KEY VALUE ，在数据库中进行设置操作，并产生命令回复 OK 。
服务器将命令回复 OK 发送给客户端。
客户端接收服务器返回的命令回复 OK ，并将这个回复打印给用户观看。

发送命令请求
Redis 服务器的命令请求来自 Redis 客户端，当用户在客户端中键入一个命令请求时，客户端会将这个命令请求转换成协议格式，然后通过连接到服务器的套接字，将协议格式的命令请求发送给服务器

读取命令请求

当客户端与服务器之间的连接套接字因为客户端的写入而变得可读时，服务器将调用命令请求处理器来执行以下操作：

读取套接字中协议格式的命令请求，并将其保存到客户端状态的输入缓冲区里面。
对输入缓冲区中的命令请求进行分析，提取出命令请求中包含的命令参数，以及命令参数的个数，然后分别将参数和参数个数保存到客户端状态的 argv 属性和 argc 属性里面。
调用命令执行器，执行客户端指定的命令。

https://www.kancloud.cn/kancloud/redisbook/63892

多机数据库的实现

旧版复制功能的实现

Redis 的复制功能分为同步（sync）和命令传播（command propagate）两个操作：

同步操作用于将从服务器的数据库状态更新至主服务器当前所处的数据库状态。
而命令传播操作则用于在主服务器的数据库状态被修改，导致主从服务器的数据库状态出现不一致时，让主从服务器的数据库重新回到一致状态。
同步
当客户端向从服务器发送 SLAVEOF 命令，要求从服务器复制主服务器时，从服务器首先需要执行同步操作，也即是，将从服务器的数据库状态更新至主服务器当前所处的数据库状态。

从服务器对主服务器的同步操作需要通过向主服务器发送 SYNC 命令来完成，以下是 SYNC 命令的执行步骤：

从服务器向主服务器发送 SYNC 命令。
收到 SYNC 命令的主服务器执行 BGSAVE 命令，在后台生成一个 RDB 文件，并使用一个缓冲区记录从现在开始执行的所有写命令。
当主服务器的 BGSAVE 命令执行完毕时，主服务器会将 BGSAVE 命令生成的 RDB 文件发送给从服务器，从服务器接收并载入这个 RDB 文件，将自己的数据库状态更新至主服务器执行 BGSAVE 命令时的数据库状态。
主服务器将记录在缓冲区里面的所有写命令发送给从服务器，从服务器执行这些写命令，将自己的数据库状态更新至主服务器数据库当前所处的状态。

2015-09-13_55f5274336096

时间	主服务器	从服务器
T0	服务器启动。	服务器启动。
T1	执行 SET k1 v1 。
T2	执行 SET k2 v2 。
T3	执行 SET k3 v3 。
T4		向主服务器发送 SYNC 命令。
T5	接收到从服务器发来的 SYNC 命令，执行 BGSAVE 命令，创建包含键 k1 、 k2 、 k3 的 RDB 文件，并使用缓冲区记录接下来执行的所有写命令。
T6	执行 SET k4 v4 ，并将这个命令记录到缓冲区里面。
T7	执行 SET k5 v5 ，并将这个命令记录到缓冲区里面。
T8	BGSAVE 命令执行完毕，向从服务器发送 RDB 文件。
T9		接收并载入主服务器发来的 RDB 文件，获得 k1 、 k2 、 k3 三个键。
T10	向从服务器发送缓冲区中保存的写命令 SET k4 v4 和 SET k5v5 。
T11		接收并执行主服务器发来的两个 SET 命令，得到 k4 和 k5 两个键。
T12	同步完成，现在主从服务器两者的数据库都包含了键 k1 、k2 、 k3 、 k4 和 k5 。	同步完成，现在主从服务器两者的数据库都包含了键 k1 、 k2 、k3 、 k4 和 k5 。

命令传播
在同步操作执行完毕之后，主从服务器两者的数据库将达到一致状态，但这种一致并不是一成不变的 —— 每当主服务器执行客户端发送的写命令时，主服务器的数据库就有可能会被修改，并导致主从服务器状态不再一致。

举个例子，假设一个主服务器和一个从服务器刚刚完成同步操作，它们的数据库都保存了相同的五个键 k1 至 k5 ，如图 IMAGE_CONSISTENT 所示。

2015-09-13_55f52746e0111

如果这时，客户端向主服务器发送命令 DEL k3 ，那么主服务器在执行完这个 DEL 命令之后，主从服务器的数据库将出现不一致：主服务器的数据库已经不再包含键 k3 ，但这个键却仍然包含在从服务器的数据库里面，如图 IMAGE_INCONSISTENT 所示。

2015-09-13_55f527488c8ad

为了让主从服务器再次回到一致状态，主服务器需要对从服务器执行命令传播操作：主服务器会将自己执行的写命令 —— 也即是造成主从服务器不一致的那条写命令 —— 发送给从服务器执行，当从服务器执行了相同的写命令之后，主从服务器将再次回到一致状态。

在上面的例子中，主服务器因为执行了命令 DEL k3 而导致主从服务器不一致，所以主服务器将向从服务器发送相同的命令 DEL k3 ：当从服务器执行完这个命令之后，主从服务器将再次回到一致状态 —— 现在主从服务器两者的数据库都不再包含键 k3 了，如图 IMAGE_PROPAGATE_DEL_k3 所示。

2015-09-13_55f5274a0ca0f

Sentinel

启动并初始化 Sentinel
启动一个 Sentinel 可以使用命令：
$ redis-sentinel /path/to/your/sentinel.conf
或者命令：
$ redis-server /path/to/your/sentinel.conf --sentinel
这两个命令的效果完全相同。

当一个 Sentinel 启动时，它需要执行以下步骤：

初始化服务器。
将普通 Redis 服务器使用的代码替换成 Sentinel 专用代码。
初始化 Sentinel 状态。
根据给定的配置文件，初始化 Sentinel 的监视主服务器列表。
创建连向主服务器的网络连接。

初始化服务器
因为 Sentinel 本质上只是一个运行在特殊模式下的 Redis 服务器，所以启动 Sentinel 的第一步，就是初始化一个普通的 Redis 服务器

功能	使用情况
数据库和键值对方面的命令，比如 SET 、 DEL 、FLUSHDB 。	不使用。
事务命令，比如 MULTI 和 WATCH 。	不使用。
脚本命令，比如 EVAL 。	不使用。
RDB 持久化命令，比如 SAVE 和 BGSAVE 。	不使用。
AOF 持久化命令，比如 BGREWRITEAOF 。	不使用。
复制命令，比如 SLAVEOF 。	Sentinel 内部可以使用，但客户端不可以使用。
发布与订阅命令，比如 PUBLISH 和 SUBSCRIBE 。	SUBSCRIBE 、 PSUBSCRIBE 、 UNSUBSCRIBE PUNSUBSCRIBE 四个命令在 Sentinel 内部和客户端都可以使用，但 PUBLISH 命令只能在 Sentinel 内部使用。
文件事件处理器（负责发送命令请求、处理命令回复）。	Sentinel 内部使用，但关联的文件事件处理器和普通 Redis 服务器不同。
时间事件处理器（负责执行 serverCron 函数）。	Sentinel 内部使用，时间事件的处理器仍然是 serverCron 函数， serverCron函数会调用 sentinel.c/sentinelTimer 函数，后者包含了 Sentinel 要执行的所有操作。

使用 Sentinel 专用代码
启动 Sentinel 的第二个步骤就是将一部分普通 Redis 服务器使用的代码替换成 Sentinel 专用代码。

比如说，普通 Redis 服务器使用 redis.h/REDIS_SERVERPORT 常量的值作为服务器端口：
#define REDIS_SERVERPORT 6379
而 Sentinel 则使用 sentinel.c/REDIS_SENTINEL_PORT 常量的值作为服务器端口：
#define REDIS_SENTINEL_PORT 26379

除此之外，普通 Redis 服务器使用 redis.c/redisCommandTable 作为服务器的命令表：

struct redisCommand redisCommandTable[] = {
    {"get",getCommand,2,"r",0,NULL,1,1,1,0,0},
    {"set",setCommand,-3,"wm",0,noPreloadGetKeys,1,1,1,0,0},
    {"setnx",setnxCommand,3,"wm",0,noPreloadGetKeys,1,1,1,0,0},
    // ...
    {"script",scriptCommand,-2,"ras",0,NULL,0,0,0,0,0},
    {"time",timeCommand,1,"rR",0,NULL,0,0,0,0,0},
    {"bitop",bitopCommand,-4,"wm",0,NULL,2,-1,1,0,0},
    {"bitcount",bitcountCommand,-2,"r",0,NULL,1,1,1,0,0}
}

初始化 Sentinel 状态
在应用了 Sentinel 的专用代码之后，接下来，服务器会初始化一个 sentinel.c/sentinelState 结构（后面简称“Sentinel 状态”），这个结构保存了服务器中所有和 Sentinel 功能有关的状态（服务器的一般状态仍然由 redis.h/redisServer 结构保存）：

struct sentinelState {

    // 当前纪元，用于实现故障转移
    uint64_t current_epoch;

    // 保存了所有被这个 sentinel 监视的主服务器
    // 字典的键是主服务器的名字
    // 字典的值则是一个指向 sentinelRedisInstance 结构的指针
    dict *masters;

    // 是否进入了 TILT 模式？
    int tilt;

    // 目前正在执行的脚本的数量
    int running_scripts;

    // 进入 TILT 模式的时间
    mstime_t tilt_start_time;

    // 最后一次执行时间处理器的时间
    mstime_t previous_time;

    // 一个 FIFO 队列，包含了所有需要执行的用户脚本
    list *scripts_queue;

} sentinel;

初始化 Sentinel 状态的 masters 属性
Sentinel 状态中的 masters 字典记录了所有被 Sentinel 监视的主服务器的相关信息，其中：
字典的键是被监视主服务器的名字。
而字典的值则是被监视主服务器对应的 sentinel.c/sentinelRedisInstance 结构。
每个 sentinelRedisInstance 结构（后面简称“实例结构”）代表一个被 Sentinel 监视的 Redis 服务器实例（instance），这个实例可以是主服务器、从服务器、或者另外一个 Sentinel 。
实例结构包含的属性非常多，以下代码展示了实例结构在表示主服务器时使用的其中一部分属性，本章接下来将逐步对实例结构中的各个属性进行介绍：

typedef struct sentinelRedisInstance {

    // 标识值，记录了实例的类型，以及该实例的当前状态
    int flags;

    // 实例的名字
    // 主服务器的名字由用户在配置文件中设置
    // 从服务器以及 Sentinel 的名字由 Sentinel 自动设置
    // 格式为 ip:port ，例如 "127.0.0.1:26379"
    char *name;

    // 实例的运行 ID
    char *runid;

    // 配置纪元，用于实现故障转移
    uint64_t config_epoch;

    // 实例的地址
    sentinelAddr *addr;

    // SENTINEL down-after-milliseconds 选项设定的值
    // 实例无响应多少毫秒之后才会被判断为主观下线（subjectively down）
    mstime_t down_after_period;

    // SENTINEL monitor <master-name> <IP> <port> <quorum> 选项中的 quorum 参数
    // 判断这个实例为客观下线（objectively down）所需的支持投票数量
    int quorum;

    // SENTINEL parallel-syncs <master-name> <number> 选项的值
    // 在执行故障转移操作时，可以同时对新的主服务器进行同步的从服务器数量
    int parallel_syncs;

    // SENTINEL failover-timeout <master-name> <ms> 选项的值
    // 刷新故障迁移状态的最大时限
    mstime_t failover_timeout;

    // ...

} sentinelRedisInstance;

sentinelRedisInstance.addr 属性是一个指向 sentinel.c/sentinelAddr 结构的指针，这个结构保存着实例的 IP 地址和端口号：

typedef struct sentinelAddr {

    char *ip;

    int port;

} sentinelAddr;

对 Sentinel 状态的初始化将引发对 masters 字典的初始化，而 masters 字典的初始化是根据被载入的 Sentinel 配置文件来进行的。

创建连向主服务器的网络连接
初始化 Sentinel 的最后一步是创建连向被监视主服务器的网络连接： Sentinel 将成为主服务器的客户端，它可以向主服务器发送命令，并从命令回复中获取相关的信息。

对于每个被 Sentinel 监视的主服务器来说， Sentinel 会创建两个连向主服务器的异步网络连接：

一个是命令连接，这个连接专门用于向主服务器发送命令，并接收命令回复。
另一个是订阅连接，这个连接专门用于订阅主服务器的 sentinel:hello 频道。
为什么有两个连接？

在 Redis 目前的发布与订阅功能中，被发送的信息都不会保存在 Redis 服务器里面，如果在信息发送时，想要接收信息的客户端不在线或者断线，那么这个客户端就会丢失这条信息。

因此，为了不丢失 sentinel:hello 频道的任何信息， Sentinel 必须专门用一个订阅连接来接收该频道的信息。
而另一方面，除了订阅频道之外， Sentinel 还又必须向主服务器发送命令，以此来与主服务器进行通讯，所以 Sentinel 还必须向主服务器创建命令连接。

并且因为 Sentinel 需要与多个实例创建多个网络连接，所以 Sentinel 使用的是异步连接。

图 IMAGE_SENTINEL_CONNECT_SERVER 展示了一个 Sentinel 向被它监视的两个主服务器 master1 和 master2 创建命令连接和订阅连接的例子。

2015-09-13_55f5282219b43

Sentinel 只是一个运行在特殊模式下的 Redis 服务器，它使用了和普通模式不同的命令表，所以 Sentinel 模式能够使用的命令和普通 Redis 服务器能够使用的命令不同。
Sentinel 会读入用户指定的配置文件，为每个要被监视的主服务器创建相应的实例结构，并创建连向主服务器的命令连接和订阅连接，其中命令连接用于向主服务器发送命令请求，而订阅连接则用于接收指定频道的消息。
Sentinel 通过向主服务器发送 INFO 命令来获得主服务器属下所有从服务器的地址信息，并为这些从服务器创建相应的实例结构，以及连向这些从服务器的命令连接和订阅连接。
在一般情况下， Sentinel 以每十秒一次的频率向被监视的主服务器和从服务器发送 INFO 命令，当主服务器处于下线状态，或者 Sentinel 正在对主服务器进行故障转移操作时， Sentinel 向从服务器发送 INFO 命令的频率会改为每秒一次。
对于监视同一个主服务器和从服务器的多个 Sentinel 来说，它们会以每两秒一次的频率，通过向被监视服务器的 sentinel:hello频道发送消息来向其他 Sentinel 宣告自己的存在。
每个 Sentinel 也会从 sentinel:hello 频道中接收其他 Sentinel 发来的信息，并根据这些信息为其他 Sentinel 创建相应的实例结构，以及命令连接。
Sentinel 只会与主服务器和从服务器创建命令连接和订阅连接， Sentinel 与 Sentinel 之间则只创建命令连接。
Sentinel 以每秒一次的频率向实例（包括主服务器、从服务器、其他 Sentinel）发送 PING 命令，并根据实例对 PING 命令的回复来判断实例是否在线：当一个实例在指定的时长中连续向 Sentinel 发送无效回复时， Sentinel 会将这个实例判断为主观下线。
当 Sentinel 将一个主服务器判断为主观下线时，它会向同样监视这个主服务器的其他 Sentinel 进行询问，看它们是否同意这个主服务器已经进入主观下线状态。
当 Sentinel 收集到足够多的主观下线投票之后，它会将主服务器判断为客观下线，并发起一次针对主服务器的故障转移操作。

Sentinel 系统选举领头 Sentinel 的方法是对 Raft 算法的领头选举方法的实现

集群

一个 Redis 集群通常由多个节点（node）组成，在刚开始的时候，每个节点都是相互独立的，它们都处于一个只包含自己的集群当中，要组建一个真正可工作的集群，我们必须将各个独立的节点连接起来，构成一个包含多个节点的集群。

连接各个节点的工作可以使用 CLUSTER MEET 命令来完成，该命令的格式如下：

CLUSTER MEET <ip> <port>

向一个节点 node 发送 CLUSTER MEET 命令，可以让 node 节点与 ip 和 port 所指定的节点进行握手（handshake），当握手成功时， node节点就会将 ip 和 port 所指定的节点添加到 node 节点当前所在的集群中。
举个例子，假设现在有三个独立的节点 127.0.0.1:7000 、 127.0.0.1:7001 、 127.0.0.1:7002 （下文省略 IP 地址，直接使用端口号来区分各个节点），我们首先使用客户端连上节点 7000 ，通过发送 CLUSTER NODE 命令可以看到，集群目前只包含 7000 自己一个节点：

1
2
3

$ redis-cli -c -p 7000
127.0.0.1:7000> CLUSTER NODES
51549e625cfda318ad27423a31e7476fe3cd2939 :0 myself,master - 0 0 0 connected

通过向节点 7000 发送以下命令，我们可以将节点 7001 添加到节点 7000 所在的集群里面：

127.0.0.1:7000> CLUSTER MEET 127.0.0.1 7001
OK

127.0.0.1:7000> CLUSTER NODES
68eef66df23420a5862208ef5b1a7005b806f2ff 127.0.0.1:7001 master - 0 1388204746210 0 connected
51549e625cfda318ad27423a31e7476fe3cd2939 :0 myself,master - 0 0 0 connected

继续向节点 7000 发送以下命令，我们可以将节点 7002 也添加到节点 7000 和节点 7001 所在的集群里面：

127.0.0.1:7000> CLUSTER MEET 127.0.0.1 7002
OK

127.0.0.1:7000> CLUSTER NODES
68eef66df23420a5862208ef5b1a7005b806f2ff 127.0.0.1:7001 master - 0 1388204848376 0 connected
9dfb4c4e016e627d9769e4c9bb0d4fa208e65c26 127.0.0.1:7002 master - 0 1388204847977 0 connected
51549e625cfda318ad27423a31e7476fe3cd2939 :0 myself,master - 0 0 0 connected

现在，这个集群里面包含了 7000 、 7001 和 7002 三个节点，图 IMAGE_CONNECT_NODES_1 至 IMAGE_CONNECT_NODES_5 展示了这三个节点进行握手的整个过程。

2015-09-13_55f52896e231a

2015-09-13_55f52898c2d57

2015-09-13_55f52899dbdb6

2015-09-13_55f5289b0e82b

启动节点
一个节点就是一个运行在集群模式下的 Redis 服务器， Redis 服务器在启动时会根据 cluster-enabled 配置选项的是否为 yes 来决定是否开启服务器的集群模式

节点（运行在集群模式下的 Redis 服务器）会继续使用所有在单机模式中使用的服务器组件，比如说：

节点会继续使用文件事件处理器来处理命令请求和返回命令回复。
节点会继续使用时间事件处理器来执行 serverCron 函数，而 serverCron 函数又会调用集群模式特有的 clusterCron 函数： clusterCron函数负责执行在集群模式下需要执行的常规操作，比如向集群中的其他节点发送 Gossip 消息，检查节点是否断线；又或者检查是否需要对下线节点进行自动故障转移，等等。
节点会继续使用数据库来保存键值对数据，键值对依然会是各种不同类型的对象。
节点会继续使用 RDB 持久化模块和 AOF 持久化模块来执行持久化工作。
节点会继续使用发布与订阅模块来执行 PUBLISH 、 SUBSCRIBE 等命令。
节点会继续使用复制模块来进行节点的复制工作。
节点会继续使用 Lua 脚本环境来执行客户端输入的 Lua 脚本。
集群数据结构
clusterNode 结构保存了一个节点的当前状态，比如节点的创建时间，节点的名字，节点当前的配置纪元，节点的 IP 和地址，等等。

每个节点都会使用一个 clusterNode 结构来记录自己的状态，并为集群中的所有其他节点（包括主节点和从节点）都创建一个相应的clusterNode 结构，以此来记录其他节点的状态：

struct clusterNode {

    // 创建节点的时间
    mstime_t ctime;

    // 节点的名字，由 40 个十六进制字符组成
    // 例如 68eef66df23420a5862208ef5b1a7005b806f2ff
    char name[REDIS_CLUSTER_NAMELEN];

    // 节点标识
    // 使用各种不同的标识值记录节点的角色（比如主节点或者从节点），
    // 以及节点目前所处的状态（比如在线或者下线）。
    int flags;

    // 节点当前的配置纪元，用于实现故障转移
    uint64_t configEpoch;

    // 节点的 IP 地址
    char ip[REDIS_IP_STR_LEN];

    // 节点的端口号
    int port;

    // 保存连接节点所需的有关信息
    clusterLink *link;

    // ...

};

clusterNode 结构的 link 属性是一个 clusterLink 结构，该结构保存了连接节点所需的有关信息，比如套接字描述符，输入缓冲区和输出缓冲区：

typedef struct clusterLink {

    // 连接的创建时间
    mstime_t ctime;

    // TCP 套接字描述符
    int fd;

    // 输出缓冲区，保存着等待发送给其他节点的消息（message）。
    sds sndbuf;

    // 输入缓冲区，保存着从其他节点接收到的消息。
    sds rcvbuf;

    // 与这个连接相关联的节点，如果没有的话就为 NULL
    struct clusterNode *node;

} clusterLink;

redisClient 结构和 clusterLink 结构的相同和不同之处

redisClient 结构和 clusterLink 结构都有自己的套接字描述符和输入、输出缓冲区，这两个结构的区别在于， redisClient 结构中的套接字和缓冲区是用于连接客户端的，而 clusterLink 结构中的套接字和缓冲区则是用于连接节点的。

最后，每个节点都保存着一个 clusterState 结构，这个结构记录了在当前节点的视角下，集群目前所处的状态 —— 比如集群是在线还是下线，集群包含多少个节点，集群当前的配置纪元，诸如此类：

typedef struct clusterState {

    // 指向当前节点的指针
    clusterNode *myself;

    // 集群当前的配置纪元，用于实现故障转移
    uint64_t currentEpoch;

    // 集群当前的状态：是在线还是下线
    int state;

    // 集群中至少处理着一个槽的节点的数量
    int size;

    // 集群节点名单（包括 myself 节点）
    // 字典的键为节点的名字，字典的值为节点对应的 clusterNode 结构
    dict *nodes;

    // ...

} clusterState;

以前面介绍的 7000 、 7001 、 7002 三个节点为例，图 IMAGE_CLUSTER_STATE_OF_7000 展示了节点 7000 创建的 clusterState 结构，这个结构从节点 7000 的角度记录了集群、以及集群包含的三个节点的当前状态（为了空间考虑，图中省略了 clusterNode 结构的一部分属性）：

结构的 currentEpoch 属性的值为 0 ，表示集群当前的配置纪元为 0 。
结构的 size 属性的值为 0 ，表示集群目前没有任何节点在处理槽：因此结构的 state 属性的值为 REDIS_CLUSTER_FAIL —— 这表示集群目前处于下线状态。
结构的 nodes 字典记录了集群目前包含的三个节点，这三个节点分别由三个 clusterNode 结构表示：其中 myself 指针指向代表节点 7000 的 clusterNode 结构，而字典中的另外两个指针则分别指向代表节点 7001 和代表节点 7002 的 clusterNode 结构，这两个节点是节点 7000 已知的在集群中的其他节点。
三个节点的 clusterNode 结构的 flags 属性都是 REDIS_NODE_MASTER ，说明三个节点都是主节点。

节点 7001 和节点 7002 也会创建类似的 clusterState 结构：

不过在节点 7001 创建的 clusterState 结构中， myself 指针将指向代表节点 7001 的 clusterNode 结构，而节点 7000 和节点 7002 则是集群中的其他节点。
而在节点 7002 创建的 clusterState 结构中， myself 指针将指向代表节点 7002 的 clusterNode 结构，而节点 7000 和节点 7001 则是集群中的其他节点。
CLUSTER MEET 命令的实现
通过向节点 A 发送 CLUSTER MEET 命令，客户端可以让接收命令的节点 A 将另一个节点 B 添加到节点 A 当前所在的集群里面：

CLUSTER MEET <ip> <port>

收到命令的节点 A 将与节点 B 进行握手（handshake），以此来确认彼此的存在，并为将来的进一步通信打好基础：

节点 A 会为节点 B 创建一个 clusterNode 结构，并将该结构添加到自己的 clusterState.nodes 字典里面。
之后，节点 A 将根据 CLUSTER MEET 命令给定的 IP 地址和端口号，向节点 B 发送一条 MEET 消息（message）。
如果一切顺利，节点 B 将接收到节点 A 发送的 MEET 消息，节点 B 会为节点 A 创建一个 clusterNode 结构，并将该结构添加到自己的 clusterState.nodes 字典里面。
之后，节点 B 将向节点 A 返回一条 PONG 消息。
如果一切顺利，节点 A 将接收到节点 B 返回的 PONG 消息，通过这条 PONG 消息节点 A 可以知道节点 B 已经成功地接收到了自己发送的 MEET 消息。
之后，节点 A 将向节点 B 返回一条 PING 消息。
如果一切顺利，节点 B 将接收到节点 A 返回的 PING 消息，通过这条 PING 消息节点 B 可以知道节点 A 已经成功地接收到了自己返回的 PONG 消息，握手完成。

2015-09-13_55f528a75c69b

之后，节点 A 会将节点 B 的信息通过 Gossip 协议传播给集群中的其他节点，让其他节点也与节点 B 进行握手，最终，经过一段时间之后，节点 B 会被集群中的所有节点认识。

独立功能的实现

https://www.kancloud.cn/kancloud/redisbook/63905

NIO

Posted on 2017-09-04 Edited on 2022-01-28 In 基础

阻塞

假设有一个管道，进程A为管道的写入方，Ｂ为管道的读出方。

假设一开始内核缓冲区是空的，B作为读出方，被阻塞着。然后首先A往管道写入，这时候内核缓冲区由空的状态变到非空状态，内核就会产生一个事件告诉Ｂ该醒来了，这个事件姑且称之为“缓冲区非空”。

但是“缓冲区非空”事件通知B后，B却还没有读出数据；且内核许诺了不能把写入管道中的数据丢掉这个时候，Ａ写入的数据会滞留在内核缓冲区中，如果内核也缓冲区满了，B仍未开始读数据，最终内核缓冲区会被填满，这个时候会产生一个I/O事件，告诉进程A，你该等等（阻塞）了，我们把这个事件定义为“缓冲区满”。

假设后来Ｂ终于开始读数据了，于是内核的缓冲区空了出来，这时候内核会告诉A，内核缓冲区有空位了，你可以从长眠中醒来了，继续写数据了，我们把这个事件叫做“缓冲区非满”

也许事件Y1已经通知了A，但是A也没有数据写入了，而Ｂ继续读出数据，知道内核缓冲区空了。这个时候内核就告诉B，你需要阻塞了！，我们把这个时间定为“缓冲区空”。

阻塞I/O模式下，一个线程只能处理一个流的I/O事件。如果想要同时处理多个流，要么多进程(fork)，要么多线程(pthread_create)

非阻塞忙轮询的I/O方式

while true {
    for i in stream[]; {
        if i has data
            read until unavailable
    }
}

如果所有的流都没有数据，那么只会白白浪费CPU

非阻塞无差别轮询的I/O方式

while true {
    select(streams[])
    for i in streams[] {
        if i has data
            read until unavailable
    }
}

为了避免CPU空转，可以引进了一个代理（一开始有一位叫做select的代理，后来又有一位叫做poll的代理，不过两者的本质是一样的）。这个代理可以同时观察许多流的I/O事件，在空闲的时候，会把当前线程阻塞掉，当有一个或多个流有I/O事件时，就从阻塞态中醒来，于是我们的程序就会轮询一遍所有的流。

使用select，我们有O(n)的无差别轮询复杂度，同时处理的流越多，每一次无差别轮询时间就越长。

epoll

epoll会把哪个流发生了怎样的I/O事件通知我们。（复杂度降低到了O(1)）

epoll_create 创建一个epoll对象，一般epollfd = epoll_create()
epoll_ctl （epoll_add/epoll_del的合体），往epoll对象中增加/删除某一个流的某一个事件
epoll_ctl(epollfd, EPOLL_CTL_ADD, socket, EPOLLIN);//注册缓冲区非空事件，即有数据流入
epoll_ctl(epollfd, EPOLL_CTL_DEL, socket, EPOLLOUT);//注册缓冲区非满事件，即流可以被写入
epoll_wait(epollfd,…)等待直到注册的事件发生
（注：当对一个非阻塞流的读写发生缓冲区满或缓冲区空，write/read会返回-1，并设置errno=EAGAIN。而epoll只关心缓冲区非满和缓冲区非空事件）。

while true {
    active_stream[] = epoll_wait(epollfd)
    for i in active_stream[] {
        read or write till
    }
}

在内核的最底层是中断，类似系统回调的机制。网卡设备对应一个中断号, 当网卡收到网络端的消息的时候会向CPU发起中断请求, 然后CPU处理该请求. 通过驱动程序进而操作系统得到通知, 系统然后通知epoll, epoll通知用户代码。epoll在被内核初始化时（操作系统启动），同时会开辟出epoll自己的内核高速cache区，用于安置每一个我们想监控的socket，这些socket会以红黑树的形式保存在内核cache里，以支持快速的查找、插入、删除。这个内核高速cache区，就是建立连续的物理内存页，然后在之上建立slab层，简单的说，就是物理上分配好你想要的size的内存对象，每次使用时都是使用空闲的已分配好的对象。

epoll和select的区别

进程通过将一个或多个fd传递给select或poll系统调用，阻塞在select;这样select/poll可以帮我们侦测许多fd是否就绪；但是select/poll是顺序扫描fd是否就绪，而且支持的fd数量有限。linux还提供了一个epoll系统调用，epoll是基于事件驱动方式，而不是顺序扫描,当有fd就绪时，立即回调函数rollback

传统的BIO里面socket.read()，如果TCP RecvBuffer里没有数据，函数会一直阻塞，直到收到数据，返回读到的数据。

对于NIO，如果TCP RecvBuffer有数据，就把数据从网卡读到内存，并且返回给用户；反之则直接返回0，永远不会阻塞。

最新的AIO(Async I/O)里面会更进一步：不但等待就绪是非阻塞的，就连数据从网卡到内存的过程也是异步的。

换句话说，BIO里用户最关心“我要读”，NIO里用户最关心”我可以读了”，在AIO模型里用户更需要关注的是“读完了”。

NIO一个重要的特点是：socket主要的读、写、注册和接收函数，在等待就绪阶段都是非阻塞的，真正的I/O操作是同步阻塞的（消耗CPU但性能非常高）。

NIO的读写函数可以立刻返回，这就给了我们不开线程利用CPU的最好机会：如果一个连接不能读写（socket.read()返回0或者socket.write()返回0），我们可以把这件事记下来，记录的方式通常是在Selector上注册标记位，然后切换到其它就绪的连接（channel）继续进行读写。

下面具体看下如何利用事件模型单线程处理所有I/O请求：

NIO的主要事件有几个：读就绪、写就绪、有新连接到来。

我们首先需要注册当这几个事件到来的时候所对应的处理器。然后在合适的时机告诉事件选择器：我对这个事件感兴趣。对于写操作，就是写不出去的时候对写事件感兴趣；对于读操作，就是完成连接和系统没有办法承载新读入的数据的时；对于accept，一般是服务器刚启动的时候；而对于connect，一般是connect失败需要重连或者直接异步调用connect的时候。

其次，用一个死循环选择就绪的事件，会执行系统调用（Linux 2.6之前是select、poll，2.6之后是epoll，Windows是IOCP），还会阻塞的等待新事件的到来。新事件到来的时候，会在selector上注册标记位，标示可读、可写或者有连接到来。

注意，select是阻塞的，无论是通过操作系统的通知（epoll）还是不停的轮询(select，poll)，这个函数是阻塞的。所以你可以放心大胆地在一个while(true)里面调用这个函数而不用担心CPU空转。

Java NIO 由以下几个核心部分组成：

Channels
Buffers
Selectors

Buffer

当我们需要与 NIO Channel 进行交互时, 我们就需要使用到 NIO Buffer, 即数据从 Buffer读取到 Channel 中, 并且从 Channel 中写入到 Buffer 中.
实际上, 一个 Buffer 其实就是一块内存区域, 我们可以在这个内存区域中进行数据的读写. NIO Buffer 其实是这样的内存块的一个封装, 并提供了一些操作方法让我们能够方便地进行数据的读写.

Buffer 类型有:

ByteBuffer 包括HeapByteBuffer和DirectByteBuffer两种。
ByteBuffer

public static ByteBuffer allocate(int capacity) {
    if (capacity < 0)
        throw new IllegalArgumentException();
    return new HeapByteBuffer(capacity, capacity);
}

HeapByteBuffer 通过初始化字节数组hd，在虚拟机堆上申请内存空间。

   HeapByteBuffer(int cap, int lim) {            // package-private

        super(-1, 0, lim, cap, new byte[cap], 0);
        /*
        hb = new byte[cap];
        offset = 0;
        */
    }


    ByteBuffer(int mark, int pos, int lim, int cap,   // package-private
                 byte[] hb, int offset)
    {
        super(mark, pos, lim, cap);
        this.hb = hb;
        this.offset = offset;
    }

final byte[] hb;

1
2
3

public static ByteBuffer allocateDirect(int capacity) {
    return new DirectByteBuffer(capacity);
}

DirectByteBuffer 通过unsafe.allocateMemory在物理内存中申请地址空间（非jvm堆内存），并在ByteBuffer的address变量中维护指向该内存的地址。
unsafe.setMemory(base, size, (byte) 0)方法把新申请的内存数据清零。

DirectByteBuffer(int cap) {                   // package-private
     super(-1, 0, cap, cap);
    //-Dsun.nio.PageAlignDirectMemory=true 判断是否开启按页分配对齐
     boolean pa = VM.isDirectMemoryPageAligned();
    //默认4k
     int ps = Bits.pageSize();
     long size = Math.max(1L, (long)cap + (pa ? ps : 0));
     Bits.reserveMemory(size, cap);

     long base = 0;
     try {
         base = unsafe.allocateMemory(size);
     } catch (OutOfMemoryError x) {
         Bits.unreserveMemory(size, cap);
         throw x;
     }
     unsafe.setMemory(base, size, (byte) 0);
     if (pa && (base % ps != 0)) {
         // 如果是按页分配对齐的，对齐到地址的页首，为什么要使用页首呢？
         //CPU不会一次读取或写入一个字节。相反，CPU一次访问2、4、8、16或32字节块中的内存。这样做的原因是性能 —在4字节或16字节边界上访问地址要比在1字节边界上访问地址快得多。
         //如果数据没有对齐为4字节的边界，CPU必须执行额外的工作来访问数据：加载2个数据块，转移不需要的字节，然后将它们组合在一起。这个过程肯定会降低性能，浪费CPU周期，只是为了从内存中获得正确的数据。
         //http://www.songho.ca/misc/alignment/dataalign.html
         address = base + ps - (base & (ps - 1));
     } else {
         address = base;
     }
     cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
     att = null;
 }

CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
MappedByteBuffer

使用Buffer读写数据一般遵循以下四个步骤：

写入数据到Buffer
调用flip()方法
从Buffer中读取数据
调用clear()方法或者compact()方法

当我们将数据写入到 Buffer 中时, Buffer 会记录我们已经写了多少的数据, 当我们需要从 Buffer 中读取数据时, 必须调用 Buffer.flip()将 Buffer 切换为读模式.
一旦读取了所有的 Buffer 数据, 那么我们必须清理 Buffer, 让其从新可写, 清理 Buffer 可以调用 Buffer.clear() 或 Buffer.compact().
例如:

IntBuffer intBuffer = IntBuffer.allocate(2);
intBuffer.put(12345678);
intBuffer.put(2);
intBuffer.flip();
System.err.println(intBuffer.get());
System.err.println(intBuffer.get());

mark：初始值为-1，用于备份当前的position
position：初始值为0。position表示当前可以写入或读取数据的位置。当写入或读取一个数据后， position向前移动到下一个位置。
limit：
写模式下，limit表示最多能往Buffer里写多少数据，等于capacity值。
读模式下，limit表示最多可以读取多少数据。
capacity：缓存数组大小

mark()：把当前的position赋值给mark

public final Buffer mark() {
    mark = position;
    return this;
}

reset()：把mark值还原给position

public final Buffer reset() {
    int m = mark;
    if (m < 0)
        throw new InvalidMarkException();
    position = m;
    return this;
}

clear()：一旦读完Buffer中的数据，需要让Buffer准备好再次被写入，clear会恢复状态值，但不会擦除数据。

public final Buffer clear() {
    position = 0;
    limit = capacity;
    mark = -1;
    return this;
}

flip()：Buffer有两种模式，写模式和读模式，flip后Buffer从写模式变成读模式。

public final Buffer flip() {
    limit = position;
    position = 0;
    mark = -1;
    return this;
}

rewind()：重置position为0，从头读写数据。

public final Buffer rewind() {
    position = 0;
    mark = -1;
    return this;
}

Channel

流是单向的，通道是双向的，可读可写。
流读写是阻塞的，通道可以异步读写。
流中的数据可以选择性的先读到缓存中，通道的数据总是要先读到一个缓存中，或从缓存中写入，如下所示：

目前已知Channel的实现类有：

FileChannel 从文件中读写数据。
DatagramChannel 能通过UDP读写网络中的数据。
SocketChannel 能通过TCP读写网络中的数据。
ServerSocketChannel 可以监听新进来的TCP连接，像Web服务器那样。对每一个新进来的连接都会创建一个SocketChannel。

从Channel写到Buffer的例子

int bytesRead = inChannel.read(buf); //read into buffer.

从Buffer读取数据到Channel的例子：

int bytesWritten = inChannel.write(buf);

FileChannel的read、write和map通过其实现类FileChannelImpl实现。

File file = new RandomAccessFile("data.txt", "rw");
FileChannel channel = file.getChannel();
ByteBuffer buffer = ByteBuffer.allocate(48);

int bytesRead = channel.read(buffer);
while (bytesRead != -1) {
    System.out.println("Read " + bytesRead);
    buffer.flip();
    while(buffer.hasRemaining()){
        System.out.print((char) buffer.get());
    }
    buffer.clear();
    bytesRead = channel.read(buffer);
}
file.close();

public int read(ByteBuffer dst) throws IOException {
    ensureOpen();
    if(!readable) {
        throw new NonReadableChannelException();
    } else {
        synchronized(positionLock) {
            int n = 0;
            int ti = -1;
            try {
               begin();
                ti = threads.add();
                if(!isOpen()) {
                    return 0;
                } else {
                    do {
                        n = IOUtil.read(this.fd, dst, -1L, this.nd);
                    } while(n == IOStatus.INTERRUPTED && this.isOpen());

                    return IOStatus.normalize(n);
                }
            } finally {
               threads.remove(ti);
               end(n > 0);
                assert IOStatus.check(n);
            }
        }
    }
}

IOUtil

static int read(FileDescriptor fd, ByteBuffer dst, long position, NativeDispatcher nd) throws IOException {
    if(dst.isReadOnly()) {
        throw new IllegalArgumentException("Read-only buffer");
    } else if(dst instanceof DirectBuffer) {
        return readIntoNativeBuffer(fd, dst, position, ));
    } else {
        ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
        try {
            int n = readIntoNativeBuffer(fd, bb, position, ));
            bb.flip();
            if(n > 0) {
                dst.put(bb);
            }
            return n;
        } finally {
            Util.offerFirstTemporaryDirectBuffer(var5);
        }
    }
}

通过上述实现可以看出，基于channel的文件数据读取步骤如下：
1、申请一块和缓存同大小的DirectByteBuffer bb。
2、读取数据到缓存bb，底层由NativeDispatcher的read实现。
3、把bb的数据读取到dst（用户定义的缓存，在jvm中分配内存）。

read方法导致数据复制了两次。

public int write(ByteBuffer src) throws IOException {
    ensureOpen();
     if(!writable) {
         throw new NonWritableChannelException();
     } else {
         synchronized(positionLock) {
             int n = 0;
             int ti = -1;

             try {
                begin();
                 ti =threads.add();
                 if(isOpen()) {
                     do {
                         n = IOUtil.write(this.fd, src, -1L, this.nd);
                     } while(n == IOStatus.INTERRUPTED && this.isOpen());
                     return IOStatus.normalize(n);
                 }
                 return 0;
             } finally {
                threads.remove(ti);
                end(n > 0);
                 assert IOStatus.check(n);
             }
         }
     }
 }

static int write(FileDescriptor fd, ByteBuffer src, long position, NativeDispatcher nd) throws IOException {
    if(src instanceof DirectBuffer) {
        return writeFromNativeBuffer(fd, src, position, nd);
    } else {
        int pos = src.position();
        int lim = src.limit();
        assert(pos <= lim);
        int rem = pos <= lim?lim - pos:0;
        ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
        try {
            bb.put(src);
            bb.flip();
            src.position(pos);
            int n = writeFromNativeBuffer(fd, bb, position, nd);
            if(n > 0) {
                src.position(pos + n);
            }
            return n;
        } finally {
            Util.offerFirstTemporaryDirectBuffer(bb);
        }
    }
}

通过上述实现可以看出，基于channel的文件数据写入步骤如下：
1、申请一块DirectByteBuffer，bb大小为byteBuffer中的limit - position。
2、复制byteBuffer中的数据到bb中。
3、把数据从bb中写入到文件，底层由NativeDispatcher的write实现，具体如下：

private static int writeFromNativeBuffer(FileDescriptor fd, 
      ByteBuffer bb, long position, NativeDispatcher nd)
  throws IOException {
  int pos = bb.position();
  int lim = bb.limit();
  assert (pos <= lim);
  int rem = (pos <= lim ? lim - pos : 0);

  int written = 0;
  if (rem == 0)
      return 0;
  if (position != -1) {
      written = nd.pwrite(fd,
                          ((DirectBuffer)bb).address() + pos,
                          rem, position);
  } else {
      written = nd.write(fd, ((DirectBuffer)bb).address() + pos, rem);
  }
  if (written > 0)
      bb.position(pos + written);
  return written;
}

transferFrom()

FileChannel的transferFrom()方法可以将数据从源通道传输到FileChannel中。在SoketChannel的实现中，SocketChannel只会传输此刻准备好的数据（可能不足count字节）。因此，SocketChannel可能不会将请求的所有数据(count个字节)全部传输到FileChannel中。

toChannel.transferFrom(0, fromChannel.size(), fromChannel);

transferTo()

transferTo()方法将数据从FileChannel传输到其他的channel中

fromChannel.transferTo(position, count, toChannel);

Scattering Reads

Scattering Reads是指数据从一个channel读取到多个buffer中

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body   = ByteBuffer.allocate(1024);
ByteBuffer[] bufferArray = { header, body };
channel.read(bufferArray);

read()方法按照buffer在数组中的顺序将从channel中读取的数据写入到buffer，当一个buffer被写满后，channel紧接着向另一个buffer中写

Gathering Writes

Gathering Writes是指数据从多个buffer写入到同一个channel

write()方法会按照buffer在数组中的顺序，将数据写入到channel，注意只有position和limit之间的数据才会被写入

Selector

Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。仅用单个线程来处理多个Channels的好处是，只需要更少的线程来处理通道。事实上，可以只用一个线程处理所有的通道。对于操作系统来说，线程之间上下文切换的开销很大，而且每个线程都要占用系统的一些资源（如内存）。因此，使用的线程越少越好。

Selector的创建
Selector selector = Selector.open();

向Selector注册通道

1 2	channel.configureBlocking(false); SelectionKey key = channel.register(selector,Selectionkey.OP_READ);

Connect SelectionKey.OP_CONNECT(8)
Accept SelectionKey.OP_ACCEPT(16)
Read SelectionKey.OP_READ(1)
Write SelectionKey.OP_WRITE(4)

可以用“位或”操作符将常量连接
int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;

interest集合

int interestSet = selectionKey.interestOps();
boolean isInterestedInAccept  = (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT；
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;

ready集合

ready 集合是通道已经准备就绪的操作的集合。在一次选择(Selection)之后，你会首先访问这个ready set。可以这样访问ready集合：

int readySet = selectionKey.readyOps();

selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

    public final boolean isReadable() {
        return (readyOps() & OP_READ) != 0;
    }

    public final boolean isWritable() {
        return (readyOps() & OP_WRITE) != 0;
    }

    public final boolean isConnectable() {
        return (readyOps() & OP_CONNECT) != 0;
    }

    public final boolean isAcceptable() {
        return (readyOps() & OP_ACCEPT) != 0;
    }

Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    if(key.isAcceptable()) {
        // a connection was accepted by a ServerSocketChannel.
    } else if (key.isConnectable()) {
        // a connection was established with a remote server.
    } else if (key.isReadable()) {
        // a channel is ready for reading
    } else if (key.isWritable()) {
        // a channel is ready for writing
    }
    keyIterator.remove();
}

注意每次迭代末尾的keyIterator.remove()调用。Selector不会自己从已选择键集中移除SelectionKey实例。必须在处理完通道时自己移除。下次该通道变成就绪时，Selector会再次将其放入已选择键集中。

SelectionKey.channel()方法返回的通道需要转型成你要处理的类型，如ServerSocketChannel或SocketChannel等。

 selector = Selector.open();//创建多路复用器
 serverSocketChannel = ServerSocketChannel.open();//打开管道
 serverSocketChannel.socket()
       .bind(new InetSocketAddress(port),1024);//绑定端口
 serverSocketChannel.configureBlocking(false);//非阻塞
 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//管道注册到多路复用器上

 Set<SelectionKey> keySet = selector.selectedKeys();
 Iterator<SelectionKey> iterator =  keySet.iterator();
 while (iterator.hasNext()){
        SelectionKey key = iterator.next();
        iterator.remove();
        try {
             if (key.isAcceptable()){
                 SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();//完成tcp握手，建立物理链路
                 socketChannel.configureBlocking(false);
                 socketChannel.register(selector,SelectionKey.OP_READ,ByteBuffer.allocate(1024));//注册客户端到多路复用器上，监听读操作
         }
        }catch (Exception e){//关闭
             if (key!=null){
                  key.cancel();
                  if (key.channel()!=null){
                        key.channel().close();
                   }
               }
        }
}

非阻塞IO通道（Non-blocking IO Pipelines）

非阻塞的IO管道（Non-blocking IO Pipelines）可以看做是整个非阻塞IO处理过程的链条。包括在以非阻塞形式进行的读与写操作

我们的组件（Component）通过Selector检查当前Channel是否有数据需要写入。此时component读入数据，并且根据输入的数据input对外提供数据输出output。这个对外的数据输出output被写到了另一个Channel中。
一个非阻塞的IO管道不必同时需要读和写数据，通常来说有些管道只需要读数据，而另一些管道则只需写数据。当然一个非阻塞的IO管道他也可以同时从多个Channel中读取数据，例如同时冲多个SocketChannel中读取数据；

非阻塞和阻塞通道比较（Non-blocking vs. Blocking IO Pipelines）

非阻塞IO管道和阻塞IO管道之间最大的区别是他们各自如何从Channel（套接字socket或文件file）读写数据。
IO管道通常直接从流中读取数据，然后把数据分割为连续的消息。这个处理与我们读取流信息，用tokenizer进行解析非常相似。不同的是我们在这里会把数据流分割为更大一些的消息块。我把这个过程叫做Message Reader.下面是一张说明的插图：

一个阻塞IO管道的使用可以和输入流一样调用，每次从Channel中读取一个字节的数据，阻塞自身直到有数据可读。这个流程就是一个阻塞的Messsage Reader实现。

使用阻塞IO大大简化了Message Reader的实现成本。阻塞的Message Reader无需关注没有数据返回的情形，无需关注返回部分数据或者数据解析需要被复用的问题。
相似的，一个阻塞的Message Writer也不需要关注写入部分数据，和数据复用的问题。

基础的非阻塞通道设计（Basic Non-blocking IO Pipeline Design）

一个非阻塞的IO通道可以用单线程读取多个数据流。这个前提是相关的流可以切换为非阻塞模式（并不是所有流都可以以非阻塞形式操作）。在非阻塞模式下，读取一个流可能返回0个或多个字节。如果流还没有可供读取的数据那么就会返回0，其他大于1的返回都表明这是实际读取到的数据；
为了避开没有数据可读的流，我们结合Java NIO中的Selector。一个Selector可以注册多个SelectableChannel实例。当我们调用select()或selectorNow()方法时Selector会返回一个有数据可读的SelectableChannel实例。这个设计可以如下插图：

读取部分信息(Reading Partial Messages)

当我们冲SelectableChannel中读取一段数据后，我们并不知道这段数据是否是完整的一个message。因为一个数据段可能包含部分message，也就是说即可能少于一个message，也可能多一个message，正如下面这张插图所示意的那样：

要处理这种截断的message，我们会遇到两个问题：

检测数据段中是否包含一个完整的message
在message剩余部分获取到之前，我们如何处理不完整的message

检测完整message要求Message Reader查看数据段中的数据是否至少包含一个完整的message。如果包含一个或多个完整message，这些message可以被下发到通道中处理。查找完整message的过程是个大量重复的操作，所以这个操作必须是越快越好的。

当数据段中有一个不完整的message时，无论不完整消息是整个数据段还是说在完整message前后，这个不完整的message数据都需要在剩余部分获得前存储起来。

检查message完整性和存储不完整message都是Message Reader的职责。为了避免混淆来自不同Channel的数据，我们为每一个Channel分配一个Message Reader。整个设计大概是这样的：

当我们通过Selector获取到一个有数据可以读取的Channel之后，改Channel关联的Message Reader会读取数据，并且把数据打断为Message块。得到完整的message后就可以通过通道下发到其他组件进行处理。
一个Message Reader自然是协议相关的。他需要知道message的格式以便读取。如果我们的服务器是跨协议复用的，那他必须实现Message Reader的协议-大致类似于接收一个Message Reader工厂作为配置参数。

存储不完整的Message（Storing Partial Messages）

现在我们已经明确了由Message Reader负责不完整消息的存储直到接收到完整的消息。闲杂我们还需要知道这个存储过程需要如何来实现。
在设计的时候我们需要考虑两个关键因素：

我们希望在拷贝消息数据的时候数据量能尽可能的小，拷贝量越大则性能相对越低；
我们希望完整的消息是以顺序的字节存储，这样方便进行数据的解析；

为每个Message Reade分配Buffer（A Buffer Per Message Reader）

显然不完整的消息数据需要存储在某种buffer中。比较直接的办法是我们为每个Message Reader都分配一个内部的buffer成员。但是，多大的buffer才合适呢？这个buffer必须能存储下一个message最大的大小。如果一个message最大是1MB，那每个Message Reader内部的buffer就至少有1MB大小。
在百万级别的并发链接数下，1MB的buffer基本没法正常工作。举例来说，1,000,000 x 1MB就是1TB的内存大小！如果消息的最大数据量是16MB又需要多少内存呢？128MB呢？

可伸缩Buffer（Resizable Buffers）

另一个方案是在每个Message Reader内部维护一个容量可变的buffer。一个可变的buffer在初始化时占用较少控件，在消息变得很大超出容量时自动扩容。这样每个链接就不需要都占用比如1MB的空间。每个链接只使用承载下一个消息所必须的内存大小。

拷贝扩容（Resize by Copy）

第一种实现可伸缩buffer的办法是初始化buffer的时候只申请较少的空间，比如4KB。如果消息超出了4KB的大小那么开赔一个更大的空间，比如8KB，然后把4KB中的数据拷贝纸8KB的内存块中。
以拷贝方式扩容的优点是一个消息的全部数据都被保存在了一个连续的字节数组中。这使得数据解析变得更加容易。
同时它的缺点是会增加大量的数据拷贝操作。
为了减少数据的拷贝操作，你可以分析整个消息流中的消息大小，一次来找到最适合当前机器的可以减少拷贝操作的buffer大小。例如，你可能会注意到觉大多数的消息都是小于4KB的，因为他们仅仅包含了一个非常请求和响应。这意味着消息的处所荣校应该设置为4KB。
同时，你可能会发现如果一个消息大于4KB，很可能是因为他包含了一个文件。你会可能注意到大多数通过系统的数据都是小于128KB的。所以我们可以在第一次扩容设置为128KB。
最后你可能会发现当一个消息大于128KB后，没有什么规律可循来确定下次分配的空间大小，这意味着最后的buffer容量应该设置为消息最大的可能数据量。
结合这三次扩容时的大小设置，可以一定程度上减少数据拷贝。4KB以下的数据无需拷贝。在1百万的连接下需要的空间例如1,000,000x4KB=4GB，目前（2015）大多数服务器都扛得住。4KB到128KB会仅需拷贝一次，即拷贝4KB数据到128KB的里面。消息大小介于128KB和最大容量的时需要拷贝两次。首先4KB数据被拷贝第二次是拷贝128KB的数据，所以总共需要拷贝132KB数据。假设没有很多的消息会超过128KB，那么这个方案还是可以接受的。
当一个消息被完整的处理完毕后，它占用的内容应当即刻被释放。这样下一个来自东一个链接通道的消息可以从最小的buffer大小重新开始。这个操作是必须的如果我们需要尽可能高效地复用不同链接之间的内存。大多数情况下并不是所有的链接都会在同一时刻需要大容量的buffer。
笔者写了一个完整的教程阐述了如何实现一个内存buffer使其支持扩容：Resizable Arrays 。这个教程也附带了一个指向GitHub上的源码仓地址，里面有实现方案的具体代码。

追加扩容（Resize by Append）

另一种实现buffer扩容的方案是让buffer包含几个数组。当需要扩容的时候只需要在开辟一个新的字节数组，然后把内容写到里面去。
这种扩容也有两个具体的办法。一中是开辟单独的字节数组，然后用一个列表把这些独立数组关联起来。另一种是开辟一些更大的，相互共享的字节数组切片，然后用列表把这些切片和buffer关联起来。个人而言，笔者认为第二种切片方案更好一点点，但是它们之前的差异比较小。
这种追加扩容的方案不管是用独立数组还是切片都有一个优点，那就是写数据的时候不需要二外的拷贝操作。所有的数据可以直接从socket（Channel）中拷贝至数组活切片当中。
这种方案的缺点也很明显，就是数据不是存储在一个连续的数组中。这会使得数据的解析变得更加复杂，因为解析器不得不同时查找每一个独立数组的结尾和所有数组的结尾。正因为我们需要在写数据时查找消息的结尾，这个模型在设计实现时会相对不那么容易。

TLV编码消息(TLV Encoded Messages)

有些协议的消息消失采用的是一种TLV格式（Type, Length, Value）。这意味着当消息到达时，消息的完整大小存储在了消息的开始部分。我们可以立刻判断为消息开辟多少内存空间。
TLV编码是的内存管理变得更加简单。我们可以立刻知道为消息分配多少内存。即便是不完整的消息，buffer结尾后面也不会有浪费的内存。
TLV编码的一个缺点是我们需要在消息的全部数据接收到之前就开辟好需要用的所有内存。因此少量链接慢，但发送了大块数据的链接会占用较多内存，导致服务器无响应。
解决上诉问题的一个变通办法是使用一种内部包含多个TLV的消息格式。这样我们为每个TLV段分配内存而不是为整个的消息分配，并且只在消息的片段到达时才分配内存。但是消息片段很大时，任然会出现一样的问题。
另一个办法是为消息设置超时，如果长时间未接收到的消息（比如10-15秒）。这可以让服务器从偶发的并发处理大块消息恢复过来，不过还是会让服务器有一段时间无响应。另外恶意的DoS攻击会导致服务器开辟大量内存。
TLV编码有不同的变种。有多少字节使用这样确切的类型和字段长度取决于每个独立的TLV编码。有的TLV编码吧字段长度放在前面，接着放类型，最后放值。尽管字段的顺序不同，但他任然是一个TLV的类型。
TLV编码使得内存管理更加简单，这也是HTTP1.1协议让人觉得是一个不太优良的的协议的原因。正因如此，HTTP2.0协议在设计中也利用TLV编码来传输数据帧。也是因为这个原因我们设计了自己的利用TLV编码的网络协议VStack.co。

写不完整的消息（Writing Partial Messages）

在非阻塞IO管道中，写数据也是一个不小的挑战。当你调用一个非阻塞模式Channel的write()方法时，无法保证有多少机字节被写入了ByteBuffer中。write方法返回了实际写入的字节数，所以跟踪记录已被写入的字节数也是可行的。这就是我们遇到的问题：持续记录被写入的不完整的小树知道一个消息中所有的数据都发送完毕。
为了管理不完整消息的写操作，我们需要创建一个Message Writer。正如前面的Message Reader，我们也需要每个Channel配备一个Message Writer来写数据。在每个Message Writer中我们记录准确的已经写入的字节数。
为了避免多个消息传递到Message Writer超出他所能处理到Channel的量，我们需要让到达的消息进入队列。Message Writer则尽可能快的将数据写到Channel里。
下面是一个流程图，展示的是不完整消息被写入的过程：

为了使Message Writer能够持续发送刚才已经发送了一部分的消息，Message Writer需要被移植调用，这样他就可以发送更多数据。

如果你有大量的链接，你会持有大量的Message Writer实例。检查比如1百万的Message Writer实例是来确定他们是否处于可写状态是很慢的操作。首先，许多Message Writer可能根本就没有数据需要发送。我们不想检查这些实例。其次，不是所有的Channel都处于可写状态。我们不想浪费时间在这些非写入状态的Channel。

为了检查一个Channel是否可写，可以把它注册到Selector上。但是我们不希望把所有的Channel实例都注册到Selector。试想一下，如果你有1百万的链接，这里面大部分是空闲的，把1百万链接都祖册到Selector上。然后调用select方法的时候就会有很多的Channel处于可写状态。你需要检查所有这些链接中的Message Writer以确认是否有数据可写。
为了避免检查所有的这些Message Writer，以及那些根本没有消息需要发送给他们的Channel实例，我么可以采用入校两步策略：

当有消息写入到Message Writer忠厚，把它关联的Channel注册到Selector上（如果还未注册的话）。
当服务器有空的时候，可以检查Selector看看注册在上面的Channel实例是否处于可写状态。每个可写的channel，使其Message Writer向Channel中写入数据。如果Message Writer已经把所有的消息都写入Channel，把Channel从Selector上解绑。

这两个小步骤确保只有有数据要写的Channel才会被注册到Selector。

集成（Putting it All Together）

正如你所知到的，一个被阻塞的服务器需要时刻检查当前是否有显得完整消息抵达。在一个消息被完整的收到前，服务器可能需要检查多次。检查一次是不够的。
类似的，服务器也需要时刻检查当前是否有任何可写的数据。如果有的话，服务器需要检查相应的链接看他们是否处于可写状态。仅仅在消息第一次进入队列时检查是不够的，因为一个消息可能被部分写入。
总而言之，一个非阻塞的服务器要三个管道，并且经常执行：

读数据管道，用来检查打开的链接是否有新的数据到达；
处理数据管道，负责处理接收到的完整消息；
写数据管道，用于检查是否有数据可以写入打开的连接中；
这三个管道在循环中重复执行。你可以尝试优化它的执行。比如，如果没有消息在队列中等候，那么可以跳过写数据管道。或者，如果没有收到新的完整消息，你甚至可以跳过处理数据管道。
下面这张流程图阐述了这整个服务器循环过程：

假如你还是感觉这比较复杂难懂，可以去clone我们的源码仓： https://github.com/jjenkov/java-nio-server 也许亲眼看到了代码会帮助你理解这一块是如何实现的。

服务器线程模型（Server Thread Model）

我们在GitHub上的源码中实现的非阻塞IO服务使用了一个包含两条线程的线程模型。第一个线程负责从ServerSocketChannel接收到达的链接。另一个线程负责处理这些链接，包括读消息，处理消息，把响应写回到链接。这个双线程模型如下：

DatagramChannel数据报通道

DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(9999));
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();

**channel.receive(buf);**

receive()方法会把接收到的数据包中的数据拷贝至给定的Buffer中。如果数据包的内容超过了Buffer的大小，剩余的数据会被直接丢弃。

String newData = "New String to wrte to file..."+System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();

**int byteSent = channel.send(buf, new InetSocketAddress("jenkov.com", 80));**

上述示例会把一个字符串发送到“jenkov.com”服务器的UDP端口80.目前这个端口没有被任何程序监听，所以什么都不会发生。当发送了数据后，我们不会收到数据包是否被接收的的通知，这是由于UDP本身不保证任何数据的发送问题。

链接特定机器地址（Connecting to a Specific Address）

DatagramChannel实际上是可以指定到网络中的特定地址的。由于UDP是面向无连接的，这种链接方式并不会创建实际的连接，这和TCP通道类似。确切的说，他会锁定DatagramChannel,这样我们就只能通过特定的地址来收发数据包。

1
2
3

channel.connect(new InetSocketAddress("jenkov.com"), 80));
int bytesRead = channel.read(buf);
int bytesWritten = channel.write(buf);

NIO Pipe管道

一个Java NIO的管道是两个线程间单向传输数据的连接。一个管道（Pipe）有一个source channel和一个sink channel(没想到合适的中文名)。我们把数据写到sink channel中，这些数据可以同过source channel再读取出来

创建管道(Creating a Pipe)

打开一个管道通过调用Pipe.open()工厂方法，如下：
Pipe pipe = Pipe.open();

向管道写入数据（Writing to a Pipe）

向管道写入数据需要访问他的sink channel,接下来就是调用write()方法写入数据了：

Pipe.SinkChannel sinkChannel = pipe.sink();
String newData = "New String to write to file..." + System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());

buf.flip();

while(buf.hasRemaining()) {
    sinkChannel.write(buf);
}

从管道读取数据（Reading from a Pipe）

类似的从管道中读取数据需要访问他的source channel,接下来调用read()方法读取数据：

Pipe.SourceChannel sourceChannel = pipe.source();
ByteBuffer buf = ByteBuffer.allocate(48);

int bytesRead = inChannel.read(buf);

NIO AsynchronousFileChannel异步文件通道

Path path = Paths.get("data/test.xml");

AsynchronousFileChannel fileChannel =
    AsynchronousFileChannel.open(path, StandardOpenOption.READ);

ByteBuffer buffer = ByteBuffer.allocate(1024);
long position = 0;
Future<Integer> operation = fileChannel.read(buffer, 0);

while(!operation.isDone());

buffer.flip();
byte[] data = new byte[buffer.limit()];
buffer.get(data);
System.out.println(new String(data));
buffer.clear();

通过CompletionHandler读取数据（Reading Data Via a CompletionHandler）

fileChannel.read(buffer, position, buffer, new CompletionHandler<Integer, ByteBuffer>() {
    @Override
    public void completed(Integer result, ByteBuffer attachment) {
        System.out.println("result = " + result);

        attachment.flip();
        byte[] data = new byte[attachment.limit()];
        attachment.get(data);
        System.out.println(new String(data));
        attachment.clear();
    }

    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {
    }
});

一旦读取完成，将会触发CompletionHandler的completed()方法，并传入一个Integer和ByteBuffer。前面的整形表示的是读取到的字节数大小。第二个ByteBuffer也可以换成其他合适的对象方便数据写入。如果读取操作失败了，那么会触发failed()方法。

参考：http://www.jianshu.com/p/052035037297
http://ifeve.com/java-nio-scattergather/
https://java-nio.avenwu.net/java-nio-channel.html
http://www.importnew.com/24794.html