TDDL 在分布式下的SEQUENCE原理

末蓝、 2022-02-05 01:39 210阅读 0赞

TDDL大家应该很熟悉了，淘宝分布式数据层。很好的为我们实现了分库分表、Master/Salve、动态数据源配置等功能。

那么分布式之后，数据库自增序列肯定用不了了，如何方便快捷的解决这个问题呢？TDDL也提供了SEQUENCE的解决方案。

下面就来简单剖析一下实现原理。。。。。。

第一步：创建一张sequence对应的表。

CREATE TABLE \`imp\_sequence\` (  
  \`BIZ\_NAME\` varchar(45) NOT NULL COMMENT '业务名称',  
  \`CURRENT\_VALUE\` int(11) NOT NULL COMMENT '当前最大值',  
  \`GMT\_CREATE\` datetime DEFAULT NULL COMMENT '创建时间',  
  \`GMT\_MODIFIED\` datetime DEFAULT NULL COMMENT '修改时间',  
  PRIMARY KEY (\`BIZ\_NAME\`)  
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='数据序列表';

表名和字段可以按各自规则定义，定义之后需要与第二步DAO中的定义相对应！

几张逻辑表需要声明几个sequence

第二步：配置sequenceDao

<bean id="sequenceDao" class="com.taobao.tddl.client.sequence.impl.DefaultSequenceDao">  
          
        <property name="dataSource"  ref="dataSource" />  
          
        <property name="step" value="1000" />  
          
        <property name="retryTimes" value="1" />  
          
        <property name="tableName" value="gt\_sequence" />  
          
        <property name="nameColumnName" value="BIZ\_NAME" />  
          
        <property name="valueColumnName" value="CURRENT\_VALUE" />  
          
        <property name="gmtModifiedColumnName" value="gmt\_modified" />  
    </bean>  
第三步：配置sequence生成器  
    <bean id="businessSequence"  class="com.taobao.tddl.client.sequence.impl.DefaultSequence">  
        <property name="sequenceDao" ref="sequenceDao"/>  
        <property name="name" value="business\_sequence" />  
    </bean>  
第四步：调用  
public class IbatisSmDAO extends SqlMapClientDaoSupport implements SmDAO \{  
   
  /\*\*smSequence\*/  
  private DefaultSequence   businessSequence;  
     
    public int insert(SmDO sm) throws DataAccessException \{  
        if (sm == null) \{  
            throw new IllegalArgumentException("Can't insert a null data object into db.");  
        \}  
          
        try \{  
            sm.setId((int)businessSequence.nextValue());  
        \} catch (SequenceException e) \{  
            throw new RuntimeException("Can't get primary key.");  
        \}  
          
        getSqlMapClientTemplate().insert("MS-SM-INSERT", sm);  
   
        return sm.getId();  
    \}  
\}  
从调用配置中，我们可以发现其中涉及到二个重要类DefaultSequenceDao和DefaultSequence，这二个都是TDDL的默认实现。DefaultSequenceDao：序列DAO默认实现，JDBC方式。DefaultSequence：序列默认实现  
先来看DefaultSequenceDao，TDDL中提供了默认的表名，列名和步长等，第一步的建表可以参照默认方式。

private static final int MIN\_STEP = 1;  
    private static final int MAX\_STEP = 100000;  
    private static final int DEFAULT\_STEP = 1000;  
    private static final int DEFAULT\_RETRY\_TIMES = 150;  
   
    private static final String DEFAULT\_TABLE\_NAME = "sequence";  
    private static final String DEFAULT\_NAME\_COLUMN\_NAME = "name";  
    private static final String DEFAULT\_VALUE\_COLUMN\_NAME = "value";  
    private static final String DEFAULT\_GMT\_MODIFIED\_COLUMN\_NAME = "gmt\_modified";  
   
    private static final long DELTA = 100000000L;  
   
    private DataSource dataSource;  
   
    /\*\*  
     \* 重试次数  
     \*/  
    private int retryTimes = DEFAULT\_RETRY\_TIMES;  
   
    /\*\*  
     \* 步长  
     \*/  
    private int step = DEFAULT\_STEP;  
   
    /\*\*  
     \* 序列所在的表名  
     \*/  
    private String tableName = DEFAULT\_TABLE\_NAME;  
   
    /\*\*  
     \* 存储序列名称的列名  
     \*/  
    private String nameColumnName = DEFAULT\_NAME\_COLUMN\_NAME;  
   
    /\*\*  
     \* 存储序列值的列名  
     \*/  
    private String valueColumnName = DEFAULT\_VALUE\_COLUMN\_NAME;  
   
    /\*\*  
     \* 存储序列最后更新时间的列名  
     \*/  
    private String gmtModifiedColumnName = DEFAULT\_GMT\_MODIFIED\_COLUMN\_NAME;  
接下来看一下nextRange方法：取得下一个可用的序列区间  
public SequenceRange nextRange(String name) throws SequenceException \{  
        if (name == null) \{  
            throw new IllegalArgumentException("序列名称不能为空");  
        \}  
   
        long oldValue;  
        long newValue;  
   
        Connection conn = null;  
        PreparedStatement stmt = null;  
        ResultSet rs = null;  
   
        for (int i = 0; i < retryTimes + 1; ++i) \{  
            try \{  
                conn = dataSource.getConnection();  
                stmt = conn.prepareStatement(getSelectSql());  
                stmt.setString(1, name);  
                rs = stmt.executeQuery();  
                rs.next();  
                oldValue = rs.getLong(1);  
   
                if (oldValue < 0) \{  
                    StringBuilder message = new StringBuilder();  
                    message.append("Sequence value cannot be less than zero, value = ").append(oldValue);  
                    message.append(", please check table ").append(getTableName());  
   
                    throw new SequenceException(message.toString());  
                \}  
   
                if (oldValue > Long.MAX\_VALUE - DELTA) \{  
                    StringBuilder message = new StringBuilder();  
                    message.append("Sequence value overflow, value = ").append(oldValue);  
                    message.append(", please check table ").append(getTableName());  
   
                    throw new SequenceException(message.toString());  
                \}  
   
                newValue = oldValue + getStep();  
            \} catch (SQLException e) \{  
                throw new SequenceException(e);  
            \} finally \{  
                closeResultSet(rs);  
                rs = null;  
                closeStatement(stmt);  
                stmt = null;  
                closeConnection(conn);  
                conn = null;  
            \}  
   
            try \{  
                conn = dataSource.getConnection();  
                stmt = conn.prepareStatement(getUpdateSql());  
                stmt.setLong(1, newValue);  
                stmt.setTimestamp(2, new Timestamp(System.currentTimeMillis()));  
                stmt.setString(3, name);  
                stmt.setLong(4, oldValue);  
                int affectedRows = stmt.executeUpdate();  
                if (affectedRows == 0) \{  
                    // retry  
                    continue;  
                \}  
   
                return new SequenceRange(oldValue + 1, newValue);  
            \} catch (SQLException e) \{  
                throw new SequenceException(e);  
            \} finally \{  
                closeStatement(stmt);  
                stmt = null;  
                closeConnection(conn);  
                conn = null;  
            \}  
        \}  
   
        throw new SequenceException("Retried too many times, retryTimes = " + retryTimes);  
    \}

通过getSelectSql查询最新的value值，然后加上步点，通过getUpdateSql更新到数据库中

private String getSelectSql() \{  
        if (selectSql == null) \{  
            synchronized (this) \{  
                if (selectSql == null) \{  
                    StringBuilder buffer = new StringBuilder();  
                    buffer.append("select ").append(getValueColumnName());  
                    buffer.append(" from ").append(getTableName());  
                    buffer.append(" where ").append(getNameColumnName()).append(" = ?");  
   
                    selectSql = buffer.toString();  
                \}  
            \}  
        \}  
   
        return selectSql;  
    \}  
   
    private String getUpdateSql() \{  
        if (updateSql == null) \{  
            synchronized (this) \{  
                if (updateSql == null) \{  
                    StringBuilder buffer = new StringBuilder();  
                    buffer.append("update ").append(getTableName());  
                    buffer.append(" set ").append(getValueColumnName()).append(" = ?, ");  
                    buffer.append(getGmtModifiedColumnName()).append(" = ? where ");  
                    buffer.append(getNameColumnName()).append(" = ? and ");  
                    buffer.append(getValueColumnName()).append(" = ?");  
   
                    updateSql = buffer.toString();  
                \}  
            \}  
        \}  
   
        return updateSql;  
    \}  
有一个特殊需要说明的，在update语句中，where需要把之前的value当成条件传入。实现了类型version的乐观锁操作。如果同一个时间AB二台机器同时请求获取到相同的value，进行update操作只有可能一条成功。失败的会按retryTimes进行重试。  
接下来看DefaultSequence，比较简单，就不说明了

public class DefaultSequence implements Sequence \{  
    private final Lock lock = new ReentrantLock();  
   
    private SequenceDao sequenceDao;  
   
    /\*\*  
     \* 序列名称  
     \*/  
    private String name;  
   
    private volatile SequenceRange currentRange;  
   
    public long nextValue() throws SequenceException \{  
        if (currentRange == null) \{  
            lock.lock();  
            try \{  
                if (currentRange == null) \{  
                    currentRange = sequenceDao.nextRange(name);  
                \}  
            \} finally \{  
                lock.unlock();  
            \}  
        \}  
   
        long value = currentRange.getAndIncrement();  
        if (value == -1) \{  
            lock.lock();  
            try \{  
                for (;;) \{  
                    if (currentRange.isOver()) \{  
                        currentRange = sequenceDao.nextRange(name);  
                    \}  
   
                    value = currentRange.getAndIncrement();  
                    if (value == -1) \{  
                        continue;  
                    \}  
   
                    break;  
                \}  
            \} finally \{  
                lock.unlock();  
            \}  
        \}  
   
        if (value < 0) \{  
            throw new SequenceException("Sequence value overflow, value = " + value);  
        \}  
   
        return value;  
    \}  
   
    public SequenceDao getSequenceDao() \{  
        return sequenceDao;  
    \}  
   
    public void setSequenceDao(SequenceDao sequenceDao) \{  
        this.sequenceDao = sequenceDao;  
    \}  
   
    public String getName() \{  
        return name;  
    \}  
   
    public void setName(String name) \{  
        this.name = name;  
    \}  
\}

\---------------------   
作者：hdu09075340   
来源：CSDN   
原文：https://blog.csdn.net/hdu09075340/article/details/79103851   
版权声明：本文为博主原创文章，转载请附上博文链接！