本文目录导读：
我来为您介绍几种生成消息唯一标识（Unique ID）的方法：
UUID/GUID 实现
Python 实现
import uuid
import hashlib
# 生成UUID
def generate_uuid():
return str(uuid.uuid4())
# 基于消息内容生成唯一ID
def generate_message_id(content, timestamp=None):
base = f"{content}_{timestamp or uuid.uuid1()}"
return hashlib.md5(base.encode()).hexdigest()
# 使用示例
print(generate_uuid()) # e.g., "550e8400-e29b-41d4-a716-446655440000"
JavaScript 实现
function generateUUID() {
return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {
var r = Math.random() * 16 | 0,
v = c == 'x' ? r : (r & 0x3 | 0x8);
return v.toString(16);
});
}
生成
function generateMessageId(content) {
const hash = btoa(content + Date.now());
return hash.substring(0, 32);
}
Snowflake ID (雪花算法)
适用于分布式系统,生成唯一且有序的ID。
Python 实现
import time
import threading
class SnowflakeIdGenerator:
def __init__(self, worker_id=0, datacenter_id=0):
self.worker_id = worker_id
self.datacenter_id = datacenter_id
self.sequence = 0
self.last_timestamp = -1
self.lock = threading.Lock()
# 设置时间纪元 (2020-01-01)
self.epoch = 1577836800000
# 各部分的位数
self.worker_id_bits = 5
self.datacenter_id_bits = 5
self.sequence_bits = 12
# 计算最大值
self.max_worker_id = (1 << self.worker_id_bits) - 1
self.max_datacenter_id = (1 << self.datacenter_id_bits) - 1
self.max_sequence = (1 << self.sequence_bits) - 1
# 位移量
self.worker_id_shift = self.sequence_bits
self.datacenter_id_shift = self.sequence_bits + self.worker_id_bits
self.timestamp_shift = self.sequence_bits + self.worker_id_bits + self.datacenter_id_bits
def generate_id(self):
with self.lock:
timestamp = int(time.time() * 1000) - self.epoch
if timestamp < self.last_timestamp:
raise Exception("Clock moved backwards")
if timestamp == self.last_timestamp:
self.sequence = (self.sequence + 1) & self.max_sequence
if self.sequence == 0:
timestamp = self.wait_next_millis()
else:
self.sequence = 0
self.last_timestamp = timestamp
return ((timestamp << self.timestamp_shift) |
(self.datacenter_id << self.datacenter_id_shift) |
(self.worker_id << self.worker_id_shift) |
self.sequence)
def wait_next_millis(self):
timestamp = int(time.time() * 1000) - self.epoch
while timestamp <= self.last_timestamp:
timestamp = int(time.time() * 1000) - self.epoch
return timestamp
# 使用示例
generator = SnowflakeIdGenerator(worker_id=1, datacenter_id=1)
message_id = generator.generate_id()
print(message_id) # e.g., 1234567890123456
基于时间戳+随机数的简单实现
Python 实现
import time
import random
import string
def generate_simple_id():
"""生成基于时间戳+随机数的ID"""
timestamp = int(time.time() * 1000)
random_part = random.randint(100000, 999999)
return f"{timestamp}{random_part}"
def generate_alphanumeric_id(length=16):
"""生成字母数字组合的ID"""
chars = string.ascii_letters + string.digits
return ''.join(random.choice(chars) for _ in range(length))
def generate_base64_id():
"""生成Base64编码的ID"""
import base64
import os
random_bytes = os.urandom(16)
return base64.urlsafe_b64encode(random_bytes).decode().rstrip('=')
JavaScript 实现
// 简单的ID生成器
function generateSimpleId() {
return Date.now().toString(36) + Math.random().toString(36).substr(2);
}
// 更复杂的版本
function generateUniqueId(prefix = 'msg') {
const timestamp = Date.now().toString(36);
const randomPart = Math.random().toString(36).substr(2, 9);
return `${prefix}_${timestamp}_${randomPart}`;
}
Redis 自增ID
适用于需要递增序列的场景。
import redis
class RedisIdGenerator:
def __init__(self, redis_client, key='message_id'):
self.redis = redis_client
self.key = key
def generate_id(self):
# 返回自增的ID
return self.redis.incr(self.key)
def generate_batch(self, count=10):
# 批量生成
return self.redis.incrby(self.key, count)
# 使用示例
r = redis.Redis(host='localhost', port=6379, db=0)
generator = RedisIdGenerator(r)
message_id = generator.generate_id()
完整的消息唯一标识生成器
import hashlib
import json
import time
from datetime import datetime
class MessageIdGenerator:
def __init__(self, strategy='uuid'):
self.strategy = strategy
self.counter = 0
def generate(self, content=None, metadata=None):
"""生成消息唯一标识"""
if self.strategy == 'uuid':
return self._uuid_strategy()
elif self.strategy == 'hash':
return self._hash_strategy(content, metadata)
elif self.strategy == 'sequential':
return self._sequential_strategy()
elif self.strategy == 'compound':
return self._compound_strategy(content)
else:
return self._default_strategy()
def _uuid_strategy(self):
import uuid
return str(uuid.uuid4())
def _hash_strategy(self, content, metadata):
if content is None:
content = str(time.time())
if metadata:
content += json.dumps(metadata, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:32]
def _sequential_strategy(self):
self.counter += 1
timestamp = int(time.time() * 1000)
return f"{timestamp}_{self.counter:06d}"
def _compound_strategy(self, content):
timestamp = datetime.now().strftime('%Y%m%d%H%M%S%f')
content_hash = hashlib.md5(content.encode()).hexdigest()[:8] if content else ''
return f"MSG_{timestamp}_{content_hash}"
# 使用示例
generator = MessageIdGenerator(strategy='compound')
# 生成消息唯一标识
message = "Hello, World!"
msg_id = generator.generate(content=message)
print(f"Message ID: {msg_id}")
# 带元数据的
metadata = {"user": "john", "channel": "email"}
msg_id = generator.generate(content=message, metadata=metadata)
print(f"Message with metadata ID: {msg_id}")
选择建议
- UUID - 最简单，适合大部分场景
- Snowflake - 分布式系统，需要有序ID
- 时间戳+随机数 - 简单易懂，适合小规模应用
- Redis自增 - 需要严格递增，有Redis环境
- 哈希 - 需要内容去重
根据您的具体需求选择最合适的实现方式。
