【系统架构设计百科】微信架构：十亿用户的即时通信挑战

// Mars STN 连接策略伪代码
class ConnectionManager {
public:
    void startConnection() {
        // 第一优先级：尝试长连接
        // 长连接使用自定义二进制协议，端口 8080 或 443
        tryLongConnection();

        // 如果长连接在 5 秒内未建立，并行尝试短连接
        scheduleTimeout(5000, [this]() {
            if (!longConnEstablished_) {
                tryShortConnection();
            }
        });
    }

private:
    void tryLongConnection() {
        // 获取服务器 IP 列表（优先使用本地缓存）
        auto ipList = getServerIPs();

        // 并行发起多个 TCP 连接（竞速模式）
        for (const auto& ip : ipList) {
            asyncConnect(ip, 8080, [this](Socket* sock) {
                if (!longConnEstablished_) {
                    longConnEstablished_ = true;
                    activateConnection(sock);
                    // 关闭其他并行连接
                    cancelOtherConnections();
                }
            });
        }
    }

    void tryShortConnection() {
        // 短连接走 HTTPS，兼容性最好
        // 用于发送即时消息，不保持长连接
        httpPost(shortConnUrl_, payload_);
    }

    std::vector<std::string> getServerIPs() {
        // 三级 IP 获取策略
        // 1. 本地缓存（上次成功连接的 IP）
        auto cached = loadCachedIPs();
        if (!cached.empty()) return cached;

        // 2. HTTP DNS（绕过运营商 DNS 劫持）
        auto httpDns = queryHTTPDNS();
        if (!httpDns.empty()) return httpDns;

        // 3. 系统 DNS（最后兜底）
        return querySystemDNS();
    }

    bool longConnEstablished_ = false;
    std::string shortConnUrl_;
    std::string payload_;
};

class SmartHeartbeat {
public:
    // 心跳间隔范围：最短 60 秒，最长 300 秒
    static constexpr int MIN_INTERVAL = 60;   // 秒
    static constexpr int MAX_INTERVAL = 300;   // 秒
    static constexpr int STEP = 30;            // 步进值

    int computeInterval() {
        if (lastHeartbeatFailed_) {
            // 上次心跳失败，回退到上一个成功的间隔
            currentInterval_ = lastSuccessInterval_;
            failureCount_++;

            if (failureCount_ >= 3) {
                // 连续失败 3 次，使用最小间隔
                currentInterval_ = MIN_INTERVAL;
                failureCount_ = 0;
            }
        } else {
            // 心跳成功，记录当前间隔并尝试增大
            lastSuccessInterval_ = currentInterval_;
            failureCount_ = 0;

            if (currentInterval_ + STEP <= MAX_INTERVAL) {
                currentInterval_ += STEP;
            }
        }
        return currentInterval_;
    }

    void onNetworkChange() {
        // 网络环境变化时（Wi-Fi/4G 切换），重置心跳探测
        currentInterval_ = MIN_INTERVAL;
        lastSuccessInterval_ = MIN_INTERVAL;
        failureCount_ = 0;
    }

private:
    int currentInterval_ = MIN_INTERVAL;
    int lastSuccessInterval_ = MIN_INTERVAL;
    int failureCount_ = 0;
    bool lastHeartbeatFailed_ = false;
};

// Xlog mmap 写入的简化实现
class MmapLogWriter {
public:
    bool init(const std::string& logPath, size_t bufferSize) {
        fd_ = open(logPath.c_str(), O_RDWR | O_CREAT, 0644);
        if (fd_ < 0) return false;

        // 预分配文件空间
        ftruncate(fd_, bufferSize);

        // mmap 映射
        buffer_ = (char*)mmap(nullptr, bufferSize,
                              PROT_READ | PROT_WRITE,
                              MAP_SHARED, fd_, 0);
        if (buffer_ == MAP_FAILED) return false;

        bufferSize_ = bufferSize;
        writePos_ = 0;
        return true;
    }

    void write(const char* data, size_t len) {
        // 压缩数据
        auto compressed = zlibCompress(data, len);

        // 加密
        auto encrypted = encrypt(compressed);

        // 写入 mmap 缓冲区（等价于写内存）
        if (writePos_ + encrypted.size() <= bufferSize_) {
            memcpy(buffer_ + writePos_,
                   encrypted.data(), encrypted.size());
            writePos_ += encrypted.size();
        } else {
            // 缓冲区满，切换到新文件
            rotateLogFile();
        }
    }

private:
    int fd_;
    char* buffer_;
    size_t bufferSize_;
    size_t writePos_;

