1. @Slf4j
    2. public class BufferManager
    3. {
    4.     /**
    5.      * 业务命名空间
    6.      */
    7.     private String namespace;
    8.     private BitSequenceConfig buffer1 = new BitSequenceConfig();
    9.     private BitSequenceConfig buffer2 = new BitSequenceConfig();
    10.     private BitSequenceConfig currentBuffer;
    11.     private PaddedLong currentSequence;
    12.     /**
    13.      * 刷新的状态
    14.      */
    15.     private volatile RefreshEnum refreshState = RefreshEnum.NON;
    16.     /**
    17.      * 这里暂时设置刷新比率为0.4
    18.      */
    19.     private float refreshRatio = 0.4f;
    20.     private int refreshValue = (int) (SEQ_MAX_SIZE * refreshRatio);
    22.     public BufferManager(BitSequenceDTO bitSequenceDTO)
    23.     {
    24.         if (null == bitSequenceDTO)
    25.         {
    26.             return;
    27.         }
    28.         this.namespace = bitSequenceDTO.getNamespace();
    29.         buffer1.update(bitSequenceDTO);
    30.         this.currentBuffer = buffer1;
    31.         this.currentSequence = new PaddedLong(0);
    32.     }
    34.     public int getRsv()
    35.     {
    36.         return currentBuffer.getRsv();
    37.     }
    39.     public long getTime()
    40.     {
    41.         return currentBuffer.getTime();
    42.     }
    44.     /**
    45.      * 双buffer方式获取新的序列值
    46.      *
    47.      * @return 新的序列
    48.      */
    49.     public long getSequence()
    50.     {
    51.         long sequence;
    52.         while (true)
    53.         {
    54.             sequence = currentSequence.get();
    56.             // 二级buffer刷新处理
    57.             if (reachRefreshBuffer(sequence))
    58.             {
    59.                 if (refreshState == RefreshEnum.READY_ASYNC)
    60.                 {
    61.                     continue;
    62.                 }
    64.                 synchronized (this)
    65.                 {
    66.                     if (refreshState == RefreshEnum.READY_ASYNC)
    67.                     {
    68.                         continue;
    69.                     }
    70.                     refreshState = RefreshEnum.READY_ASYNC;
    71.                 }
    73.                 // 刷新二级buffer
    74.                 refreshBuffer();
    75.             }
    77.             // 到达最后
    78.             if (sequence >= SEQ_MAX_SIZE)
    79.             {
    80.                 if (refreshState != RefreshEnum.READY_RPC && refreshState != RefreshEnum.FINISH)
    81.                 {
    82.                     throw new SnowflakeException("server端获取新的buffer失败");
    83.                 }
    84.                 switchBuffer();
    85.             }
    86.             return currentSequence.getAndIncrement();
    87.         }
    88.     }
    90.     public int getWorkId()
    91.     {
    92.         return currentBuffer.getWorkId();
    93.     }
    95.     /**
    96.      * 到达刷新二级buffer的点
    97.      */
    98.     private boolean reachRefreshBuffer(long sequence)
    99.     {
    100.         return (sequence >= refreshValue && refreshState != RefreshEnum.READY_RPC && refreshState != RefreshEnum.FINISH);
    101.     }
    103.     private void refreshBuffer()
    104.     {
    105.         CompletableFuture.runAsync(() ->
    106.         {
    107.             try
    108.             {
    109.                 synchronized (this)
    110.                 {
    111.                     refreshState = RefreshEnum.READY_RPC;
    112.                     updateBuffer(SnowflakeSeqGeneratorFactory.getInstance().getSequenceApi().getNext(namespace));
    113.                     refreshState = RefreshEnum.FINISH;
    114.                 }
    115.             }
    116.             catch (Throwable e)
    117.             {
    118.                 log.error("rpc 调用异常", e);
    119.             }
    120.         });
    121.     }
    123.     /**
    124.      * 到达最后则进行buffer切换
    125.      */
    126.     private synchronized void switchBuffer()
    127.     {
    128.         if (refreshState == RefreshEnum.READY_RPC)
    129.         {
    130.             throw new SnowflakeException("server端获取新的buffer失败");
    131.         }
    132.         // 表示切换完成
    133.         if (refreshState == RefreshEnum.NON)
    134.         {
    135.             return;
    136.         }
    138.         if (currentBuffer == buffer1)
    139.         {
    140.             currentBuffer = buffer2;
    141.         }
    142.         else if (currentBuffer == buffer2)
    143.         {
    144.             currentBuffer = buffer1;
    145.         }
    147.         currentSequence.set(0);
    148.         refreshState = RefreshEnum.NON;
    149.     }
    151.     private void updateBuffer(PlainResult<BitSequenceDTO> sequenceDTO)
    152.     {
    153.         if (!sequenceDTO.isOk())
    154.         {
    155.             log.error("server返回信息异常：rsp={}", sequenceDTO);
    156.             throw new SnowflakeException("server返回信息异常");
    157.         }
    159.         BitSequenceDTO bitSequence = sequenceDTO.getData();
    160.         if (currentBuffer == buffer1)
    161.         {
    162.             buffer2.update(bitSequence);
    163.         }
    164.         else if (currentBuffer == buffer2)
    165.         {
    166.             buffer1.update(bitSequence);
    167.         }
    168.     }
    170.     /**
    171.      * 二级buffer的刷新状态
    172.      */
    173.     enum RefreshEnum
    174.     {
    175.         /**
    176.          * 无状态
    177.          */
    178.         NON,
    179.         /**
    180.          * 准备创建异步
    181.          */
    182.         READY_ASYNC,
    183.         /**
    184.          * 准备rpc获取远端数据
    185.          */
    186.         READY_RPC,
    187.         /**
    188.          * 完成
    189.          */
    190.         FINISH
    191.     }
    192. }