PostgreSQL做所有事:队列、搜索、向量在一个数据库

1. 概述

一个典型的系统通常由许多组件组成：PostgreSQL存储数据、一个消息代理做队列、一个搜索引擎用于搜索，最近还加上一个向量数据库用于AI功能。每个组件都会带来运维、监控、备份和数据同步的工作。

“Postgres做所有事”的趋势反过来提问：如果PostgreSQL已经能够以足够好的水平完成大部分这些工作，是否需要再添加新系统呢？在这篇文章中，你将学到如何使用PostgreSQL做job队列、全文搜索和向量搜索，配合Spring Boot 3.x。你也会了解到什么时候应该停止，转而选择专用工具。

2. 为什么这个趋势受到关注

有几个主要原因：

• 降低运维复杂性。 每添加一个系统就多一个可能在半夜故障的东西。对于小团队来说，这个成本通常大于性能收益。
• 事务一致性。 当订单和发送邮件的job都在同一个数据库中时，你可以在一个事务内同时写入两者。不会出现订单已保存但消息丢失，或反之的情况。
• 生态系统中的扩展。 PostgreSQL有强大的扩展机制。比如pgvector添加了向量数据类型和最近邻搜索索引。
• AI功能带来了存储embedding的需求。 许多团队想要添加语义搜索功能，但不想为此建立一个新的数据库。

这个趋势的精神是从简单开始，当有数据证明需要分离时再分离。它并不意味着PostgreSQL可以替代所有系统。

3. 准备

3.1. Maven依赖

你需要JDBC、PostgreSQL驱动和Testcontainers用于测试。Spring Boot已经管理版本，所以你不需要声明它们：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-testcontainers</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>postgresql</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>junit-jupiter</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

3.2. 运行带pgvector的PostgreSQL

pgvector/pgvector镜像预装了PostgreSQL和vector扩展。选择与你使用的PostgreSQL主版本相应的标签：

services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_DB: shop
      POSTGRES_USER: shop
      POSTGRES_PASSWORD: secret
    ports:
      - "5432:5432"

配置Spring Boot以连接并在启动时自动运行schema.sql：

spring.datasource.url=jdbc:postgresql://localhost:5432/shop
spring.datasource.username=shop
spring.datasource.password=secret
spring.sql.init.mode=always

在实际项目中，你应该使用Flyway或Liquibase替代schema.sql。

4. 使用FOR UPDATE SKIP LOCKED的Job队列

4.1. 数据库模式

一个最小的队列只需要一张表：

