你有没有发现：PostgreSQL可以做到Redis能做的一切

一、设置：我使用Redis的目的

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

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

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

二、我为什么考虑替换Redis

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

Postgres backup ✅Postgres monitoring ✅Postgres failover ✅

// Update databaseawait db.query('UPDATE users SET name = $1 WHERE id = $2', [name, id]);
// Invalidate cacheawait redis.del(`user:${id}`);
// ⚠️ What if Redis is down?// ⚠️ What if this fails?// Now cache and DB are out of sync

三、PostgreSQL特性

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

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 cacheWHERE key = $1 AND expires_at > NOW();

DELETE FROM cache WHERE expires_at < NOW();

Redis SET: 0.05msPostgres UNLOGGED INSERT: 0.08ms

// Publisherredis.publish('notifications', JSON.stringify({ userId: 123, msg: 'Hello' }));
// Subscriberredis.subscribe('notifications');redis.on('message', (channel, message) => {  console.log(message);});

-- PublisherNOTIFY 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-2msPostgres NOTIFY latency: 2-5ms

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

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_insertedAFTER INSERT ON logsFOR 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`);  });});

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

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 jobsSET attempts = attempts + 1FROM next_jobWHERE jobs.id = next_job.idRETURNING *;

Redis BRPOP: 0.1msPostgres SKIP LOCKED: 0.3ms

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

CREATE TABLE rate_limits (  user_id INT PRIMARY KEY,  request_count INT DEFAULT 0,  window_start TIMESTAMPTZ DEFAULT NOW());
-- Check and incrementWITH 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_limitsSET   request_count = (SELECT new_count FROM current),  window_start = CASE    WHEN window_start < NOW() - INTERVAL '1 minute'    THEN NOW()    ELSE window_start  ENDWHERE user_id = $1RETURNING request_count;

CREATE TABLE api_requests (  user_id INT NOT NULL,  created_at TIMESTAMPTZ DEFAULT NOW());
-- Check rate limitSELECT COUNT(*) FROM api_requestsWHERE user_id = $1  AND created_at > NOW() - INTERVAL '1 minute';
  -- If under limit, insertINSERT INTO api_requests (user_id) VALUES ($1);
-- Cleanup old requests periodicallyDELETE FROM api_requests WHERE created_at < NOW() - INTERVAL '5 minutes';

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

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/UpdateINSERT 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;
      -- ReadSELECT data FROM sessionsWHERE id = $1 AND expires_at > NOW();

-- Find all sessions for a specific userSELECT * FROM sessionsWHERE data->>'userId' = '123';
-- Find sessions with specific roleSELECT * FROM sessionsWHERE data->'user'->>'role' = 'admin';

四、实际生产环境基准测试

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

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

五、哪些场景仍建议保留Redis

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

-- Leaderboard in Postgres (slower)SELECT user_id, scoreFROM leaderboardORDER BY score DESCLIMIT 10;
-- vs RedisZREVRANGE leaderboard 0 9 WITHSCORES

六、迁移方案

// Write to bothawait redis.set(key, value);await pg.query('INSERT INTO cache ...');
// Read from Redis (still primary)let data = await redis.get(key);

// Try Postgres firstlet data = await pg.query('SELECT value FROM cache WHERE key = $1', [key]);
// Fallback to Redisif (!data) {  data = await redis.get(key);}

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

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

七、代码示例：完整实现

// cache.jsclass 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.jsclass 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.jsclass PostgresQueue {  constructor(pool) {    this.pool = pool;  }
    async enqueue(queue, payload, scheduledAt = newDate()) {    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;

八、性能优化技巧

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

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;

# postgresql.confshared_buffers = 2GB           # 25% of RAMeffective_cache_size = 6GB     # 75% of RAMwork_mem = 50MB                # For complex queriesmaintenance_work_mem = 512MB   # For VACUUM

-- Run dailyVACUUM ANALYZE cache;VACUUM ANALYZE jobs;
-- Or enable autovacuum (recommended)ALTER TABLE cache SET (autovacuum_vacuum_scale_factor = 0.1);

九、三个月后的结果

十、决策矩阵

十一、参考资料

十二、最后