我的 Web 應用技術堆疊很典型：

• PostgreSQL 用於持久化資料

• Redis 用於快取、發布/訂閱和後台作業

兩個資料庫。兩件事需要管理。兩個故障點。

然後我意識到： PostgreSQL 可以做到 Redis 所能做的一切。

我徹底移除了Redis。事情經過是這樣的。

設定：我使用 Redis 的目的

在更改之前，Redis 主要處理三件事：

1. 快取（佔使用量的 70%）

// Cache API responses
await redis.set(`user:${id}`, JSON.stringify(user), 'EX', 3600);

2. 發布/訂閱（佔使用量的 20%）

// Real-time notifications
redis.publish('notifications', JSON.stringify({ userId, message }));

3. 後台作業佇列（使用率 10%）

// Using Bull/BullMQ
queue.add('send-email', { to, subject, body });

痛點：

• 需要備份兩個資料庫

• Redis 使用記憶體（規模化時成本很高）

• Redis 持久化機制…很複雜。

• Postgres 和 Redis 之間的網路跳躍

為什麼我會考慮替換 Redis

原因一：成本

我的 Redis 配置：

• AWS ElastiCache：每月 45 美元（2GB）

• 升級到 5GB 流量每月需花費 110 美元。

PostgreSQL：

• 已付費使用 RDS：每月 50 美元（20GB 儲存空間）

• 增加 5GB 資料流量：每月 0.50 美元

潛在節省：每月約 100 美元

原因二：營運複雜性

使用 Redis：

Postgres backup ✅
Redis backup ❓ (RDB? AOF? Both?)
Postgres monitoring ✅
Redis monitoring ❓
Postgres failover ✅
Redis Sentinel/Cluster ❓

不使用 Redis：

Postgres backup ✅
Postgres monitoring ✅
Postgres failover ✅

少了一個活動部件。

原因三：資料一致性

經典問題：

// Update database
await db.query('UPDATE users SET name = $1 WHERE id = $2', [name, id]);

// Invalidate cache
await redis.del(`user:${id}`);

// ⚠️ What if Redis is down?
// ⚠️ What if this fails?
// Now cache and DB are out of sync

Postgres 的所有功能都透過事務解決了這個問題。

PostgreSQL 功能 #1：使用未記錄表進行快取

Redis：

await redis.set('session:abc123', JSON.stringify(sessionData), 'EX', 3600);

PostgreSQL：

CREATE UNLOGGED TABLE cache (
  key TEXT PRIMARY KEY,
  value JSONB NOT NULL,
  expires_at TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_cache_expires ON cache(expires_at);

插入：

INSERT INTO cache (key, value, expires_at)
VALUES ($1, $2, NOW() + INTERVAL '1 hour')
ON CONFLICT (key) DO UPDATE
  SET value = EXCLUDED.value,
      expires_at = EXCLUDED.expires_at;

讀：

SELECT value FROM cache
WHERE key = $1 AND expires_at > NOW();

清理（定期執行）：

DELETE FROM cache WHERE expires_at < NOW();

什麼是未記錄資料？

未記錄的表：

• 跳過預寫式日誌（WAL）

• 寫入速度快得多。

• 不要在崩潰後保留資料（非常適合快取！）

表現：

Redis SET: 0.05ms
Postgres UNLOGGED INSERT: 0.08ms

距離夠近，可以快取。

PostgreSQL 功能 #2：附有 LISTEN/NOTIFY 的發布/訂閱

接下來就精彩了。

PostgreSQL 具有原生的發布/訂閱功能，但大多數開發人員並不了解。

Redis 發佈/訂閱

// Publisher
redis.publish('notifications', JSON.stringify({ userId: 123, msg: 'Hello' }));

// Subscriber
redis.subscribe('notifications');
redis.on('message', (channel, message) => {
  console.log(message);
});

PostgreSQL 發布/訂閱

-- Publisher
NOTIFY notifications, '{"userId": 123, "msg": "Hello"}';

// Subscriber (Node.js with pg)
const client = new Client({ connectionString: process.env.DATABASE_URL });
await client.connect();

await client.query('LISTEN notifications');

client.on('notification', (msg) => {
  const payload = JSON.parse(msg.payload);
  console.log(payload);
});

性能比較：

Redis pub/sub latency: 1-2ms
Postgres NOTIFY latency: 2-5ms

速度稍慢，但是：

• 無需額外基礎設施

• 可用於交易

• 可以與查詢結合使用

現實世界的例子：活尾巴

我的日誌管理應用程式需要即時日誌流。

使用 Redis：

// When new log arrives
await db.query('INSERT INTO logs ...');
await redis.publish('logs:new', JSON.stringify(log));

// Frontend listens
redis.subscribe('logs:new');

問題：有兩個操作。如果發布失敗怎麼辦？

使用 PostgreSQL：

CREATE FUNCTION notify_new_log() RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify('logs_new', row_to_json(NEW)::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER log_inserted
AFTER INSERT ON logs
FOR EACH ROW EXECUTE FUNCTION notify_new_log();

現在是原子操作了。插入和通知要么同時發生，要么都不發生。

// Frontend (via SSE)
app.get('/logs/stream', async (req, res) => {
  const client = await pool.connect();

  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
  });

  await client.query('LISTEN logs_new');

  client.on('notification', (msg) => {
    res.write(`data: ${msg.payload}\n\n`);
  });
});