    void rotateLogFile();
    std::vector<char> zlibCompress(const char* data, size_t len);
    std::vector<char> encrypt(const std::vector<char>& data);
};

/**
 * SeqSvr 段分配策略的简化实现
 * 每次从存储层分配一个步长为 10000 的段
 * 段内的序列号分配在内存中完成，无需访问存储
 */
public class SeqAllocator {

    private static final long SECTION_SIZE = 10000L;

    private final ConcurrentHashMap<String, SeqSection> sections
        = new ConcurrentHashMap<>();

    private final PaxosStoreClient storeClient;

    public SeqAllocator(PaxosStoreClient storeClient) {
        this.storeClient = storeClient;
    }

    public long allocSeq(String userId) {
        SeqSection section = sections.computeIfAbsent(
            userId, k -> loadNewSection(k)
        );

        synchronized (section) {
            if (section.current < section.max) {
                return ++section.current;
            }
            // 当前段用完，加载新段
            SeqSection newSection = loadNewSection(userId);
            sections.put(userId, newSection);
            return ++newSection.current;
        }
    }

    private SeqSection loadNewSection(String userId) {
        // 从存储层原子性地分配一个新段
        // CAS 操作保证并发安全
        long currentMax = storeClient.getAndAdd(
            "seq:" + userId, SECTION_SIZE
        );
        SeqSection section = new SeqSection();
        section.current = currentMax;
        section.max = currentMax + SECTION_SIZE;
        return section;
    }

    private static class SeqSection {
        long current;
        long max;
    }
}

class MessageSyncClient:
    """客户端消息同步逻辑"""

    PAGE_SIZE = 20  # 每次拉取 20 条消息

    def __init__(self):
        self.local_seq = self.load_local_seq()

    def on_new_message_notify(self, server_seq: int):
        """收到服务器推送的新消息通知"""
        if server_seq <= self.local_seq:
            return  # 没有新消息

        self.sync_messages(server_seq)

    def sync_messages(self, target_seq: int):
        """增量同步消息，分页拉取"""
        current_seq = self.local_seq

        while current_seq < target_seq:
            # 分页拉取
            response = self.fetch_messages(
                from_seq=current_seq,
                limit=self.PAGE_SIZE
            )

            if not response.messages:
                break

            # 处理并存储消息
            for msg in response.messages:
                self.save_message_to_local_db(msg)
                current_seq = max(current_seq, msg.sequence)

            # 更新本地 sequence
            self.local_seq = current_seq
            self.save_local_seq(current_seq)

            # 通知 UI 层刷新
            self.notify_ui_update(response.messages)

    def fetch_messages(self, from_seq: int, limit: int):
        """从服务器拉取消息"""
        request = SyncRequest(
            user_id=self.user_id,
            last_seq=from_seq,
            count=limit
        )
        return self.network_client.sync(request)

    def load_local_seq(self) -> int:
        """从本地数据库加载最大 sequence"""
        pass

    def save_local_seq(self, seq: int):
        """持久化最新 sequence"""
        pass

    def save_message_to_local_db(self, msg):
        """存储消息到本地 SQLite 数据库"""
        pass

    def notify_ui_update(self, messages):
        """通知界面刷新"""
        pass

/**
 * 朋友圈时间线的读扩散实现
 */
public class MomentTimelineService {

    private final ContentStore contentStore;
    private final ContactService contactService;
    private final CommentStore commentStore;
    private final CacheService cache;

