PostgreSQL macht alles: Queue, Search, Vektor in einer Datenbank
1. Übersicht
Ein typisches System hat oft viele Komponenten: PostgreSQL speichert die Daten, ein Message Broker verwaltet die Queue, eine Search-Engine für die Suche und neuerdings eine Vector-Database für KI-Features. Jede Komponente bringt ihre eigenen Anforderungen an Betrieb, Monitoring, Backup und Datensynchronisation mit sich.
Der Trend „Postgres macht alles” stellt die Gegenfrage: Wenn PostgreSQL die meisten dieser Aufgaben bereits gut genug erfüllt, brauchst du wirklich noch neue Systeme? In diesem Artikel wirst du PostgreSQL als Job-Queue, Full-Text-Search und Vector-Search mit Spring Boot 3.x nutzen. Du wirst auch lernen, wann du damit aufhören und spezialisierte Tools wählen solltest.
2. Warum dieser Trend beachtenswert ist
Es gibt mehrere Hauptgründe:
- Weniger Betriebskomplexität. Jedes zusätzliche System ist etwas, das nachts kaputtgehen kann. Bei kleinen Teams überwiegt dieser Aufwand oft den Nutzen der zusätzlichen Performance.
- Transaktionale Konsistenz. Wenn Bestellungen und E-Mail-Versand-Jobs in der gleichen Datenbank liegen, schreibst du beide in einer Transaction. Es gibt keine Situation, in der die Bestellung gespeichert ist, aber die Message verloren geht, oder umgekehrt.
- Extension-Ökosystem. PostgreSQL hat einen starken Extension-Mechanismus. Zum Beispiel fügt
pgvectoreinen Vektortyp und Indizes für die k-Nearest-Neighbor-Suche hinzu. - KI-Features erfordern Embedding-Speicher. Viele Teams möchten semantische Suchfunktionen hinzufügen, ohne dafür eine neue Datenbank zu bauen.
Der Geist dieses Trends ist mit dem Einfachen beginnen, erst auslagern, wenn die Metriken das rechtfertigen. Das bedeutet nicht, dass PostgreSQL jedes System ersetzen kann.
3. Vorbereitung
3.1. Maven-Abhängigkeiten
Du brauchst JDBC, den PostgreSQL-Treiber und Testcontainers für Tests. Spring Boot verwaltet die Versionen, also musst du sie nicht deklarieren:
<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. PostgreSQL mit pgvector ausführen
Das Image pgvector/pgvector bündelt PostgreSQL und die vector-Extension. Wähle das Tag der PostgreSQL-Major-Version, die du verwendest:
services:
postgres:
image: pgvector/pgvector:pg16
environment:
POSTGRES_DB: shop
POSTGRES_USER: shop
POSTGRES_PASSWORD: secret
ports:
- "5432:5432"Konfiguriere Spring Boot so, dass es sich verbindet und schema.sql beim Start automatisch ausführt:
spring.datasource.url=jdbc:postgresql://localhost:5432/shop
spring.datasource.username=shop
spring.datasource.password=secret
spring.sql.init.mode=alwaysIn echten Projekten solltest du Flyway oder Liquibase statt schema.sql verwenden.
4. Job-Queue mit FOR UPDATE SKIP LOCKED
4.1. Schema
Eine minimale Queue braucht nur eine Tabelle:
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);Die run_at-Spalte ermöglicht es dir, Jobs zeitlich zu planen. Der Index auf run_at hilft dem Worker, den frühesten Job zu erhalten, ohne die gesamte Tabelle zu scannen.
4.2. Job abrufen und verarbeiten
Der Knackpunkt ist die FOR UPDATE SKIP LOCKED-Klausel. Mehrere Worker führen die Abfrage aus. Jeder Worker sperrt eine Zeile, und andere Worker überspringen die bereits gesperrten Zeilen, anstatt zu warten:
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;
}
}Erklärung:
enqueuefügt einen Job ein.?::jsonbcast den JSON-String zumJSONB-Typ.processNextläuft in einer Transaction. DasSELECT ... FOR UPDATE SKIP LOCKEDsperrt genau einen Job, der noch nicht vergeben ist.- Wenn der
handlererfolgreich läuft, wird der Job gelöscht und die Transaction committed. - Falls der
handlereine Exception wirft, rollback die Transaction. Der Job bleibt in der Tabelle und wird erneut verarbeitet. Falls der Worker abstürzt, gibt PostgreSQL die Sperre automatisch frei, wenn die Verbindung geschlossen wird.
Ein Worker kann eine einfache @Scheduled-Methode sein:
@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("E-Mail für Job " + job.id() + " versenden: " + job.payload()))) {
// Weiter, bis keine Jobs mehr übrig sind
}
}
}Vergiss nicht, @EnableScheduling zu einer Konfigurationsklasse hinzuzufügen. Wenn ein Job da ist, gibt die Konsole folgendes aus:
E-Mail für Job 1 versenden: {"orderId": 42, "email": "an@example.com"}Hinweis: Das Behalten der Transaction während der Job-Verarbeitung ist einfach und sicher, wenn der Worker crasht. Im Gegenzug wird die Transaction so lange wie die Verarbeitung dauern. Wenn Jobs lange laufen (langsamer API-Aufruf, große Dateiverarbeitung), solltest du zu einem Modell mit
status- undlocked_until-Spalten übergehen oder auf eine spezialisierte Queue-Bibliothek wechseln. Du solltest auch eineattempts-Spalte hinzufügen, um zu vermeiden, dass ein fehlerhafter Job endlos wiederholt wird.
4.3. Der echte Vorteil: Enqueue in derselben Geschäfts-Transaction
Das ist der stärkste Grund, Postgres als Queue zu nutzen:
@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));
}
}Die Bestellung und der Job werden zusammen committed oder zusammen zurückgerollt. Mit einem separaten Broker müsstest du selbst das Transactional Outbox-Muster implementieren, um diese Garantie zu bekommen. (Das Beispiel nutzt formatted zur Kürze. Echter Code sollte die Payload mit Jackson serialisieren, um Zeichen richtig zu escapen.)
Wenn du nicht möchtest, dass der Worker ständig abfragt, kannst du LISTEN/NOTIFY von PostgreSQL verwenden, um den Worker zu wecken, wenn neue Jobs kommen. Polling sollte als Fallback erhalten bleiben.
5. Volltextsuche mit tsvector und GIN-Index
5.1. Schema
PostgreSQL hat von Haus aus Volltextsuche. Du nutzt eine generierte Spalte, um den Suchvektor automatisch zu aktualisieren, wenn sich die Daten ändern:
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);Erklärung:
to_tsvector('simple', ...)zerlegt Text in Token. Die Konfigurationsimplekonvertiert nur zu Kleinbuchstaben, ohne Stemming. Diese Wahl passt zu Vietnamesisch, da PostgreSQL standardmäßig kein vietnamesisches Wörterbuch hat.setweightweist GewichtAdem Namen undBder Beschreibung zu. Treffer im Namen werden höher bewertet.- GIN-Index ermöglicht schnelle Abfragen auf der
tsvector-Spalte.
5.2. Abfragen
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 akzeptiert Syntax wie eine normale Suchleiste: "áo khoác" -len bedeutet enthält die Phrase „áo khoác” und nicht „len”. Der @@-Operator prüft, ob der Vektor die Query erfüllt. ts_rank wird zum Sortieren der Ergebnisse verwendet.
Mit Beispieldaten „Áo khoác gió” und „Quần jean” (Beschreibung enthält „khoác ngoài”), gibt die Abfrage khoác zurück:
ProductHit[id=1, name=Áo khoác gió, rank=0.6079271]
ProductHit[id=2, name=Quần jean, rank=0.24317084]Hinweis: Benutzer geben oft ohne Diakritika ein („ao khoac”). Du kannst die
unaccent-Extension verwenden. Dieunaccent-Funktion ist nicht standardmäßig alsIMMUTABLEmarkiert, also musst du sie in einer separatenIMMUTABLE-Funktion wrappen, um sie in generated columns oder Indizes zu nutzen. Für Tippfehlertoleranz kannst du zusätzlich diepg_trgm-Extension nutzen.
6. Vektorsuche mit pgvector
6.1. Schema
Embeddings sind ein Array von Floats, das von einem KI-Modell erzeugt wird. Text mit ähnlicher Bedeutung hat ähnliche Vektoren. Aktiviere die Extension und füge eine Spalte hinzu:
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);Das Beispiel nutzt 3 Dimensionen zur Verdeutlichung. In der Praxis muss die Anzahl der Dimensionen exakt mit dem Embedding-Modell deiner Wahl übereinstimmen, üblicherweise einige hundert bis tausend. Ein HNSW-Index ermöglicht sehr schnelle Approximate Nearest Neighbor-Suche. vector_cosine_ops gibt an, dass der Index Cosine-Distanz nutzt.
6.2. Speichern und nach Vektoren suchen
Keine zusätzliche Bibliothek nötig. Du konvertierst float[] zu einem String wie [0.1,0.2,0.3] und castest dann zu ::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();
}
}Der <=>-Operator berechnet die Cosine-Distanz. ORDER BY ... LIMIT holt die nächsten Produkte. Der Query-Vektor kommt vom gleichen Modell, das die Produkt-Embeddings erzeugt hat, zum Beispiel wenn der Benutzer „đồ mặc khi trời mưa” eingibt.
Da Produkte und Embeddings in der gleichen Tabelle liegen, kannst du einfach filtern: WHERE category = 'outerwear' AND price < 500000 ORDER BY embedding <=> ?. Du kannst auch ts_rank mit der Vektor-Distanz kombinieren für Hybrid Search. Mit einer separaten Vector-Datenbank müsstest du diese Attribute dort abgleichen.
Hinweis: Wenn du ein
WHEREmit hoher Selektivität mit einem HNSW-Index kombinierst, kann die Anzahl der Ergebnisse unterLIMITliegen. Der Grund ist, dass der Index nach Durchsuchen eines begrenzten Kandidatensatzes filtert. Teste das gründlich mit echten Daten und konsultiere die Tuning-Parameter von pgvector.
7. Tests mit Testcontainers schreiben
Tests laufen auf echtem PostgreSQL mit pgvector-Image. @ServiceConnection konfiguriert die Datenquelle automatisch:
@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ó");
}
}Der zweite Test überprüft das wichtigste Verhalten der Queue: Wenn der Handler fehlschlägt, geht der Job nicht verloren. Der Test setzt Daten direkt in jeder Methode. In echten Projekten solltest du Tabellen in @BeforeEach aufräumen, damit Tests unabhängig sind.
8. Wann du PostgreSQL NICHT für alles verwenden solltest
Alles an einem Ort zu haben hat Kosten. Du solltest spezialisierte Systeme in Betracht ziehen, wenn:
- Der Queue-Durchsatz ist sehr hoch oder Streaming ist erforderlich. Wenn du Event-Streams replay musst, viele unabhängige Consumer-Gruppen hast oder Messages lange speichern musst, nutze Kafka oder ähnliches. Queues in Postgres erzeugen viele Dead Tuples durch ständige
DELETE/UPDATE. Du musst Autovacuum sorgfältig überwachen. - Komplexe Suche. Multi-dimensionales Faceting, Suchvorschläge, gute Sprachunterstützung für Vietnamesisch, detailliertes Relevanz-Tuning – eine spezialisierte Search Engine ist weitaus besser.
- Sehr große Vektormengen. HNSW-Indizes brauchen viel RAM. Wenn Vektor-Daten die meisten Ressourcen beanspruchen, konkurrieren sie mit deiner Transaktions-Workload.
- Ungleichmäßige Last beeinträchtigt sich gegenseitig. Eine schwere Vektor-Abfrage oder ein Batch-Job kann die Payment-API verlangsamen. Read-Replicas helfen, etwas Read-Last zu verteilen, aber lösen das Problem nicht vollständig.
- Das Team führt spezialisierte Systeme bereits erfolgreich aus. Wenn die Infrastruktur bereits vorhanden und stabil ist, ist alles zu PostgreSQL umzuziehen, nur weil es „einfacher” wirkt, möglicherweise nicht sinnvoll.
Hinweis: Eine sinnvolle Strategie ist, jede Funktionalität hinter einer Schnittstelle zu verbergen (
JobQueue,ProductSearch…). Wenn Monitoring zeigt, dass Postgres der Engpass ist, kannst du die Implementierung darunter austauschen, ohne den Business-Code zu ändern.
9. Fazit
PostgreSQL ist durchaus in der Lage, Job-Queues mit FOR UPDATE SKIP LOCKED, Volltextsuche mit tsvector und GIN-Index sowie Vector-Search über die pgvector-Extension zu handhaben. Der größte Vorteil ist weniger Infrastruktur zum Betreiben und die Möglichkeit, Business-Daten, Jobs und Embeddings in derselben Transaction zu schreiben.
Betrachte das als einen sinnvollen Startpunkt für kleine bis mittlere Systeme, nicht als endgültiges Ziel. Beginne mit Postgres, miss wirklich genau, und gehe erst zu spezialisierten Tools über, wenn die Metriken das wirklich rechtfertigen.
Gefällt dir diese Art von Erklärung? Cris The Dev veröffentlicht jeden Tag ein kurzes Video auf YouTube und TikTok.
Mehr zu diesem Thema
AI Code Review: so nutzt du es effektiv und sicher
Verstehe, was AI Code Review ist, integriere es in deine Review- und CI-Pipeline, verstecke Secrets vor dem Versand von Diffs, wehre Prompt Injection ab und erkenne, wann du es nicht nutzen solltest.
Local-first software: App läuft offline und synchronisiert mit CRDT
Erfahren Sie, was Local-first Software ist, warum CRDT Apps offline zum Laufen bringt und Daten automatisch zusammenführt – mit Java 21 und Spring Boot 3 Beispielen.
Observability für LLM-Anwendungen: Monitoring von Prompts, Kosten und Qualität
So messen Sie Latenz, Token, Kosten und Antwortqualität von LLM-Anwendungen mit Micrometer, OpenTelemetry und Spring Boot 3.
Agent braucht Dokumentation statt Gedächtnis: Repos für Menschen und Agenten
Coding-Agenten vergessen den Kontext nach jeder Sitzung. Dieser Artikel zeigt, wie man Konventionen ins Repository bringt – mit AGENTS.md, Verifizierungsskripten, ArchUnit und Tests, die Mensch und Agent verstehen.
Edge und Serverless hybrid: Code in der Nähe von Nutzern ausführen
Erfahren Sie mehr über das Hybrid-Modell zwischen Edge und Serverless: welcher Teil sollte nah bei Benutzern laufen, welcher im Origin. Mit Beispielen aus Spring Boot, Spring Cloud Function und Edge Functions.