Postgres làm mọi thứ: queue, search, vector trong một database

1. Tổng quan

Một hệ thống điển hình hay có nhiều thành phần: PostgreSQL lưu dữ liệu, một message broker làm queue, một search engine cho tìm kiếm, và gần đây thêm một vector database cho tính năng AI. Mỗi thành phần lại kéo theo việc vận hành, giám sát, backup và đồng bộ dữ liệu.

Xu hướng “Postgres làm mọi thứ” đặt câu hỏi ngược lại: nếu PostgreSQL đã làm được phần lớn những việc đó ở mức đủ tốt, có cần thêm hệ thống mới không? Trong bài này, bạn sẽ dùng PostgreSQL làm job queue, full-text search và vector search với Spring Boot 3.x. Bạn cũng sẽ biết khi nào nên dừng lại và chọn công cụ chuyên dụng.

2. Vì sao xu hướng này được quan tâm

Có vài lý do chính:

• Giảm độ phức tạp vận hành. Mỗi hệ thống thêm vào là thêm một thứ có thể hỏng lúc nửa đêm. Với team nhỏ, chi phí này thường lớn hơn lợi ích về hiệu năng.
• Tính nhất quán theo transaction. Khi đơn hàng và job gửi email nằm chung một database, bạn ghi cả hai trong một transaction. Không có tình huống đơn hàng đã lưu mà message bị mất, hoặc ngược lại.
• Hệ sinh thái extension. PostgreSQL có cơ chế extension mạnh. Ví dụ pgvector thêm kiểu dữ liệu vector và index tìm kiếm lân cận gần nhất.
• Tính năng AI kéo theo nhu cầu lưu embedding. Nhiều team muốn thêm tính năng tìm kiếm ngữ nghĩa mà không muốn dựng thêm một database mới chỉ cho việc đó.

Tinh thần của xu hướng này là bắt đầu đơn giản, tách ra khi có số liệu chứng minh cần tách. Nó không có nghĩa là PostgreSQL thay thế được mọi hệ thống.

3. Chuẩn bị

3.1. Dependency Maven

Bạn cần JDBC, driver PostgreSQL và Testcontainers cho phần test. Spring Boot đã quản lý version nên bạn không cần khai báo:

<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. Chạy PostgreSQL có sẵn pgvector

Image pgvector/pgvector đóng gói sẵn PostgreSQL và extension vector. Bạn chọn tag ứng với major version PostgreSQL đang dùng:

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

Cấu hình Spring Boot để kết nối và tự chạy schema.sql khi khởi động:

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

Trong dự án thật, bạn nên dùng Flyway hoặc Liquibase thay cho schema.sql.

4. Job queue với FOR UPDATE SKIP LOCKED

4.1. Schema

Một queue tối giản chỉ cần một bảng:

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

Cột run_at cho phép bạn hẹn giờ chạy job. Index trên run_at giúp worker lấy job sớm nhất mà không phải quét cả bảng.

4.2. Lấy và xử lý job

Điểm mấu chốt là mệnh đề FOR UPDATE SKIP LOCKED. Nhiều worker cùng chạy câu query. Mỗi worker khoá một dòng, và worker khác bỏ qua dòng đã bị khoá thay vì đứng chờ:

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;
    }
}

Giải thích:

• enqueue chèn một job. ?::jsonb ép chuỗi JSON sang kiểu JSONB.
• processNext chạy trong một transaction. Câu SELECT ... FOR UPDATE SKIP LOCKED khoá đúng một job chưa ai nhận.
• Nếu handler chạy xong, job bị xoá và transaction commit.
• Nếu handler ném exception, transaction rollback. Job vẫn còn trong bảng và sẽ được xử lý lại. Nếu worker chết giữa chừng, PostgreSQL tự nhả khoá khi kết nối đóng.

Worker có thể là một method @Scheduled đơn giản:

@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("Gửi email cho job " + job.id() + ": " + job.payload()))) {
            // tiếp tục cho tới khi hết job
        }
    }
}

Nhớ thêm @EnableScheduling vào một class cấu hình. Khi có job, console in ra:

Gửi email cho job 1: {"orderId": 42, "email": "an@example.com"}