    /**
     * 获取用户的朋友圈时间线
     * 读扩散：实时聚合所有好友的朋友圈内容
     */
    public List<MomentItem> getTimeline(
            String userId, long beforeTime, int count) {

        // 获取好友列表
        List<String> friendIds = contactService.getFriends(userId);
        friendIds.add(userId);  // 包含自己

        // 从缓存或存储中获取每个好友的最新内容
        PriorityQueue<MomentItem> mergeHeap = new PriorityQueue<>(
            (a, b) -> Long.compare(b.getCreateTime(), a.getCreateTime())
        );

        for (String friendId : friendIds) {
            // 每个好友最多取 count 条（减少不必要的扫描）
            List<MomentItem> items = contentStore.getByUser(
                friendId, beforeTime, count
            );
            mergeHeap.addAll(items);
        }

        // 归并排序取 Top-N
        List<MomentItem> result = new ArrayList<>();
        while (!mergeHeap.isEmpty() && result.size() < count) {
            MomentItem item = mergeHeap.poll();
            // 权限过滤（检查是否对当前用户可见）
            if (isVisible(item, userId)) {
                // 加载评论和点赞（只加载共同好友的）
                item.setComments(
                    commentStore.getVisibleComments(
                        item.getId(), userId
                    )
                );
                result.add(item);
            }
        }
        return result;
    }

    private boolean isVisible(MomentItem item, String viewerId) {
        // 检查朋友圈可见性设置
        // 部分可见、不给谁看等逻辑
        return true; // 简化
    }
}

/**
 * 支付幂等性处理
 * 基于全局唯一的交易号保证同一笔支付只执行一次
 */
public class PaymentService {

    private final TransactionStore txStore;
    private final AccountService accountService;
    private final DistributedLock lock;

    public PayResult pay(PayRequest request) {
        String txId = request.getTransactionId();

        // 第一步：查询是否已处理过
        Transaction existingTx = txStore.findByTxId(txId);
        if (existingTx != null) {
            // 幂等返回：直接返回上次的处理结果
            return buildResultFromTx(existingTx);
        }

        // 第二步：获取分布式锁，防止并发重复执行
        String lockKey = "pay_lock:" + txId;
        if (!lock.tryLock(lockKey, 10, TimeUnit.SECONDS)) {
            throw new PayException("PROCESSING",
                "支付正在处理中，请勿重复提交");
        }

        try {
            // 双重检查
            existingTx = txStore.findByTxId(txId);
            if (existingTx != null) {
                return buildResultFromTx(existingTx);
            }

            // 第三步：执行扣款
            Transaction tx = new Transaction();
            tx.setTxId(txId);
            tx.setStatus("PROCESSING");
            txStore.insert(tx);  // 先写入事务记录

            try {
                accountService.debit(
                    request.getUserId(), request.getAmount()
                );
                tx.setStatus("SUCCESS");
            } catch (Exception e) {
                tx.setStatus("FAILED");
                tx.setErrorMsg(e.getMessage());
            }

            txStore.update(tx);
            return buildResultFromTx(tx);

        } finally {
            lock.unlock(lockKey);
        }
    }

    private PayResult buildResultFromTx(Transaction tx) {
        PayResult result = new PayResult();
        result.setTransactionId(tx.getTxId());
        result.setStatus(tx.getStatus());
        return result;
    }
}

/**
 * 红包金额拆分算法——二倍均值法
 * 保证每个人抢到的金额期望值相等
 */
public class RedPacketSplitter {

    private static final int MIN_AMOUNT = 1; // 最小 1 分钱（单位：分）

    /**
     * 预拆分红包金额
     * @param totalAmount 总金额（单位：分）
     * @param count 红包个数
     * @return 每个红包的金额列表
     */
    public static List<Integer> split(int totalAmount, int count) {
        List<Integer> amounts = new ArrayList<>();
        int remainAmount = totalAmount;
        int remainCount = count;

        Random random = new Random();

        for (int i = 0; i < count - 1; i++) {
            // 二倍均值法
            // 最大值 = 剩余金额 / 剩余人数 * 2
            int maxAmount = (remainAmount / remainCount) * 2;

            // 随机金额范围：[1, maxAmount]
            int amount = random.nextInt(maxAmount - MIN_AMOUNT)
                         + MIN_AMOUNT;

            // 确保剩余金额够分给剩下的人
            if (remainAmount - amount < (remainCount - 1) * MIN_AMOUNT) {
                amount = remainAmount - (remainCount - 1) * MIN_AMOUNT;
            }

            amounts.add(amount);
            remainAmount -= amount;
            remainCount--;
        }

        // 最后一个人拿剩余所有金额
        amounts.add(remainAmount);
        return amounts;
    }
}

// 抢红包请求排队处理
package redpacket

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// GrabRequest 抢红包请求
type GrabRequest struct {
    PacketID string
    UserID   string
    Result   chan GrabResult
}

// GrabResult 抢红包结果
type GrabResult struct {
    Success bool
    Amount  int64
    ErrMsg  string
}

