工作原理
数据建模从系统必须回答的问题和绝不能违反的规则出发。在此基础上识别实体、选择标识符、确定哪些关系是强制的,并把不变量编码为主键、外键、唯一约束、检查约束或可空性。模型不只是一张图,它还定义归属、生命周期、保留和删除行为。
不同存储提供不同的原语。关系型引擎偏好表、连接和约束;文档存储偏好嵌入式聚合与应用层校验;宽列存储把分区键当作分布和查询的单位;图存储把边建模为一等记录。推理方式相似,但可用的约束能力差别很大。
从访问模式出发
在选择键之前,先列出主要的读写路径。一个写入分布均匀却无法回答主查询的分区键会迫使全表扫描;一个纸面上很优雅的范式模型,可能为了首页需要六次连接。记录基数、预期增长以及每条路径所需的一致性。
权衡
刚性模式能尽早拦截坏数据、简化报表,但每次变更都变成迁移。灵活模式能快速接纳新形状,却把校验推给服务层,并常常产生同一实体的多个版本。冗余提高读取速度,同时带来更新异常。没有免费的选项,只能选择自己能够运维的失败模式。
何时使用
任何比单个请求存活更久的状态都值得认真建模,尤其是多个团队共同读写的状态。对于短期缓存或原型,最小模型就足够,但要记录假设,以便日后重新审视。
误解与示例
常见误解是应用上线后模型就完成了。模型会随产品演进,迁移本身就是设计的一部分。例如订单在购买时保存收货地址快照:如果只引用客户当前地址,客户搬家就会静默改写历史。把地址复制到订单中保留快照,这是有意的反规范化,而不是意外。
代码示例
package main
import (
"fmt"
"math/bits"
)
type Bitset []uint64
func NewBitset(size int) Bitset {
if size < 0 {
panic("negative bitset size")
}
return make(Bitset, (size+63)/64)
}
func (b Bitset) Set(index int) {
b[index/64] |= uint64(1) << (uint(index) % 64)
}
func (b Bitset) Clear(index int) {
b[index/64] &^= uint64(1) << (uint(index) % 64)
}
func (b Bitset) Has(index int) bool {
return b[index/64]&(uint64(1)<<(uint(index)%64)) != 0
}
func (b Bitset) Union(other Bitset) Bitset {
size := len(b)
if len(other) > size {
size = len(other)
}
out := make(Bitset, size)
for i := range out {
var left, right uint64
if i < len(b) {
left = b[i]
}
if i < len(other) {
right = other[i]
}
out[i] = left | right
}
return out
}
func (b Bitset) Intersect(other Bitset) Bitset {
size := len(b)
if len(other) < size {
size = len(other)
}
out := make(Bitset, size)
for i := range out {
out[i] = b[i] & other[i]
}
return out
}
func (b Bitset) Count() int {
total := 0
for _, word := range b {
total += bits.OnesCount64(word)
}
return total
}
func (b Bitset) Slice() []int {
out := make([]int, 0, b.Count())
for wordIndex, word := range b {
for word != 0 {
bit := bits.TrailingZeros64(word)
out = append(out, wordIndex*64+bit)
word &^= uint64(1) << uint(bit)
}
}
return out
}
func main() {
active := NewBitset(80)
for _, id := range []int{1, 5, 64, 79} {
active.Set(id)
}
blocked := NewBitset(80)
blocked.Set(5)
blocked.Set(9)
fmt.Println("count:", active.Count())
fmt.Println("has 64:", active.Has(64), "has 65:", active.Has(65))
fmt.Println("members:", active.Slice())
fmt.Println("union count:", active.Union(blocked).Count())
fmt.Println("intersect:", active.Intersect(blocked).Slice())
active.Clear(64)
fmt.Println("after clear:", active.Count())
}
from collections.abc import Iterator
from typing import Generic, TypeVar
T = TypeVar("T")
class RingBuffer(Generic[T]):
def __init__(self, capacity: int) -> None:
if capacity < 1:
raise ValueError("capacity must be positive")
self._slots: list[T | None] = [None] * capacity
self._head = 0
self._count = 0
@property
def capacity(self) -> int:
return len(self._slots)
def __len__(self) -> int:
return self._count
def push(self, item: T) -> T | None:
evicted: T | None = None
if self._count == self.capacity:
evicted = self._slots[self._head]
self._head = (self._head + 1) % self.capacity
index = (self._head + self._count - 1) % self.capacity
else:
index = (self._head + self._count) % self.capacity
self._slots[index] = item
if self._count < self.capacity:
self._count += 1
return evicted
def items(self) -> Iterator[T]:
for offset in range(self._count):
value = self._slots[(self._head + offset) % self.capacity]
if value is not None:
yield value
def latest(self, count: int) -> list[T]:
if count < 0:
raise ValueError("count must not be negative")
if count == 0:
return []
return list(self.items())[-count:]
def main() -> None:
buffer: RingBuffer[str] = RingBuffer(3)
for event in ["a", "b", "c"]:
print("push", event, "evicted", buffer.push(event))
print("full:", list(buffer.items()))
print("push d evicted:", buffer.push("d"))
print("after:", list(buffer.items()))
print("latest 2:", buffer.latest(2))
print("length:", len(buffer))
if __name__ == "__main__":
main()