Сервис сокращения ссылок

Сервис сокращения ссылок

Это одна из самых популярных задач на system design интервью. Нужно создать масштабируемый сервис, который преобразует длинные URL в короткие и обеспечивает быструю переадресацию.

1. Архитектура приложения

Клиент → API Gateway → Load Balancer → Сервисы (Sharding) → БД/Кэш
         ↓                                    ↓
      Validation                          Analytics
      Rate Limiting
      Authentication

Компоненты:

• API сервер (Flask/FastAPI): обрабатывает POST (создание) и GET (переадресация)
• Кэш слой (Redis): хранит популярные ссылки для быстрого доступа
• Первичное хранилище (PostgreSQL): надёжное хранение сопоставлений
• Analytics (ElasticSearch/Analytics DB): отслеживание статистики кликов

2. Генерирование коротких URL

Вариант 1: Base62 кодирование с инкрементирующимся ID

import base64

BASE62 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"

def id_to_short_url(url_id: int) -> str:
    """Преобразует числовой ID в короткую ссылку."""
    if url_id == 0:
        return BASE62[0]
    
    short_url = ""
    while url_id > 0:
        short_url = BASE62[url_id % 62] + short_url
        url_id //= 62
    
    return short_url

def short_url_to_id(short_url: str) -> int:
    """Декодирует короткую ссылку обратно в ID."""
    url_id = 0
    for char in short_url:
        url_id = url_id * 62 + BASE62.index(char)
    return url_id

# Пример
print(id_to_short_url(12345))      # "3d7"
print(short_url_to_id("3d7"))      # 12345

Вариант 2: MD5/SHA1 хеширование с усечением

import hashlib

def hash_based_short_url(long_url: str) -> str:
    """Генерирует короткую ссылку через хеширование."""
    # Берём первые 6 символов хеша
    hash_obj = hashlib.md5(long_url.encode())
    return hash_obj.hexdigest()[:6]

Вариант 3: Случайная строка

import string
import random

def random_short_url(length: int = 6) -> str:
    """Генерирует случайную короткую ссылку."""
    chars = string.ascii_letters + string.digits
    return .join(random.choice(chars) for _ in range(length))

3. Обработка коллизий хешей

Проблема: Если используем MD5 или случайные строки, могут быть коллизии

Решения:

from typing import Optional
import uuid

class URLShorteningService:
    def __init__(self, db):
        self.db = db
    
    # Подход 1: Инкрементирующийся счётчик (Best для масштабирования)
    def create_short_url_with_id(self, long_url: str) -> str:
        """Создаёт короткую ссылку с использованием ID счётчика."""
        # Распределённый счётчик в Redis/ZooKeeper
        url_id = self.db.increment_counter("url_id_counter")
        short_url = self.id_to_short_url(url_id)
        
        self.db.set_mapping(short_url, long_url)
        return short_url
    
    # Подход 2: Детерминированный хеш с попыткой
    def create_short_url_with_retry(self, long_url: str) -> str:
        """Создаёт ссылку с обработкой коллизий."""
        for attempt in range(5):
            hash_val = hashlib.md5((long_url + str(attempt)).encode())
            short_url = hash_val.hexdigest()[:6]
            
            # Проверяем, занята ли ссылка
            if not self.db.exists(short_url):
                self.db.set_mapping(short_url, long_url)
                return short_url
        
        # Если все попытки неудачны, используем UUID
        short_url = str(uuid.uuid4())[:6]
        self.db.set_mapping(short_url, long_url)
        return short_url

4. Выбор хранилища данных

PostgreSQL (Primary Store):

• ✅ Надёжность (ACID транзакции)
• ✅ Исторические данные и аналитика
• ✅ Возможность восстановления
• ❌ Медленнее для высоконагруженного чтения

# Схема БД
CREATE TABLE url_mappings (
    id BIGSERIAL PRIMARY KEY,
    short_url VARCHAR(10) UNIQUE NOT NULL,
    long_url TEXT NOT NULL,
    user_id UUID,
    created_at TIMESTAMP DEFAULT NOW(),
    clicks INT DEFAULT 0,
    INDEX idx_short_url (short_url)
);

CREATE TABLE url_clicks (
    id BIGSERIAL PRIMARY KEY,
    short_url VARCHAR(10),
    ip_address INET,
    user_agent TEXT,
    clicked_at TIMESTAMP DEFAULT NOW()
);

Redis (Кэш):

• ✅ Очень быстро (in-memory)
• ✅ Идеально для горячих данных
• ❌ Потеря данных при перезагрузке
• ❌ Ограниченный объём памяти

class URLService:
    def __init__(self, db, cache):
        self.db = db      # PostgreSQL
        self.cache = cache  # Redis
    
    def expand_url(self, short_url: str) -> Optional[str]:
        """Получает длинный URL с кэшированием."""
        # Проверяем кэш
        cached = self.cache.get(f"url:{short_url}")
        if cached:
            return cached
        
        # Если нет в кэше - берём из БД
        long_url = self.db.get_url(short_url)
        if long_url:
            # Кэшируем на 24 часа
            self.cache.setex(f"url:{short_url}", 86400, long_url)
            return long_url
        
        return None

5. Масштабирование сервиса

Проблема: Миллионы запросов в день требуют горизонтального масштабирования

Решение - Sharding по short_url:

class ShardedURLService:
    def __init__(self, num_shards: int = 4):
        self.num_shards = num_shards
        self.shards = [self._create_shard() for _ in range(num_shards)]
    
    def get_shard_id(self, short_url: str) -> int:
        """Определяет шард по хешу ссылки."""
        return hash(short_url) % self.num_shards
    
    def shorten(self, long_url: str) -> str:
        """Создаёт сокращённую ссылку в нужном шарде."""
        short_url = self.generate_short_url()
        shard_id = self.get_shard_id(short_url)
        self.shards[shard_id].store(short_url, long_url)
        return short_url
    
    def expand(self, short_url: str) -> str:
        """Получает длинный URL из нужного шарда."""
        shard_id = self.get_shard_id(short_url)
        return self.shards[shard_id].get(short_url)

Дополнительные техники масштабирования:

1. Load Balancing: Nginx/HAProxy распределяет нагрузку
2. CDN: Кэширование популярных ссылок на edge nodes
3. Read Replicas: Несколько read-only копий БД для чтения
4. Асинхронная запись статистики: Analytics пишутся в очередь (RabbitMQ/Kafka)
5. Consistenhash: Для добавления новых шардов без пересчёта

Пример FastAPI реализации

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, HttpUrl
import redis
from sqlalchemy import create_engine

app = FastAPI()
redis_client = redis.Redis()

class ShortenRequest(BaseModel):
    long_url: HttpUrl

@app.post("/shorten")
async def shorten_url(request: ShortenRequest):
    """Создаёт короткую ссылку."""
    service = URLShorteningService(db, redis_client)
    short_url = service.create_short_url(str(request.long_url))
    return {"short_url": f"https://short.url/{short_url}"}

@app.get("/expand/{short_url}")
async def expand_url(short_url: str):
    """Переадресует на длинный URL."""
    service = URLShorteningService(db, redis_client)
    long_url = service.expand_url(short_url)
    if not long_url:
        raise HTTPException(status_code=404, detail="URL not found")
    return {"long_url": long_url}

Ключевые метрики для интервью

• QPS: ~10K запросов/сек для сокращения, ~100K для расширения
• Латенсия: <100ms для 99th percentile
• Хранилище: ~100 байт на запись, 1 млн ссылок = ~100MB
• Availability: 99.99% uptime