Lưu ý: Cách giữ transaction trong lúc xử lý job rất đơn giản và an toàn khi worker crash. Đổi lại, transaction sẽ kéo dài theo thời gian xử lý. Nếu job chạy lâu (gọi API chậm, xử lý file lớn), bạn nên chuyển sang mô hình có cột status và locked_until, hoặc chuyển sang thư viện queue chuyên dụng. Bạn cũng nên thêm cột attempts để tránh một job lỗi bị retry vô hạn.

4.3. Lợi ích thật sự: enqueue cùng transaction nghiệp vụ

Đây là lý do mạnh nhất để dùng Postgres làm queue:

@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));
    }
}

Đơn hàng và job cùng commit hoặc cùng rollback. Với một broker riêng, bạn phải tự dựng pattern transactional outbox mới đạt được đảm bảo này. (Ví dụ dùng formatted cho ngắn gọn. Code thật nên serialize payload bằng Jackson để escape ký tự đúng.)

Nếu không muốn worker poll liên tục, bạn có thể kết hợp LISTEN/NOTIFY của PostgreSQL để đánh thức worker khi có job mới. Polling vẫn nên giữ lại làm phương án dự phòng.

5. Full-text search với tsvector và GIN index

5.1. Schema

PostgreSQL có sẵn full-text search. Bạn dùng một generated column để tự cập nhật vector tìm kiếm mỗi khi dữ liệu thay đổi:

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

Giải thích:

• to_tsvector('simple', ...) tách văn bản thành các token. Cấu hình simple chỉ chuyển về chữ thường, không stemming. Lựa chọn này phù hợp với tiếng Việt, vì PostgreSQL không có từ điển tiếng Việt mặc định.
• setweight gán trọng số A cho tên và B cho mô tả. Kết quả khớp ở tên sẽ được xếp hạng cao hơn.
• GIN index giúp truy vấn nhanh trên cột tsvector.

5.2. Truy vấn

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 nhận cú pháp giống ô tìm kiếm thông thường: "áo khoác" -len nghĩa là chứa cụm “áo khoác” và không chứa “len”. Toán tử @@ kiểm tra vector có khớp query không. ts_rank dùng để sắp xếp kết quả.

Với dữ liệu mẫu gồm “Áo khoác gió” và “Quần jean” (mô tả có chữ “khoác ngoài”), truy vấn khoác trả về:

ProductHit[id=1, name=Áo khoác gió, rank=0.6079271]
ProductHit[id=2, name=Quần jean, rank=0.24317084]

Lưu ý: Người dùng hay gõ không dấu (“ao khoac”). Bạn có thể dùng extension unaccent. Hàm unaccent mặc định không được đánh dấu IMMUTABLE, nên muốn dùng trong generated column hoặc index, bạn cần bọc nó trong một hàm IMMUTABLE riêng. Để tìm kiếm chịu lỗi chính tả, bạn có thể dùng thêm extension pg_trgm.

6. Vector search với pgvector

6.1. Schema

Embedding là một mảng số thực do model AI sinh ra. Văn bản có nghĩa gần nhau sẽ cho vector gần nhau. Bật extension và thêm cột:

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

Ví dụ dùng 3 chiều cho dễ đọc. Thực tế, số chiều phải khớp chính xác với model embedding bạn chọn, thường là vài trăm tới vài nghìn chiều. Index HNSW cho phép tìm lân cận gần đúng rất nhanh. vector_cosine_ops cho biết index phục vụ khoảng cách cosine.

6.2. Lưu và tìm theo vector

Không cần thư viện thêm. Bạn chuyển float[] thành chuỗi dạng [0.1,0.2,0.3] rồi ép kiểu ::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();
    }
}

Toán tử <=> tính khoảng cách cosine. ORDER BY ... LIMIT lấy các sản phẩm gần nhất. Vector truy vấn lấy từ cùng model đã sinh embedding cho sản phẩm, ví dụ khi người dùng gõ “đồ mặc khi trời mưa”.

