Как создать ленту для социальной сети?

Создание ленты для социальной сети: архитектура и реализация

Лента (Feed) социальной сети - это одна из самых сложных задач в backend разработке. Требуется обрабатывать миллионы записей, обеспечивать низкую задержку и масштабируемость.

Архитектура Feed

Основные компоненты:

1. Database (хранилище постов)
2. Cache (Redis для быстрого доступа)
3. Queue (для обработки событий)
4. Search Engine (Elasticsearch для полнотекстового поиска)
5. Real-time система (WebSockets для обновлений)

Простая реализация с использованием PostgreSQL

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True)
    email = Column(String, unique=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    posts = relationship("Post", back_populates="author")
    follows = relationship("Follow", foreign_keys="Follow.follower_id", back_populates="follower")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    author_id = Column(Integer, ForeignKey("users.id"))
    content = Column(String)
    created_at = Column(DateTime, default=datetime.utcnow)
    likes_count = Column(Integer, default=0)
    
    author = relationship("User", back_populates="posts")

class Follow(Base):
    __tablename__ = "follows"
    
    follower_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
    following_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    follower = relationship("User", foreign_keys=[follower_id])

Pull-Based подход (самый распространенный)

Лента собирается в момент запроса из постов, на которые подписан пользователь:

def get_feed(user_id: int, limit: int = 20, offset: int = 0):
    session = Session()
    
    # Получить ID всех пользователей, на которых подписан пользователь
    following_ids = session.query(Follow.following_id).filter(
        Follow.follower_id == user_id
    ).subquery()
    
    # Получить посты этих пользователей
    posts = session.query(Post).filter(
        Post.author_id.in_(session.query(Follow.following_id).filter(
            Follow.follower_id == user_id
        ))
    ).order_by(Post.created_at.desc()).limit(limit).offset(offset).all()
    
    return posts

# Проблема: медленно при большом количестве подписок

Оптимизация с Redis Cache

import redis
import json
from datetime import timedelta

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_feed_cached(user_id: int, limit: int = 20, offset: int = 0):
    cache_key = f"feed:{user_id}:{offset}:{limit}"
    
    # Проверить кэш
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Получить из БД
    posts = get_feed(user_id, limit, offset)
    
    # Сохранить в кэш на 5 минут
    post_data = [{"id": p.id, "content": p.content, "author_id": p.author_id} for p in posts]
    redis_client.setex(cache_key, timedelta(minutes=5), json.dumps(post_data))
    
    return post_data

Push-Based подход (для high-volume сетей)

Посты сразу записываются в ленты подписчиков:

class FeedEntry:
    def __init__(self, user_id: int, post_id: int):
        self.user_id = user_id
        self.post_id = post_id

def push_post_to_followers(post_id: int, author_id: int):
    session = Session()
    
    # Получить всех подписчиков автора
    followers = session.query(Follow.follower_id).filter(
        Follow.following_id == author_id
    ).all()
    
    # Добавить пост в ленту каждого подписчика
    for follower_id in followers:
        feed_key = f"feed:{follower_id[0]}"
        redis_client.lpush(feed_key, post_id)
        redis_client.ltrim(feed_key, 0, 999)  # Хранить только последние 1000 постов
        
        # Инвалидировать кэш
        redis_client.delete(f"feed:{follower_id[0]}:*")

def get_push_feed(user_id: int, limit: int = 20, offset: int = 0):
    feed_key = f"feed:{user_id}"
    
    # Получить посты из Redis
    post_ids = redis_client.lrange(feed_key, offset, offset + limit - 1)
    
    if not post_ids:
        return []
    
    # Получить полные данные из БД
    session = Session()
    posts = session.query(Post).filter(Post.id.in_([int(p) for p in post_ids])).all()
    
    return sorted(posts, key=lambda p: list(post_ids).index(str(p.id).encode()))

Гибридный подход

Сочетание pull и push для разных сценариев:

MAX_FOLLOWERS_FOR_PUSH = 1000

def publish_post(post_id: int, author_id: int):
    session = Session()
    follower_count = session.query(func.count(Follow.follower_id)).filter(
        Follow.following_id == author_id
    ).scalar()
    
    if follower_count < MAX_FOLLOWERS_FOR_PUSH:
        # Push для небольших подписок
        push_post_to_followers(post_id, author_id)
    else:
        # Pull для больших (знаменитостей)
        # Посты будут получены при запросе ленты
        pass

Real-time обновления с WebSockets

from fastapi import FastAPI, WebSocket
import asyncio
import json

app = FastAPI()

active_connections = []

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: int):
    await websocket.accept()
    active_connections.append((user_id, websocket))
    
    try:
        while True:
            # Ждем сообщений
            data = await websocket.receive_text()
    except Exception:
        active_connections.remove((user_id, websocket))

async def broadcast_new_post(post_id: int, author_id: int):
    session = Session()
    followers = session.query(Follow.follower_id).filter(
        Follow.following_id == author_id
    ).all()
    
    # Отправить обновление всем online подписчикам
    for user_id, websocket in active_connections:
        if user_id in [f[0] for f in followers]:
            await websocket.send_json({
                "type": "new_post",
                "post_id": post_id,
                "author_id": author_id
            })

Счетчики и метрики

def increment_post_likes(post_id: int):
    # Использовать Redis Streams для счетчиков
    redis_client.incr(f"post:{post_id}:likes")

def get_post_likes(post_id: int) -> int:
    return int(redis_client.get(f"post:{post_id}:likes") or 0)

def sync_likes_to_db():
    # Периодически синхронизировать счетчики с БД
    session = Session()
    for post_id in redis_client.keys("post:*:likes"):
        count = redis_client.get(post_id)
        session.query(Post).filter(Post.id == post_id).update({"likes_count": count})
    session.commit()

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

Проблемы при росте:

• БД не может справиться с трафиком
• Лента становится слишком медленной
• Real-time обновления не масштабируются

Решения:

# 1. Вертикальное масштабирование (больше памяти/CPU)
# 2. Горизонтальное (несколько инстансов с load balancer)
# 3. Кэширование (Redis, Memcached)
# 4. Очереди (Celery, RabbitMQ)
# 5. CQRS (разделение чтения и записи)
# 6. Event Sourcing
# 7. Elasticsearch для поиска

Лучшие практики

• Используйте pull-based для начала (проще)
• Переходите на push-based когда масштабируется
• Кэшируйте агрессивно
• Используйте асинхронную обработку событий
• Мониторьте performance
• Тестируйте с реальными объемами данных
• Имейте план для вертикального/горизонтального масштабирования
• Используйте CDN для изображений и медиа