// QueueProcessor 排队处理器
// 将并发请求转换为串行处理，消除数据库并发冲突
type QueueProcessor struct {
    queues map[string]chan *GrabRequest
    mu     sync.RWMutex
}

func NewQueueProcessor() *QueueProcessor {
    return &QueueProcessor{
        queues: make(map[string]chan *GrabRequest),
    }
}

// Submit 提交抢红包请求
func (p *QueueProcessor) Submit(
    ctx context.Context, req *GrabRequest,
) GrabResult {
    queue := p.getOrCreateQueue(req.PacketID)

    req.Result = make(chan GrabResult, 1)

    select {
    case queue <- req:
        // 成功入队，等待结果
    case <-ctx.Done():
        return GrabResult{
            Success: false,
            ErrMsg:  "请求超时",
        }
    }

    select {
    case result := <-req.Result:
        return result
    case <-ctx.Done():
        return GrabResult{
            Success: false,
            ErrMsg:  "处理超时",
        }
    }
}

// getOrCreateQueue 为每个红包创建独立的处理队列
func (p *QueueProcessor) getOrCreateQueue(
    packetID string,
) chan *GrabRequest {
    p.mu.RLock()
    q, exists := p.queues[packetID]
    p.mu.RUnlock()

    if exists {
        return q
    }

    p.mu.Lock()
    defer p.mu.Unlock()

    // 双重检查
    if q, exists = p.queues[packetID]; exists {
        return q
    }

    q = make(chan *GrabRequest, 1000)
    p.queues[packetID] = q

    // 启动消费者协程
    go p.consume(packetID, q)

    return q
}

// consume 串行消费队列中的请求
func (p *QueueProcessor) consume(
    packetID string, queue chan *GrabRequest,
) {
    timeout := time.NewTimer(30 * time.Minute)
    defer timeout.Stop()

    for {
        select {
        case req := <-queue:
            result := processGrab(packetID, req.UserID)
            req.Result <- result
        case <-timeout.C:
            // 红包超时未抢完，清理队列
            p.mu.Lock()
            delete(p.queues, packetID)
            p.mu.Unlock()
            return
        }
    }
}

func processGrab(packetID, userID string) GrabResult {
    // 实际的抢红包逻辑
    fmt.Printf("处理抢红包: packet=%s, user=%s\n",
        packetID, userID)
    return GrabResult{Success: true, Amount: 100}
}

import redis
import json