Vì sản phẩm và embedding nằm chung bảng, bạn kết hợp lọc dễ dàng: WHERE category = 'outerwear' AND price < 500000 ORDER BY embedding <=> ?. Bạn cũng có thể trộn điểm ts_rank với khoảng cách vector để làm hybrid search. Với một vector database riêng, bạn phải đồng bộ các thuộc tính này sang đó.

Lưu ý: Khi kết hợp WHERE có độ chọn lọc cao với index HNSW, số kết quả trả về có thể ít hơn LIMIT. Lý do là index lọc sau khi đã duyệt một tập ứng viên giới hạn. Hãy kiểm tra kỹ với dữ liệu thật và tham khảo các tham số tinh chỉnh của pgvector.

7. Viết test với Testcontainers

Test chạy trên PostgreSQL thật, cùng image có pgvector. @ServiceConnection tự cấu hình 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 ('Áo khoác gió', 'Chống nước nhẹ')");
        jdbc.update("INSERT INTO product (name, description) VALUES ('Quần jean', 'Vải dày')");
        long jacketId = jdbc.queryForObject("SELECT id FROM product WHERE name = 'Áo khoác gió'", Long.class);
        long jeansId = jdbc.queryForObject("SELECT id FROM product WHERE name = 'Quần jean'", 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("khoác"))
                .extracting(ProductHit::name)
                .containsExactly("Áo khoác gió");
        assertThat(vectorSearch.findSimilar(new float[]{1.0f, 0.0f, 0.0f}, 1))
                .containsExactly("Áo khoác gió");
    }
}

Test thứ hai kiểm tra hành vi quan trọng nhất của queue: khi handler lỗi, job không bị mất. Test đặt dữ liệu ngay trong từng method. Ở dự án thật, bạn nên dọn bảng trong @BeforeEach để các test độc lập với nhau.

8. Khi nào KHÔNG nên dùng Postgres cho mọi thứ

Gom mọi thứ vào một chỗ cũng có giá. Bạn nên cân nhắc hệ thống chuyên dụng khi:

• Throughput queue rất cao hoặc cần streaming. Nếu bạn cần replay lại luồng sự kiện, có nhiều consumer group độc lập, hoặc giữ message lâu dài, hãy dùng Kafka hoặc công cụ tương tự. Queue trên Postgres tạo nhiều dead tuple do DELETE/UPDATE liên tục. Bạn phải theo dõi autovacuum cẩn thận.
• Search phức tạp. Faceting nhiều chiều, gợi ý khi gõ, xử lý ngôn ngữ tốt cho tiếng Việt, tinh chỉnh relevance chi tiết: một search engine chuyên dụng làm tốt hơn hẳn.
• Lượng vector rất lớn. Index HNSW cần nhiều RAM. Khi dữ liệu vector chiếm phần lớn tài nguyên, nó tranh chấp với workload giao dịch chính.
• Tải không đồng đều làm ảnh hưởng nhau. Một truy vấn vector nặng hay một đợt job dồn dập có thể làm chậm API thanh toán. Read replica giúp tách một phần tải đọc, nhưng không giải quyết hết.
• Team đã vận hành tốt hệ thống chuyên dụng. Nếu hạ tầng đã có sẵn và ổn định, việc dời mọi thứ về Postgres chỉ để “gọn” chưa chắc đáng công.

Lưu ý: Một chiến lược hợp lý là bọc từng chức năng sau một interface (JobQueue, ProductSearch…). Khi số liệu giám sát cho thấy Postgres là nút thắt, bạn thay phần cài đặt bên dưới mà không phải sửa code nghiệp vụ.

9. Kết luận

PostgreSQL đủ sức đảm nhận job queue nhờ FOR UPDATE SKIP LOCKED, full-text search với tsvector và GIN index, và vector search qua extension pgvector. Lợi ích lớn nhất là ít hệ thống phải vận hành hơn, và có thể ghi dữ liệu nghiệp vụ, job, embedding trong cùng một transaction.

Hãy coi đây là điểm khởi đầu mặc định cho hệ thống nhỏ và vừa, không phải đích đến vĩnh viễn. Bắt đầu với Postgres, đo đạc thật kỹ, và chỉ tách sang công cụ chuyên dụng khi số liệu cho thấy cần thiết.