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()