class RedPacketCache:
    """红包缓存层，使用 Redis 管理红包状态"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def create_packet(
        self, packet_id: str, amounts: list, expire_seconds: int = 86400
    ):
        """创建红包：将预拆分的金额存入 Redis"""
        key = f"rp:{packet_id}"

        # 使用 List 存储预拆分的金额
        pipe = self.redis.pipeline()
        for amount in amounts:
            pipe.rpush(key, amount)
        pipe.expire(key, expire_seconds)

        # 记录已抢用户集合
        grab_key = f"rp_grabbed:{packet_id}"
        pipe.expire(grab_key, expire_seconds)
        pipe.execute()

    def try_grab(self, packet_id: str, user_id: str):
        """
        尝试抢红包
        使用 Redis 原子操作保证并发安全
        返回 (success, amount)
        """
        grab_key = f"rp_grabbed:{packet_id}"

        # 检查是否已抢过（幂等性）
        if self.redis.sismember(grab_key, user_id):
            return False, 0

        # 原子操作：弹出一个金额
        key = f"rp:{packet_id}"
        amount = self.redis.lpop(key)

        if amount is None:
            # 红包已抢完
            return False, 0

        # 记录已抢用户
        self.redis.sadd(grab_key, user_id)

        return True, int(amount)

/**
 * PaxosStore 写入流程的简化示意
 */
public class PaxosWriter {

    private final List<ReplicaNode> replicas;
    private final int quorum;

    public PaxosWriter(List<ReplicaNode> replicas) {
        this.replicas = replicas;
        // 多数派：3 副本需要 2 个确认
        this.quorum = replicas.size() / 2 + 1;
    }

    public boolean write(String key, byte[] value) {
        // Phase 1: Prepare
        long proposalId = generateProposalId();
        int prepareOk = 0;

        for (ReplicaNode replica : replicas) {
            try {
                boolean accepted = replica.prepare(proposalId);
                if (accepted) prepareOk++;
            } catch (Exception e) {
                // 副本不可用，跳过
            }
        }

        if (prepareOk < quorum) {
            return false;  // Prepare 未达到多数派
        }

        // Phase 2: Accept
        int acceptOk = 0;
        for (ReplicaNode replica : replicas) {
            try {
                boolean accepted = replica.accept(
                    proposalId, key, value
                );
                if (accepted) acceptOk++;
            } catch (Exception e) {
                // 副本不可用，跳过
            }
        }

        return acceptOk >= quorum;
    }

    private long generateProposalId() {
        // 生成全局递增的提案 ID
        // 通常使用时间戳 + 节点 ID 的组合
        return System.nanoTime();
    }
}

/**
 * 请求合并器：将短时间内对同一红包的多次请求合并处理
 * 减少对下游存储的并发访问
 */
public class RequestMerger {

    // 合并窗口：5 毫秒
    private static final long MERGE_WINDOW_MS = 5;

    // 每个红包的请求缓冲区
    private final ConcurrentHashMap<String, RequestBuffer> buffers
        = new ConcurrentHashMap<>();

    public CompletableFuture<GrabResult> submit(GrabRequest request) {
        String packetId = request.getPacketId();
        CompletableFuture<GrabResult> future = new CompletableFuture<>();

        RequestBuffer buffer = buffers.computeIfAbsent(
            packetId,
            k -> new RequestBuffer(k, this::processBatch)
        );

        buffer.add(request, future);
        return future;
    }

    /**
     * 批量处理同一个红包的多个抢夺请求
     */
    private void processBatch(
            String packetId,
            List<Pair<GrabRequest, CompletableFuture<GrabResult>>> batch) {

        // 一次性从 Redis 获取红包状态
        RedPacketState state = redisClient.getState(packetId);

        // 已抢完的请求直接返回失败
        if (state.getRemainCount() <= 0) {
            for (var pair : batch) {
                pair.getSecond().complete(
                    new GrabResult(false, 0, "红包已抢完")
                );
            }
            return;
        }

        // 过滤已抢过的用户
        Set<String> grabbedUsers = redisClient.getGrabbedUsers(packetId);
        int successCount = 0;

        for (var pair : batch) {
            GrabRequest req = pair.getFirst();
            CompletableFuture<GrabResult> future = pair.getSecond();

            if (grabbedUsers.contains(req.getUserId())) {
                future.complete(
                    new GrabResult(false, 0, "已抢过")
                );
                continue;
            }

            if (successCount >= state.getRemainCount()) {
                future.complete(
                    new GrabResult(false, 0, "红包已抢完")
                );
                continue;
            }

            // 执行抢红包逻辑
            long amount = redisClient.popAmount(packetId);
            if (amount > 0) {
                redisClient.recordGrab(packetId, req.getUserId());
                future.complete(new GrabResult(true, amount, ""));
                successCount++;
            } else {
                future.complete(
                    new GrabResult(false, 0, "红包已抢完")
                );
            }
        }
    }

    /**
     * 请求缓冲区：在合并窗口内收集请求
     */
    static class RequestBuffer {
        private final String packetId;
        private final BiConsumer<String,
            List<Pair<GrabRequest, CompletableFuture<GrabResult>>>>
            processor;
        private final List<Pair<GrabRequest,
            CompletableFuture<GrabResult>>> pending
            = new ArrayList<>();
        private ScheduledFuture<?> flushTask;

        RequestBuffer(String packetId,
                BiConsumer<String,
                    List<Pair<GrabRequest,
                        CompletableFuture<GrabResult>>>> processor) {
            this.packetId = packetId;
            this.processor = processor;
        }

        synchronized void add(GrabRequest request,
                CompletableFuture<GrabResult> future) {
            pending.add(Pair.of(request, future));
            if (flushTask == null) {
                flushTask = scheduler.schedule(
                    this::flush,
                    MERGE_WINDOW_MS,
                    TimeUnit.MILLISECONDS
                );
            }
        }

        synchronized void flush() {
            if (pending.isEmpty()) return;
            List<Pair<GrabRequest,
                CompletableFuture<GrabResult>>> batch
                = new ArrayList<>(pending);
            pending.clear();
            flushTask = null;
            processor.accept(packetId, batch);
        }
    }
}