Tong's Digital Garden
welcome to Tong's Digital Garden
bigguaiwutong@qq.com
雪花算法
by tong
问题
由于业务没有提前告知进行多次全量用户推送，9 月 25 日中午我们服务告警，CPU 使用率持续高于 90%，在 Grafana 上查看后，发现每台机器 QPS 已经超过 100。我们进行紧急扩容，首先由两台扩容到 4 台，发现 QPS 仍在不断增长，进行第二次扩容，4 台扩容至 6 台。此时 CPU 的使用率恢复到平稳，大约（50%）。但是告警群内不断出现主键冲突，我们使用的是雪花算法，从 Mybatis -plus 从拷贝过来的 Id 生成器。在服务平稳下掉两台服务器后，不再出现报错。
雪花算法是什么
雪花算法（Snowflake）是一种生成分布式全局唯一 ID 的算法，生成的 ID 称为Snowflake ID 或snowflake。这种算法由 Twitter 创建，并用于推文的 ID。Discord 和 Instagram 等其他公司采用了修改后的版本。
一个 Snowflake ID 有64位元。
格式
- 前 41 位是时间戳，表示了自选定的时期以来的毫秒数。
- 接下来的 10 位代表计算机 ID，防止冲突。
- 其余 12 位代表每台机器上生成 ID 的序列号，这允许在同一毫秒内创建多个 Snowflake ID。
常用的实现方式
public class Sequence {
private static final Log logger = LogFactory.getLog(Sequence.class);
/**
* 时间起始标记点，作为基准，一般取系统的最近时间（一旦确定不能变动）
*/
private static final long twepoch = 1288834974657L;
/**
* 机器标识位数
*/
private final long workerIdBits = 5L;
private final long datacenterIdBits = 5L;
private final long maxWorkerId = -1L ^ (-1L << workerIdBits);
private final long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);
/**
* 毫秒内自增位
*/
private final long sequenceBits = 12L;
private final long workerIdShift = sequenceBits;
private final long datacenterIdShift = sequenceBits + workerIdBits;
/**
* 时间戳左移动位
*/
private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;
private final long sequenceMask = -1L ^ (-1L << sequenceBits);
private final long workerId;
/**
* 数据标识 ID 部分
*/
private final long datacenterId;
/**
* 并发控制
*/
private long sequence = 0L;
/**
* 上次生产 ID 时间戳
*/
private long lastTimestamp = -1L;
/**
* IP 地址
*/
private InetAddress inetAddress;
public Sequence(InetAddress inetAddress) {
this.inetAddress = inetAddress;
this.datacenterId = getDatacenterId(maxDatacenterId);
this.workerId = getMaxWorkerId(datacenterId, maxWorkerId);
initLog();
}
private void initLog() {
if (logger.isDebugEnabled()) {
logger.debug("Initialization Sequence datacenterId:" + this.datacenterId + " workerId:" + this.workerId);
}
}
/**
* 有参构造器
* @param workerId 工作机器 ID
* @param datacenterId 序列号
*/
public Sequence(long workerId, long datacenterId) {
Assert.isFalse(workerId > maxWorkerId || workerId < 0,
String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));
Assert.isFalse(datacenterId > maxDatacenterId || datacenterId < 0,
String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
this.workerId = workerId;
this.datacenterId = datacenterId;
initLog();
}
/**
* 获取 maxWorkerId
*/
protected long getMaxWorkerId(long datacenterId, long maxWorkerId) {
StringBuilder mpid = new StringBuilder();
mpid.append(datacenterId);
String name = ManagementFactory.getRuntimeMXBean().getName();
if (StringUtils.isNotBlank(name)) {
/*
* GET jvmPid
*/
mpid.append(name.split(StringPool.AT)[0]);
}
/*
* MAC + PID 的 hashcode 获取16个低位
*/
return (mpid.toString().hashCode() & 0xffff) % (maxWorkerId + 1);
}
/**
* 数据标识id部分
*/
protected long getDatacenterId(long maxDatacenterId) {
long id = 0L;
try {
if (null == this.inetAddress) {
this.inetAddress = InetAddress.getLocalHost();
}
NetworkInterface network = NetworkInterface.getByInetAddress(this.inetAddress);
if (null == network) {
id = 1L;
} else {
byte[] mac = network.getHardwareAddress();
if (null != mac) {
id = ((0x000000FF & (long) mac[mac.length - 2]) | (0x0000FF00 & (((long) mac[mac.length - 1]) << 8))) >> 6;
id = id % (maxDatacenterId + 1);
}
}
} catch (Exception e) {
logger.warn(" getDatacenterId: " + e.getMessage());
}
return id;
}
/**
* 获取下一个 ID
*
* @return 下一个 ID
*/
public synchronized long nextId() {
long timestamp = timeGen();
//闰秒
if (timestamp < lastTimestamp) {
long offset = lastTimestamp - timestamp;
if (offset <= 5) {
try {
wait(offset << 1);
timestamp = timeGen();
if (timestamp < lastTimestamp) {
throw new RuntimeException(String.format("Clock moved backwards. Refusing to generate id for %d milliseconds", offset));
}
} catch (Exception e) {
throw new RuntimeException(e);
}
} else {
throw new RuntimeException(String.format("Clock moved backwards. Refusing to generate id for %d milliseconds", offset));
}
}
if (lastTimestamp == timestamp) {
// 相同毫秒内，序列号自增
sequence = (sequence + 1) & sequenceMask;
if (sequence == 0) {
// 同一毫秒的序列数已经达到最大
timestamp = tilNextMillis(lastTimestamp);
}
} else {
// 不同毫秒内，序列号置为 1 - 2 随机数
sequence = ThreadLocalRandom.current().nextLong(1, 3);
}
lastTimestamp = timestamp;
// 时间戳部分 | 数据中心部分 | 机器标识部分 | 序列号部分
return ((timestamp - twepoch) << timestampLeftShift)
| (datacenterId << datacenterIdShift)
| (workerId << workerIdShift)
| sequence;
}
protected long tilNextMillis(long lastTimestamp) {
long timestamp = timeGen();
while (timestamp <= lastTimestamp) {
timestamp = timeGen();
}
return timestamp;
}
protected long timeGen() {
return SystemClock.now();
}
/**
* 反解id的时间戳部分
*/
public static long parseIdTimestamp(long id) {
return (id>>22)+twepoch;
}
}
问题分析
首先排除时间回拨的问题，在这个生成 Id 的方法里，会进行时间比较，通过 tilNextMillis() 方法避免时间回拨导致的冲突。
这个生成 id 的方法是线程安全的。但是集群部署的微服务，同一时刻时间戳相同，序列号的生成是递增的并且起始的序列号接近，请求轮训的情况下，不同机器上的序列号可能出现重复。此时，如果 datacenterId 和 workId 相同就会导致 Id 重复。如果不重视该问题，会出现数据入库失败，从而丢失关键数据的问题。
代码分析
然后看一下 datacenterId 和 workId 生成的过程
public Sequence(InetAddress inetAddress) {
this.inetAddress = inetAddress;
this.datacenterId = getDatacenterId(maxDatacenterId);
this.workerId = getMaxWorkerId(datacenterId, maxWorkerId);
initLog();
}
protected long getMaxWorkerId(long datacenterId, long maxWorkerId) {
StringBuilder mpid = new StringBuilder();
mpid.append(datacenterId);
String name = ManagementFactory.getRuntimeMXBean().getName();
if (StringUtils.isNotBlank(name)) {
/*
* GET jvmPid
*/
mpid.append(name.split(StringPool.AT)[0]);
}
/*
* MAC + PID 的 hashcode 获取16个低位
*/
return (mpid.toString().hashCode() & 0xffff) % (maxWorkerId + 1);
}
/**
* 数据标识id部分
*/
protected long getDatacenterId(long maxDatacenterId) {
long id = 0L;
try {
if (null == this.inetAddress) {
this.inetAddress = InetAddress.getLocalHost();
}
NetworkInterface network = NetworkInterface.getByInetAddress(this.inetAddress);
if (null == network) {
id = 1L;
} else {
byte[] mac = network.getHardwareAddress();
if (null != mac) {
id = ((0x000000FF & (long) mac[mac.length - 2]) | (0x0000FF00 & (((long) mac[mac.length - 1]) << 8))) >> 6;
id = id % (maxDatacenterId + 1);
}
}
} catch (Exception e) {
logger.warn(" getDatacenterId: " + e.getMessage());
}
return id;
}
- DatacenterId 的生成：首先需要得到 InetAddress 地址，再读取宿主机的硬件地址。当这两个有任意一个读不到时，datacenterId 就会变成一个默认值（读不到 InetAddress 时默认为 1，读不到硬件地址时默认为 0），此时 datacenterId 相同。当都能读取到时，将取出 mac 地址后两位进行位运算，若后两位相同，也会出现 datacenterId 相同的情况。
- WorkId 的生成：他的结果与 DatacenterId 和容器中的 pid 相关，虽然服务恢复正常后我们下掉了另外两台机器不再发生冲突，但是我们可以用现在运行的机器进行验证，结果在线上环境中我们的 3 个 pod 就有 2 个 pod 的 pid 相同。换句话说，在 DataCenterId 相同的情况下，只要 pid 相同，高并发下极大概率 ID 冲突。
结论
此算法依赖 MAC 地址后两位散列来保证 DataCenterId 不重复，同时由于在我们服务中 pid 极为容易重复，同一服务在数量较多或者未能获取到硬件地址时，DataCenterId 和 WorkId 相同概率极高。
解决方案
手动设置
- 在 CI 流程中，手动指定要启动的 Pod 的 datacenterId 和 workId，写入环境变量，服务内部接收环境变量并用此参数初始化 Sequence 对象，由于我们的机器是自动扩容的，显然不能使用这种方式
自动化方案
- 在应用启动时，通过 Lua 脚本去 Redis 获取标识位。DataCenterId 和 workerId 的获取与自增在 Lua 脚本中完成，调用返回后就是可用的标示位。lua脚本雪花算法但是引入了一个新的组件后，将会有新的风险，当 redis 宕机后怎么办，或者 redis 抖动怎么办
- 自己实现一个简单的 workId 分配器 雪花算法实现
- 业内有比较成熟的 Id 生成框架，使用其框架
tags: 编程 - 分布式 - ID