后端架构面试准备指南¶

import hashlib
import redis

class ShortLinkService:
    def __init__(self, redis_client):
        self.redis = redis_client

    def generate_short_link(self, long_url):
        # 生成短链接
        hash_value = hashlib.md5(long_url.encode()).hexdigest()[:8]
        short_code = self._base62_encode(int(hash_value, 16))

        # 检查是否重复
        if self.redis.exists(f"short:{short_code}"):
            return self.generate_short_link(long_url + "_")

        # 存储映射
        self.redis.setex(f"short:{short_code}", 86400, long_url)
        self.redis.incr(f"count:{short_code}")

        return f"http://short.link/{short_code}"

    def get_long_link(self, short_code):
        # 获取长链接
        long_url = self.redis.get(f"short:{short_code}")
        if long_url:
            # 增加访问计数
            self.redis.incr(f"count:{short_code}")
            return long_url.decode()
        return None

    def _base62_encode(self, num):
        characters = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
        if num == 0:
            return characters[0]
        base62 = []
        while num:
            num, rem = divmod(num, 62)
            base62.append(characters[rem])
        return ''.join(reversed(base62))

import redis
import json

class SeckillService:
    def __init__(self, redis_client, producer):
        self.redis = redis_client
        self.producer = producer

    def seckill(self, user_id, product_id):
        # 检查用户是否已购买
        if self.redis.sismember(f"bought:{product_id}", user_id):
            return {"success": False, "message": "Already bought"}

        # 检查库存
        stock = self.redis.decr(f"stock:{product_id}")
        if stock < 0:
            self.redis.incr(f"stock:{product_id}")
            return {"success": False, "message": "Out of stock"}

        # 标记用户已购买
        self.redis.sadd(f"bought:{product_id}", user_id)

        # 发送消息到MQ
        self.producer.send('seckill-orders', {
            'user_id': user_id,
            'product_id': product_id,
            'timestamp': time.time()
        })

        return {"success": True, "message": "Success"}

from fastapi import FastAPI, WebSocket
import json
from typing import Dict

app = FastAPI()

# 存储WebSocket连接
connections: Dict[int, WebSocket] = {}

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: int):
    await websocket.accept()
    connections[user_id] = websocket

    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)

            # 处理消息
            await handle_message(user_id, message)
    finally:
        del connections[user_id]

async def handle_message(sender_id, message):
    receiver_id = message['receiver_id']

    # 如果接收者在线，直接发送
    if receiver_id in connections:
        await connections[receiver_id].send_json(message)
    else:
        # 否则存储离线消息
        store_offline_message(receiver_id, message)

import time
import threading

class SnowflakeIDGenerator:
    def __init__(self, datacenter_id, worker_id):
        self.datacenter_id = datacenter_id
        self.worker_id = worker_id
        self.sequence = 0
        self.last_timestamp = -1
        self.lock = threading.Lock()

    def generate_id(self):
        with self.lock:
            timestamp = int(time.time() * 1000)

            if timestamp < self.last_timestamp:
                raise Exception("Clock moved backwards")

            if timestamp == self.last_timestamp:
                self.sequence = (self.sequence + 1) & 0xFFF
                if self.sequence == 0:
                    timestamp = self.wait_next_millis(self.last_timestamp)
            else:
                self.sequence = 0

            self.last_timestamp = timestamp

            return ((timestamp - 1288834974657) << 22) | \
                   (self.datacenter_id << 17) | \
                   (self.worker_id << 12) | \
                   self.sequence

    def wait_next_millis(self, last_timestamp):
        timestamp = int(time.time() * 1000)
        while timestamp <= last_timestamp:
            timestamp = int(time.time() * 1000)
        return timestamp