結果：無需 Redis 即可實現即時日誌串流傳輸。

PostgreSQL 功能 #3：帶有 SKIP LOCKED 的作業佇列

Redis（使用 Bull/BullMQ）：

queue.add('send-email', { to, subject, body });

queue.process('send-email', async (job) => {
  await sendEmail(job.data);
});

PostgreSQL：

CREATE TABLE jobs (
  id BIGSERIAL PRIMARY KEY,
  queue TEXT NOT NULL,
  payload JSONB NOT NULL,
  attempts INT DEFAULT 0,
  max_attempts INT DEFAULT 3,
  scheduled_at TIMESTAMPTZ DEFAULT NOW(),
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_jobs_queue ON jobs(queue, scheduled_at) 
WHERE attempts < max_attempts;

隊列：

INSERT INTO jobs (queue, payload)
VALUES ('send-email', '{"to": "[email protected]", "subject": "Hi"}');

工作進程（出隊）：

WITH next_job AS (
  SELECT id FROM jobs
  WHERE queue = $1
    AND attempts < max_attempts
    AND scheduled_at <= NOW()
  ORDER BY scheduled_at
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
UPDATE jobs
SET attempts = attempts + 1
FROM next_job
WHERE jobs.id = next_job.id
RETURNING *;

神奇之處FOR UPDATE SKIP LOCKED

這使得PostgreSQL成為一個無鎖佇列：

• 多個工人可以同時拉取任務。

• 任何作業都不會被處理兩次。

• 如果某個工作進程崩潰，則該任務將再次可用。

表現：

Redis BRPOP: 0.1ms
Postgres SKIP LOCKED: 0.3ms

對於大多數工作負載而言，差異可以忽略不計。

PostgreSQL 特性 #4：速率限制

Redis（經典速率限制器）：

const key = `ratelimit:${userId}`;
const count = await redis.incr(key);
if (count === 1) {
  await redis.expire(key, 60); // 60 seconds
}

if (count > 100) {
  throw new Error('Rate limit exceeded');
}

PostgreSQL：

CREATE TABLE rate_limits (
  user_id INT PRIMARY KEY,
  request_count INT DEFAULT 0,
  window_start TIMESTAMPTZ DEFAULT NOW()
);

-- Check and increment
WITH current AS (
  SELECT 
    request_count,
    CASE 
      WHEN window_start < NOW() - INTERVAL '1 minute'
      THEN 1  -- Reset counter
      ELSE request_count + 1
    END AS new_count
  FROM rate_limits
  WHERE user_id = $1
  FOR UPDATE
)
UPDATE rate_limits
SET 
  request_count = (SELECT new_count FROM current),
  window_start = CASE
    WHEN window_start < NOW() - INTERVAL '1 minute'
    THEN NOW()
    ELSE window_start
  END
WHERE user_id = $1
RETURNING request_count;

或者用視窗函數更簡單：

CREATE TABLE api_requests (
  user_id INT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Check rate limit
SELECT COUNT(*) FROM api_requests
WHERE user_id = $1
  AND created_at > NOW() - INTERVAL '1 minute';

-- If under limit, insert
INSERT INTO api_requests (user_id) VALUES ($1);

-- Cleanup old requests periodically
DELETE FROM api_requests WHERE created_at < NOW() - INTERVAL '5 minutes';

Postgres 的優點：

• 需要基於複雜的邏輯進行速率限制（而不僅僅是計數）。

• 希望將速率限制資料與業務邏輯放在同一事務中

Redis 的優勢時刻：

• 需要亞毫秒速率限制

• 極高的吞吐量（每秒數百萬個請求）

PostgreSQL 特性 #5：使用 JSONB 的會話

Redis：

await redis.set(`session:${sessionId}`, JSON.stringify(sessionData), 'EX', 86400);

PostgreSQL：

CREATE TABLE sessions (
  id TEXT PRIMARY KEY,
  data JSONB NOT NULL,
  expires_at TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_sessions_expires ON sessions(expires_at);

-- Insert/Update
INSERT INTO sessions (id, data, expires_at)
VALUES ($1, $2, NOW() + INTERVAL '24 hours')
ON CONFLICT (id) DO UPDATE
  SET data = EXCLUDED.data,
      expires_at = EXCLUDED.expires_at;

-- Read
SELECT data FROM sessions
WHERE id = $1 AND expires_at > NOW();

附加內容：JSONB 運算符

您可以在會話內部進行查詢：

-- Find all sessions for a specific user
SELECT * FROM sessions
WHERE data->>'userId' = '123';

-- Find sessions with specific role
SELECT * FROM sessions
WHERE data->'user'->>'role' = 'admin';

你用 Redis 做不到這一點！

真實世界基準測試

我使用生產資料集執行了基準測試：

測試設定

• 硬體： AWS RDS db.t3.medium（2 個虛擬 CPU，4GB 記憶體）

• 資料集： 100 萬個快取條目，1 萬個會話

• 工具： pgbench（自訂腳本）

結果

| 操作 | Redis | PostgreSQL | 區別 |

|-----------|-------|------------|------------|

|快取設定| 0.05毫秒 | 0.08毫秒 | 速度降低60% |

|快取 GET | 0.04毫秒 | 0.06毫秒 | 速度降低50% |

|發佈/訂閱| 1.2毫秒 | 3.1毫秒 | 速度降低158% |

|佇列推送| 0.08毫秒 | 0.15毫秒 | 速度降低87% |

|佇列彈出| 0.12毫秒 | 0.31毫秒 | 速度降低158% |

PostgreSQL 速度較慢…但是：

• 所有操作均在 1 毫秒以內。

• 消除到 Redis 的網路跳轉

• 降低基礎設施複雜性

聯合行動（真正的勝利）

場景：插入資料 + 使快取失效 + 通知訂閱者

使用 Redis：

await db.query('INSERT INTO posts ...');       // 2ms
await redis.del('posts:latest');                // 1ms (network hop)
await redis.publish('posts:new', data);         // 1ms (network hop)
// Total: ~4ms

使用 PostgreSQL：

BEGIN;
INSERT INTO posts ...;                          -- 2ms
DELETE FROM cache WHERE key = 'posts:latest';  -- 0.1ms (same connection)
NOTIFY posts_new, '...';                        -- 0.1ms (same connection)
COMMIT;
-- Total: ~2.2ms

PostgreSQL 在合併操作時速度更快。

何時保留 Redis

若符合以下條件，請勿替換 Redis：

1. 你需要極致的性能

Redis: 100,000+ ops/sec (single instance)
Postgres: 10,000-50,000 ops/sec

如果你每秒執行數百萬次快取讀取操作，那就繼續使用 Redis。

2. 你正在使用 Redis 特有的資料結構

Redis 具有：

• 已排序集合（排行榜）

• HyperLogLog（唯一計數估計）

• 地理空間指數

• 串流媒體（高級發布/訂閱）

Postgres也有類似的功能，但比較笨拙：

-- Leaderboard in Postgres (slower)
SELECT user_id, score
FROM leaderboard
ORDER BY score DESC
LIMIT 10;

-- vs Redis
ZREVRANGE leaderboard 0 9 WITHSCORES

3. 您有單獨的快取層需求

如果你的架構要求使用單獨的快取層（例如，微服務），請保留 Redis。

移民策略

不要一夜之間就徹底放棄 Redis。以下是我的做法：

第一階段：並排練習（第 1 週）

// Write to both
await redis.set(key, value);
await pg.query('INSERT INTO cache ...');

// Read from Redis (still primary)
let data = await redis.get(key);

監控：比較命中率和延遲。

第二階段：從Postgres讀取資料（第二週）

// Try Postgres first
let data = await pg.query('SELECT value FROM cache WHERE key = $1', [key]);

// Fallback to Redis
if (!data) {
  data = await redis.get(key);
}

監控：錯誤率、效能。

第三階段：僅寫入Postgres資料庫（第三週）

// Only write to Postgres
await pg.query('INSERT INTO cache ...');

監控：一切正常嗎？

第四階段：移除 Redis（第四週）

# Turn off Redis
# Watch for errors
# Nothing breaks? Success!

程式碼範例：完整實現

快取模組（PostgreSQL）

// cache.js
class PostgresCache {
  constructor(pool) {
    this.pool = pool;
  }

  async get(key) {
    const result = await this.pool.query(
      'SELECT value FROM cache WHERE key = $1 AND expires_at > NOW()',
      [key]
    );
    return result.rows[0]?.value;
  }

  async set(key, value, ttlSeconds = 3600) {
    await this.pool.query(
      `INSERT INTO cache (key, value, expires_at)
       VALUES ($1, $2, NOW() + INTERVAL '${ttlSeconds} seconds')
       ON CONFLICT (key) DO UPDATE
         SET value = EXCLUDED.value,
             expires_at = EXCLUDED.expires_at`,
      [key, value]
    );
  }

  async delete(key) {
    await this.pool.query('DELETE FROM cache WHERE key = $1', [key]);
  }

  async cleanup() {
    await this.pool.query('DELETE FROM cache WHERE expires_at < NOW()');
  }
}

module.exports = PostgresCache;

發布/訂閱模組

// pubsub.js
class PostgresPubSub {
  constructor(pool) {
    this.pool = pool;
    this.listeners = new Map();
  }

  async publish(channel, message) {
    const payload = JSON.stringify(message);
    await this.pool.query('SELECT pg_notify($1, $2)', [channel, payload]);
  }

  async subscribe(channel, callback) {
    const client = await this.pool.connect();

    await client.query(`LISTEN ${channel}`);

    client.on('notification', (msg) => {
      if (msg.channel === channel) {
        callback(JSON.parse(msg.payload));
      }
    });

    this.listeners.set(channel, client);
  }

  async unsubscribe(channel) {
    const client = this.listeners.get(channel);
    if (client) {
      await client.query(`UNLISTEN ${channel}`);
      client.release();
      this.listeners.delete(channel);
    }
  }
}

module.exports = PostgresPubSub;

作業佇列模組

// queue.js
class PostgresQueue {
  constructor(pool) {
    this.pool = pool;
  }

  async enqueue(queue, payload, scheduledAt = new Date()) {
    await this.pool.query(
      'INSERT INTO jobs (queue, payload, scheduled_at) VALUES ($1, $2, $3)',
      [queue, payload, scheduledAt]
    );
  }

  async dequeue(queue) {
    const result = await this.pool.query(
      `WITH next_job AS (
        SELECT id FROM jobs
        WHERE queue = $1
          AND attempts < max_attempts
          AND scheduled_at <= NOW()
        ORDER BY scheduled_at
        LIMIT 1
        FOR UPDATE SKIP LOCKED
      )
      UPDATE jobs
      SET attempts = attempts + 1
      FROM next_job
      WHERE jobs.id = next_job.id
      RETURNING jobs.*`,
      [queue]
    );

    return result.rows[0];
  }

  async complete(jobId) {
    await this.pool.query('DELETE FROM jobs WHERE id = $1', [jobId]);
  }

  async fail(jobId, error) {
    await this.pool.query(
      `UPDATE jobs
       SET attempts = max_attempts,
           payload = payload || jsonb_build_object('error', $2)
       WHERE id = $1`,
      [jobId, error.message]
    );
  }
}

module.exports = PostgresQueue;

性能調校技巧

1. 使用連線池

const { Pool } = require('pg');

const pool = new Pool({
  max: 20,  // Max connections
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
});

2. 加入合適的索引

CREATE INDEX CONCURRENTLY idx_cache_key ON cache(key) WHERE expires_at > NOW();
CREATE INDEX CONCURRENTLY idx_jobs_pending ON jobs(queue, scheduled_at) 
  WHERE attempts < max_attempts;

3. 調整 PostgreSQL 配置

# postgresql.conf
shared_buffers = 2GB           # 25% of RAM
effective_cache_size = 6GB     # 75% of RAM
work_mem = 50MB                # For complex queries
maintenance_work_mem = 512MB   # For VACUUM

4. 定期維護

-- Run daily
VACUUM ANALYZE cache;
VACUUM ANALYZE jobs;

-- Or enable autovacuum (recommended)
ALTER TABLE cache SET (autovacuum_vacuum_scale_factor = 0.1);

三個月後的結果

我省下了：

• ✅ 每月 100 美元（不再使用 ElastiCache）

• ✅ 備份複雜度降低 50%

• ✅ 少監控一項服務

• ✅ 更簡單的部署（少一個依賴項）

我失去了什麼：

• ❌ 快取操作延遲約 0.5 毫秒

• ❌ Redis 的奇特資料結構（其實不需要）

我會再做一次嗎？是的，就這種使用場景而言。

我會向所有人推薦嗎？不會。

決策矩陣

如果符合以下條件，請將 Redis 替換為 Postgres：

• ✅ 你使用 Redis 只是為了簡單的快取/會話管理。

• ✅ 快取命中率低於 95%（寫入次數過多）

• ✅ 您需要事務一致性

• ✅ 您可以接受操作速度慢 0.1-1 毫秒。

• ✅ 你們團隊規模小，營運資源有限。

若符合以下條件，則保留 Redis：

• ❌ 你需要每秒 10 萬次以上的操作。

• ❌ 您使用了 Redis 資料結構（有序集合等）。

• ❌ 您擁有專門的維運團隊

• ❌ 亞毫秒延遲至關重要

• ❌ 你正在進行地理複製。

資源

PostgreSQL 功能：

• 監聽/通知文件

• 跳過鎖定

• 未記錄表

工具：

• pgBouncer - 連線池

• pg_stat_statements - 查詢效能

其他解決方案：

• Graphile Worker - 基於 Postgres 的作業佇列

• pg-boss - 另一個 Postgres 隊列

太長不看

我用 PostgreSQL 取代了 Redis，原因是：

1. 快取 → 未記錄表

2. 發布/訂閱 → 收聽/通知

3. 作業佇列 → 跳過鎖定

4. 會話 → JSONB 表

結果：

• 每月省 100 美元

• 降低操作複雜性

• 稍慢一些（0.1-1毫秒），但可以接受。

• 保證交易一致性

何時這樣做：

• 中小型應用

• 簡單的快取需求

• 希望減少活動部件

何時不應這樣做：

• 高效能需求（每秒 10 萬次以上操作）

• 使用 Redis 特有的功能

• 擁有專門的維運團隊

你是否用 Postgres 取代了 Redis（或反過來）？你的體驗如何？請在評論區分享你的基準測試結果！ 👇

PS：想了解「PostgreSQL 的隱藏功能」或「Redis 實際上何時更勝一籌」的後續內容嗎？請告訴我！

原文出處：https://dev.to/polliog/i-replaced-redis-with-postgresql-and-its-faster-4942