利用Redis实现全局唯一ID

利用Redis实现全局唯一ID

背景

场景分析：如果我们的id具有太明显的规则，用户或者说商业对手很容易猜测出来我们的一些敏感信息，比如商城在一天时间内，卖出了多少单，这明显不合适。

场景分析二：随着我们商城规模越来越大，mysql的单表的容量不宜超过500W，数据量过大之后，我们要进行拆库拆表，但拆分表了之后，他们从逻辑上讲他们是同一张表，所以他们的id是不能一样的， 于是乎我们需要保证id的唯一性。

全局ID生成器，是一种在分布式系统下用来生成全局唯一ID的工具，一般要满足下列特性：

• 唯一性
• 高可用
• 递增性
• 安全性
• 高性能

为了增加ID的安全性，我们可以不直接使用Redis自增的数据，而是拼接一些其他的信息，下图是一个long类型的64位示意图，ID设计示意图

D的组成部分：符号位：1bit，永远为0

时间戳：31bit，以秒为单位，可以使用69年

序列号：32bit，秒内的计数器，支持每秒产生2^32个不同ID

实现

@Component
public class UniqueIdWorker {
    // 我们定义2023.01.01为开始计数时间
    /**
     * 开始时间戳
     */
    private static final long BEGIN_TIMESTAMP = 1672531200L;

    /**
     * 获取2023.1.1 秒的时间戳
     *
     * @param args
     */
    public static void main(String[] args) {
        System.out.println(LocalDateTime.of(2023, 1, 1, 0, 0).toEpochSecond(ZoneOffset.UTC));
    }

    /**
     * 序列号位数
     */
    private static final int COUNT_BITS = 32;

    private RedisTemplate<String, Object> redisTemplate;

    public UniqueIdWorker(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 生成ID的函数
     *
     * @param keyPrefix 存储在Redis中的识别名
     * @return
     */
    public long nextId(String keyPrefix) {
        // 1. 生成时间戳
        LocalDateTime now = LocalDateTime.now();
        long timestamp = now.toEpochSecond(ZoneOffset.UTC) - BEGIN_TIMESTAMP;
        // 2. a. 生成序列号为了就是在Redis中存储识别
        String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
        // 2. b. 自增长
        long count = redisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
        // 3. 拼接并返回, 这里进行了位运算，首先把timestamp左移了32位，然后与count进行或运算，这样就能得到拼接结果，具体可以用二进制试试。
        return timestamp << 32 | count;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

测试代码

关于countdownlatch

countdownlatch名为信号枪：主要的作用是同步协调在多线程的等待于唤醒问题

我们如果没有CountDownLatch ，那么由于程序是异步的，当异步程序没有执行完时，主线程就已经执行完了，然后我们期望的是分线程全部走完之后，主线程再走，所以我们此时需要使用到CountDownLatch

CountDownLatch 中有两个最重要的方法

1、countDown

2、await

await 方法 是阻塞方法，我们担心分线程没有执行完时，main线程就先执行，所以使用await可以让main线程阻塞，那么什么时候main线程不再阻塞呢？当CountDownLatch 内部维护的 变量变为0时，就不再阻塞，直接放行，那么什么时候CountDownLatch 维护的变量变为0 呢，我们只需要调用一次countDown ，内部变量就减少1，我们让分线程和变量绑定， 执行完一个分线程就减少一个变量，当分线程全部走完，CountDownLatch 维护的变量就是0，此时await就不再阻塞，统计出来的时间也就是所有分线程执行完后的时间。

@Resource
UniqueIdWorker uniqueIdWorker;
@Resource
RedisTemplate<String, Object> redisTemplate;
@Test
void testIdWorker() throws Exception{
    CountDownLatch latch = new CountDownLatch(300);
    Runnable task = () -> {
        for (int i = 0; i < 100; i++) {
            long id = uniqueIdWorker.nextId("order");
            System.out.println("id = " + id);
        }
        latch.countDown();
    };
    ExecutorService es = Executors.newFixedThreadPool(100); //线程池，把任务提交，然后分配任务
    long begin = System.currentTimeMillis();
    for (int i = 0; i < 300; i++) {
        es.submit(task);
    }
    latch.await();
    long end = System.currentTimeMillis();
    System.out.println("time = " + (end - begin));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

结果对比
打印结果

数据二进制对比结果

Redis数据