PostgreSQL高效处理上亿级图片URL与MD5映射关系的设计方案

CREATE TABLE image_md5_url (
    md5   CHAR(32) PRIMARY KEY,      -- 32位小写MD5，无索引膨胀
    url   TEXT NOT NULL              -- 图片URL，可能很长
);

-- 仅当需要“URL → MD5”查询时，添加以下索引
-- 为反向查询（URL → MD5）建唯一索引（如果需要）
CREATE UNIQUE INDEX CONCURRENTLY idx_image_url ON image_md5_url (url);

CREATE TABLE image_md5_url (
    md5 CHAR(32) PRIMARY KEY,
    url TEXT NOT NULL
);

INSERT INTO image_md5_url (md5, url)
VALUES ('d41d...', 'https://a.com/1.jpg')
ON CONFLICT (md5) DO NOTHING;

# 核心逻辑：批量 + 冲突忽略
async def save_batch(session, batch):
    stmt = text("""
        INSERT INTO image_md5_url (md5, url)
        VALUES (:md5, :url)
        ON CONFLICT (md5) DO NOTHING
    """)
    await session.execute(stmt, [
        {"md5": md5.lower(), "url": url} for md5, url in batch
    ])
    await session.commit()

for attempt in range(3):
    try:
        await save_batch(...)
        break
    except (OperationalError, TimeoutError):
        await asyncio.sleep(2 ** attempt)

# postgresql.conf
shared_buffers = 4GB          # 总内存 25%
effective_cache_size = 12GB   # OS 缓存预估
work_mem = 256MB              # 排序/哈希内存
max_connections = 200         # 避免过多连接竞争
wal_compression = on          # 减少 WAL 体积

-- 针对大表单独设置
ALTER TABLE image_md5_url SET (
    autovacuum_vacuum_scale_factor = 0.01,  -- 1% 变化即触发
    autovacuum_vacuum_cost_delay = 0        -- 不限速
);

CREATE UNIQUE INDEX CONCURRENTLY idx_image_url ON image_md5_url (url);

# Python 示例（asyncpg 或 psycopg3）
async def insert_batch(records):
    # records: [(md5, url), ...]
    await conn.executemany(
        "INSERT INTO image_md5_url (md5, url) VALUES ($1, $2) ON CONFLICT DO NOTHING",
        records
    )

-- 按 MD5 前两位哈希分区（256 分区）
CREATE TABLE image_md5_url (
    md5 CHAR(32) NOT NULL,
    url TEXT NOT NULL,
    PRIMARY KEY (md5)
) PARTITION BY HASH (md5);

CREATE TABLE image_md5_url (
    md5 CHAR(32) NOT NULL,
    url TEXT NOT NULL
) PARTITION BY HASH (md5);

-- 创建 256 个分区（md5 前两位）
DO $$
BEGIN
  FOR i IN 0..255 LOOP
    EXECUTE format('
      CREATE TABLE image_md5_url_p%s PARTITION OF image_md5_url
      FOR VALUES WITH (MODULUS 256, REMAINDER %s)
    ', i, i);
  END LOOP;
END $$;

-- 在 Citus 中分布表
SELECT create_distributed_table('image_md5_url', 'md5');

CREATE TABLE image_ref_count (
    md5 CHAR(32) PRIMARY KEY,
    count INT NOT NULL DEFAULT 1
);

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20, max_overflow=30, pool_pre_ping=True
)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# main.py
import asyncio
from sqlalchemy import text

async def worker(queue, worker_id):
    async with AsyncSessionLocal() as session:
        batch = []
        while True:
            try:
                item = await asyncio.wait_for(queue.get(), timeout=2.0)
                if item is None: break
                batch.append(item)
                
                if len(batch) >= 5000:
                    await save_batch(session, batch)
                    batch.clear()
            except asyncio.TimeoutError:
                if batch: await save_batch(session, batch)
                break

async def save_batch(session, batch):
    stmt = text("""
        INSERT INTO image_md5_url (md5, url)
        VALUES (:md5, :url)
        ON CONFLICT (md5) DO NOTHING
    """)
    await session.execute(stmt, [{"md5": m.lower(), "url": u} for m, u in batch])
    await session.commit()