CREATE TABLE IF NOT EXISTS job_queue (
    id         BIGSERIAL PRIMARY KEY,
    job_type   TEXT        NOT NULL,
    payload    JSONB       NOT NULL,
    run_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_job_queue_run_at ON job_queue (run_at);

run_at列允许你安排job何时运行。对run_at的索引有助于worker快速获取最早的job，而不是扫描整个表。

4.2. 获取并处理job

关键是FOR UPDATE SKIP LOCKED子句。多个worker可以同时运行这个查询。每个worker锁定一行，其他worker跳过已被锁定的行而不是等待：

public record Job(long id, String type, String payload) {}

@Repository
public class JobQueue {

    private final JdbcTemplate jdbc;

    public JobQueue(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public void enqueue(String type, String payloadJson) {
        jdbc.update("INSERT INTO job_queue (job_type, payload) VALUES (?, ?::jsonb)",
                type, payloadJson);
    }

    @Transactional
    public boolean processNext(Consumer<Job> handler) {
        List<Job> jobs = jdbc.query("""
                SELECT id, job_type, payload::text
                FROM job_queue
                WHERE run_at <= now()
                ORDER BY run_at
                LIMIT 1
                FOR UPDATE SKIP LOCKED
                """,
                (rs, i) -> new Job(rs.getLong(1), rs.getString(2), rs.getString(3)));

        if (jobs.isEmpty()) {
            return false;
        }
        Job job = jobs.getFirst();
        handler.accept(job);
        jdbc.update("DELETE FROM job_queue WHERE id = ?", job.id());
        return true;
    }
}

解释：

• enqueue插入一个job。?::jsonb将字符串强制转换为JSONB类型。
• processNext在一个事务内运行。SELECT ... FOR UPDATE SKIP LOCKED语句锁定恰好一个未被任何人处理的job。
• 如果handler运行完成，job被删除并且事务提交。
• 如果handler抛出异常，事务回滚。Job仍然在表中并将被重新处理。如果worker在中途崩溃，PostgreSQL会在连接关闭时自动释放锁。

Worker可以是一个简单的@Scheduled方法：

@Component
public class EmailWorker {

    private final JobQueue jobQueue;

    public EmailWorker(JobQueue jobQueue) {
        this.jobQueue = jobQueue;
    }

    @Scheduled(fixedDelay = 1000)
    public void poll() {
        while (jobQueue.processNext(job ->
                System.out.println("为工作 " + job.id() + " 发送邮件: " + job.payload()))) {
            // 继续直到没有job
        }
    }
}

记得在一个配置类中添加@EnableScheduling。当有job时，控制台会输出：

为工作 1 发送邮件: {"orderId": 42, "email": "an@example.com"}

注意： 在处理job时保持事务的方式非常简单安全，即使worker崩溃也不会丢失数据。代价是事务会根据处理时间拉长。如果job需要很长时间运行（调用缓慢的API、处理大文件），你应该改用带有status和locked_until列的模型，或者切换到专用的队列库。你还应该添加一个attempts列以防止一个失败的job无限重试。

4.3. 真正的优势：在业务事务中一起入队

这是使用Postgres做队列的最强理由：

@Service
public class OrderService {

    private final JdbcTemplate jdbc;
    private final JobQueue jobQueue;

    public OrderService(JdbcTemplate jdbc, JobQueue jobQueue) {
        this.jdbc = jdbc;
        this.jobQueue = jobQueue;
    }

    @Transactional
    public void placeOrder(long orderId, String customerEmail) {
        jdbc.update("INSERT INTO orders (id, customer_email) VALUES (?, ?)", orderId, customerEmail);
        jobQueue.enqueue("SEND_CONFIRMATION_EMAIL",
                "{\"orderId\": %d, \"email\": \"%s\"}".formatted(orderId, customerEmail));
    }
}

订单和job一起提交或一起回滚。使用单独的代理，你必须自己构建事务性outbox模式才能达到这个保证。（例子使用formatted以求简洁。真实代码应该使用Jackson序列化payload以正确转义字符。）

如果你不想让worker持续polling，你可以结合PostgreSQL的LISTEN/NOTIFY来在有新job时唤醒worker。Polling仍然应该保留为备选方案。

5. 使用tsvector和GIN索引的全文搜索

5.1. 数据库模式

PostgreSQL内置了全文搜索。你可以使用一个生成列在数据更改时自动更新搜索向量：

CREATE TABLE IF NOT EXISTS product (
    id            BIGSERIAL PRIMARY KEY,
    name          TEXT NOT NULL,
    description   TEXT NOT NULL,
    search_vector tsvector GENERATED ALWAYS AS (
        setweight(to_tsvector('simple', name), 'A') ||
        setweight(to_tsvector('simple', description), 'B')
    ) STORED
);

CREATE INDEX IF NOT EXISTS idx_product_search ON product USING GIN (search_vector);

解释：

• to_tsvector('simple', ...)将文本分解为标记。simple配置仅转换为小写，不进行词根提取。这个选择适合中文，因为PostgreSQL没有中文分词的默认配置。
• setweight为名称分配权重A，为描述分配权重B。名称中的匹配将被排名更高。
• GIN索引帮助在tsvector列上快速查询。

5.2. 查询

public record ProductHit(long id, String name, double rank) {}

@Repository
public class ProductSearch {

    private final JdbcTemplate jdbc;

    public ProductSearch(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public List<ProductHit> search(String userQuery) {
        return jdbc.query("""
                SELECT id, name, ts_rank(search_vector, q) AS rank
                FROM product, websearch_to_tsquery('simple', ?) AS q
                WHERE search_vector @@ q
                ORDER BY rank DESC
                LIMIT 10
                """,
                (rs, i) -> new ProductHit(rs.getLong("id"), rs.getString("name"), rs.getDouble("rank")),
                userQuery);
    }
}

websearch_to_tsquery接受类似于普通搜索框的语法："防风衣" -长 表示包含短语“防风衣”且不包含“长”。@@运算符检查向量是否与查询匹配。ts_rank用于对结果排序。

使用包含“防风衣”和“牛仔裤”的示例数据（其描述中有“防水”），查询防风返回：

ProductHit[id=1, name=防风衣, rank=0.6079271]
ProductHit[id=2, name=牛仔裤, rank=0.24317084]

注意： 用户经常输入不同形式的查询。你可以使用扩展来改进匹配。默认情况下某些函数没有标记为IMMUTABLE，所以要在生成列或索引中使用它，你需要将其包装在一个单独的IMMUTABLE函数中。要进行容错拼写搜索，你可以使用pg_trgm扩展。

6. 使用pgvector的向量搜索

6.1. 数据库模式

Embedding是由AI模型生成的实数数组。含义相近的文本会生成相近的向量。启用扩展并添加列：

CREATE EXTENSION IF NOT EXISTS vector;

ALTER TABLE product ADD COLUMN IF NOT EXISTS embedding vector(3);

CREATE INDEX IF NOT EXISTS idx_product_embedding
    ON product USING hnsw (embedding vector_cosine_ops);

例子使用3维来便于阅读。实际上，维数必须恰好匹配你选择的embedding模型，通常是几百到几千维。HNSW索引允许非常快速的近似最近邻查询。vector_cosine_ops指示索引用于余弦距离。

6.2. 保存和按向量搜索

不需要额外的库。你将float[]转换为[0.1,0.2,0.3]格式的字符串，然后强制转换为::vector：

@Repository
public class ProductVectorSearch {

    private final JdbcTemplate jdbc;

    public ProductVectorSearch(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public void saveEmbedding(long productId, float[] embedding) {
        jdbc.update("UPDATE product SET embedding = ?::vector WHERE id = ?",
                toLiteral(embedding), productId);
    }

    public List<String> findSimilar(float[] queryEmbedding, int limit) {
        return jdbc.queryForList("""
                SELECT name FROM product
                WHERE embedding IS NOT NULL
                ORDER BY embedding <=> ?::vector
                LIMIT ?
                """, String.class, toLiteral(queryEmbedding), limit);
    }

    private static String toLiteral(float[] v) {
        StringJoiner joiner = new StringJoiner(",", "[", "]");
        for (float f : v) {
            joiner.add(Float.toString(f));
        }
        return joiner.toString();
    }
}

<=>运算符计算余弦距离。ORDER BY ... LIMIT获取最相似的产品。查询向量来自与为产品生成embedding的同一模型，例如当用户输入“下雨天穿的衣服”时。

由于产品和embedding位于同一张表中，你可以轻松组合过滤：WHERE category = 'outerwear' AND price < 500000 ORDER BY embedding <=> ?。你也可以混合ts_rank分数和向量距离来进行混合搜索。使用单独的向量数据库，你必须将这些属性同步到那里。

注意： 当将高选择性的WHERE与HNSW索引结合时，返回的结果数量可能少于LIMIT。原因是索引在扫描有限的候选集合后进行过滤。请使用真实数据仔细测试，并参考pgvector的调优参数。

7. 使用Testcontainers编写测试

测试在真实的PostgreSQL上运行，带有pgvector的镜像。@ServiceConnection自动配置datasource：

@SpringBootTest
@Testcontainers
class PostgresFeaturesTest {

    @Container
    @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
            DockerImageName.parse("pgvector/pgvector:pg16").asCompatibleSubstituteFor("postgres"));

    @Autowired JobQueue jobQueue;
    @Autowired ProductSearch productSearch;
    @Autowired ProductVectorSearch vectorSearch;
    @Autowired JdbcTemplate jdbc;

    @Test
    void processesEachJobExactlyOnce() {
        jobQueue.enqueue("SEND_EMAIL", "{\"orderId\": 1}");
        jobQueue.enqueue("SEND_EMAIL", "{\"orderId\": 2}");
        List<Long> processed = new ArrayList<>();

        while (jobQueue.processNext(job -> processed.add(job.id()))) { }

        assertThat(processed).hasSize(2).doesNotHaveDuplicates();
        assertThat(jdbc.queryForObject("SELECT count(*) FROM job_queue", Long.class)).isZero();
    }

    @Test
    void failedJobStaysInQueue() {
        jobQueue.enqueue("SEND_EMAIL", "{\"orderId\": 3}");

        assertThatThrownBy(() -> jobQueue.processNext(job -> {
            throw new IllegalStateException("SMTP down");
        })).isInstanceOf(IllegalStateException.class);

        assertThat(jdbc.queryForObject("SELECT count(*) FROM job_queue", Long.class)).isEqualTo(1);
    }

    @Test
    void searchesByTextAndVector() {
        jdbc.update("INSERT INTO product (name, description) VALUES ('防风衣', '防水轻质')");
        jdbc.update("INSERT INTO product (name, description) VALUES ('牛仔裤', '厚布料')");
        long jacketId = jdbc.queryForObject("SELECT id FROM product WHERE name = '防风衣'", Long.class);
        long jeansId = jdbc.queryForObject("SELECT id FROM product WHERE name = '牛仔裤'", Long.class);
        vectorSearch.saveEmbedding(jacketId, new float[]{0.9f, 0.1f, 0.0f});
        vectorSearch.saveEmbedding(jeansId, new float[]{0.0f, 0.2f, 0.9f});

        assertThat(productSearch.search("防风"))
                .extracting(ProductHit::name)
                .containsExactly("防风衣");
        assertThat(vectorSearch.findSimilar(new float[]{1.0f, 0.0f, 0.0f}, 1))
                .containsExactly("防风衣");
    }
}

第二个测试检查队列最重要的行为：当处理程序失败时，job不会丢失。测试将每个方法的数据设置到位。在实际项目中，你应该在@BeforeEach中清空表，以使测试相互独立。

8. 什么时候不应该为所有事情使用Postgres

把所有东西都放在一个地方也有代价。当满足以下条件时，你应该考虑专用系统：

• 队列吞吐量非常高或需要流式处理。 如果你需要重放事件流、有多个独立的消费者组，或者需要长期保留消息，使用Kafka或类似工具。Postgres上的队列由于持续的DELETE/UPDATE会产生许多死元组。你必须仔细监控autovacuum。
• 复杂的搜索。 多维分面、输入时建议、优质的多语言支持、细粒度的相关性调整：专用搜索引擎做得更好。
• 向量数据量非常大。 HNSW索引需要大量RAM。当向量数据占用大部分资源时，它会与主要事务工作量竞争。
• 不均匀的负载互相影响。 一个繁重的向量查询或一批job涌入可能会减慢支付API的速度。读副本帮助分离一些读取负载，但不能完全解决问题。
• 你的团队已经成功运维了专用系统。 如果基础设施已经到位且稳定，为了“简洁”而将所有内容迁移回Postgres可能不值得费力。

注意： 一个合理的策略是在每个功能的后面放一个接口（JobQueue、ProductSearch…）。当监控数据表明Postgres成为瓶颈时，你可以替换底层实现而无需修改业务代码。

9. 结论

PostgreSQL足以通过FOR UPDATE SKIP LOCKED承载job队列，通过tsvector和GIN索引进行全文搜索，以及通过pgvector扩展进行向量搜索。最大的优势是需要运维的系统更少，并且可以在同一个事务中写入业务数据、job和embedding。

将这个视为小型和中型系统的默认起点，而不是永久的终点。从Postgres开始，仔细测量，仅当数据表明需要时才切换到专用工具。