Tôi Đã Thay Thế Redis Bằng PostgreSQL (Và Nó Nhanh Hơn)

Tôi đã có một ngăn xếp ứng dụng web điển hình:

• PostgreSQL cho dữ liệu bền vững
• Redis để lưu đệm, pub/sub và các công việc nền

Hai cơ sở dữ liệu.
Hai thứ cần quản lý.
Hai điểm dễ hỏng.

Sau đó tôi nhận ra:PostgreSQL có thể làm mọi thứ Redis làm.

Tôi đã loại bỏ hoàn toàn Redis.
Đây là những gì đã xảy ra.

Thiết lập:
Tôi Đã Dùng Redis Để Làm Gì

Trước khi thay đổi, Redis xử lý ba việc:

1. Lưu đệm (70% mức sử dụng)

// Lưu đệm phản hồi APIawaitredis.set(`user:${id}`,JSON.stringify(user),'EX',3600);

2. Pub/Sub (20% mức sử dụng)

// Thông báo thời gian thựcredis.publish('notifications',JSON.stringify({userId,message}));

3. Hàng đợi Công việc Nền (10% mức sử dụng)

// Sử dụng Bull/BullMQqueue.add('send-email',{to,subject,body});

Những điểm đau:

• Hai cơ sở dữ liệu cần sao lưu
• Redis sử dụng RAM (tốn kém ở quy mô lớn)
• Tính bền vững của Redis thì…
  phức tạp
• Một bước nhảy mạng giữa Postgres và Redis

Tại Sao Tôi Cân Nhắc Thay Thế Redis

Lý do #1:
Chi phí

Thiết lập Redis của tôi:

• AWS ElastiCache:
  $45/tháng (2GB)
• Tăng lên 5GB sẽ tốn $110/tháng

PostgreSQL:

• Đã trả tiền cho RDS:
  $50/tháng (20GB lưu trữ)
• Thêm 5GB dữ liệu:
  $0.50/tháng

Khoản tiết kiệm tiềm năng:~$100/tháng

Lý do #2:
Độ phức tạp vận hành

Với Redis:

Sao lưu Postgres ✅ Sao lưu Redis ❓ (RDB? AOF? Cả hai?) Giám sát Postgres ✅ Giám sát Redis ❓ Chuyển đổi dự phòng Postgres ✅ Sentinel/Cluster Redis ❓ 

Không có Redis:

Sao lưu Postgres ✅ Giám sát Postgres ✅ Chuyển đổi dự phòng Postgres ✅ 

Bớt đi một thành phần động.

Lý do #3:
Tính nhất quán dữ liệu

Vấn đề kinh điển:

// Cập nhật cơ sở dữ liệuawaitdb.query('UPDATE users SET name = $1 WHERE id = $2',[name,id]);// Vô hiệu hóa bộ nhớ đệmawaitredis.del(`user:${id}`);// ⚠️ Nếu Redis bị sập thì sao?// ⚠️ Nếu việc này thất bại thì sao?// Giờ thì bộ nhớ đệm và DB không đồng bộ

Với mọi thứ trong Postgres:giao dịch giải quyết việc này.

Tính năng PostgreSQL #1:
Lưu đệm với Bảng UNLOGGED

Redis:

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

PostgreSQL:

CREATEUNLOGGEDTABLEcache(keyTEXTPRIMARYKEY,valueJSONBNOTNULL,expires_atTIMESTAMPTZNOTNULL);CREATEINDEXidx_cache_expiresONcache(expires_at);

Chèn:

INSERTINTOcache(key,value,expires_at)VALUES($1,$2,NOW()+INTERVAL'1 hour')ONCONFLICT(key)DOUPDATESETvalue=EXCLUDED.value,expires_at=EXCLUDED.expires_at;

Đọc:

SELECTvalueFROMcacheWHEREkey=$1ANDexpires_at>NOW();

Dọn dẹp (chạy định kỳ):

DELETEFROMcacheWHEREexpires_at<NOW();

UNLOGGED là gì?

Bảng UNLOGGED:

• Bỏ qua Nhật ký Ghi trước (WAL)
• Ghi nhanh hơn nhiều
• Không tồn tại sau sự cố (hoàn hảo cho bộ nhớ đệm!)

Redis SET: 0.05ms Postgres UNLOGGED INSERT: 0.08ms 

// Nhà xuất bảnredis.publish('notifications',JSON.stringify({userId:123,msg:'Xin chào'}));// Người đăng kýredis.subscribe('notifications');redis.on('message',(channel,message)=>{console.log(message);});

-
- Nhà xuất bảnNOTIFYnotifications,'{"userId":
123, "msg":
"Xin chào"}';

// Người đăng ký (Node.js với pg)constclient=newClient({connectionString:process.env.DATABASE_URL});awaitclient.connect();awaitclient.query('LISTEN notifications');client.on('notification',(msg)=>{constpayload=JSON.parse(msg.payload);console.log(payload);});

Độ trễ Redis pub/sub: 1-2ms Độ trễ Postgres NOTIFY: 2-5ms 

// Khi có nhật ký mới đếnawaitdb.query('INSERT INTO logs ...');awaitredis.publish('logs:new',JSON.stringify(log));// Frontend lắng ngheredis.subscribe('logs:new');

CREATEFUNCTIONnotify_new_log()RETURNSTRIGGERAS$$BEGINPERFORMpg_notify('logs_new',row_to_json(NEW)::text);RETURNNEW;END;$$LANGUAGEplpgsql;CREATETRIGGERlog_insertedAFTERINSERTONlogsFOREACHROWEXECUTEFUNCTIONnotify_new_log();

// Frontend (qua SSE)app.get('/logs/stream',async(req,res)=>{constclient=awaitpool.connect();res.writeHead(200,{'Content-Type':'text/event-stream','Cache-Control':'no-cache',});awaitclient.query('LISTEN logs_new');client.on('notification',(msg)=>{res.write(`data:${msg.payload}\n\n`);});});

hàng đợi.thêm('gửi-email',{đến,chủ đề,nội dung});hàng đợi.xử lý('gửi-email',async(công việc)=>{awaitgửiEmail(công việc.dữ liệu);});

TẠOBẢNGcông_việc(idBIGSERIALKHÓA CHÍNH,hàng_đợiVĂN BẢNKHÔNG NULL,tải_trọngJSONBKHÔNG NULL,lần_thửSỐ NGUYÊNMẶC ĐỊNH0,lần_thử_tối_đaSỐ NGUYÊNMẶC ĐỊNH3,lên_lịch_vàoTHỜI GIAN CÓ MÚI GIỜMẶC ĐỊNHBÂY GIỜ(),tạo_vàoTHỜI GIAN CÓ MÚI GIỜMẶC ĐỊNHBÂY GIỜ());TẠOCHỈ MỤCidx_công_việc_hàng_đợiTRÊNcông_việc(hàng_đợi,lên_lịch_vào)KHIlần_thử<lần_thử_tối_đa;

CHÈNVÀOcông_việc(hàng_đợi,tải_trọng)GIÁ TRỊ('gửi-email','{"đến":
"user@example.com", "chủ đề":
"Xin chào"}');

VỚIcông_việc_tiếp_theoNHƯ(CHỌNidTỪcông_việcNƠIhàng_đợi=$1VÀlần_thử<lần_thử_tối_đaVÀlên_lịch_vào<=BÂY GIỜ()SẮP XẾP THEOlên_lịch_vàoGIỚI HẠN1CHOCẬP NHẬTBỎ QUAĐÃ KHÓA)CẬP NHẬTcông_việcĐẶTlần_thử=lần_thử+1TỪcông_việc_tiếp_theoNƠIcông_việc.id=công_việc_tiếp_theo.idTRẢ VỀ*;

Redis BRPOP: 0.1ms Postgres BỎ QUA ĐÃ KHÓA: 0.3ms 

constkhóa=`giới_hạn_tốc_độ:${id_người_dùng}`;constsố_lượng=awaitredis.tăng(khóa);if(số_lượng===1){awaitredis.hết_hạn(khóa,60);// 60 giây}if(số_lượng>100){thrownewLỗi('Đã vượt quá giới hạn tốc độ');}

TẠOBẢNGgiới_hạn_tốc_độ(id_người_dùngSỐ NGUYÊNKHÓA CHÍNH,số_lượng_yêu_cầuSỐ NGUYÊNMẶC ĐỊNH0,bắt_đầu_cửa_sổTHỜI GIAN CÓ MÚI GIỜMẶC ĐỊNHBÂY GIỜ());-
- Kiểm tra và tăngVỚIhiện_tạiNHƯ(CHỌNsố_lượng_yêu_cầu,TRƯỜNG HỢPKHIbắt_đầu_cửa_sổ<BÂY GIỜ()-KHOẢNG THỜI GIAN'1 phút'THÌ1-
- Đặt lại bộ đếmKHÁCsố_lượng_yêu_cầu+1KẾT THÚCNHƯsố_lượng_mớiTỪgiới_hạn_tốc_độNƠIid_người_dùng=$1CHOCẬP NHẬT)CẬP NHẬTgiới_hạn_tốc_độĐẶTsố_lượng_yêu_cầu=(CHỌNsố_lượng_mớiTỪhiện_tại),bắt_đầu_cửa_sổ=TRƯỜNG HỢPKHIbắt_đầu_cửa_sổ<BÂY GIỜ()-KHOẢNG THỜI GIAN'1 phút'THÌBÂY GIỜ()KHÁCbắt_đầu_cửa_sổKẾT THÚCNƠIid_người_dùng=$1TRẢ VỀsố_lượng_yêu_cầu;

TẠOBẢNGapi_requests(user_idSỐ NGUYÊNKHÔNG ĐƯỢC NULL,created_atTHỜI GIAN CÓ MÚI GIỜMẶC ĐỊNHBÂY GIỜ());-
- Kiểm tra giới hạn tốc độCHỌNĐẾM(*)TỪapi_requestsNƠIuser_id=$1VÀcreated_at>BÂY GIỜ()-KHOẢNG'1 phút';-
- Nếu dưới giới hạn, chèn vàoCHÈN VÀOapi_requests(user_id)GIÁ TRỊ($1);-
- Dọn dẹp các yêu cầu cũ định kỳXÓATỪapi_requestsNƠIcreated_at<BÂY GIỜ()-KHOẢNG'5 phút';

đợiredis.đặt(`phiên:${sessionId}`,JSON.chuỗi hóa(sessionData),'EX',86400);

TẠOBẢNGsessions(idVĂN BẢNKHÓA CHÍNH,dataJSONBKHÔNG ĐƯỢC NULL,expires_atTHỜI GIAN CÓ MÚI GIỜKHÔNG ĐƯỢC NULL);TẠOCHỈ MỤCidx_sessions_expiresTRÊNsessions(expires_at);-
- Chèn/Cập nhậtCHÈN VÀOsessions(id,data,expires_at)GIÁ TRỊ($1,$2,BÂY GIỜ()+KHOẢNG'24 giờ')KHI XUNG ĐỘT(id)THỰC HIỆN CẬP NHẬTĐẶTdata=ĐÃ LOẠI TRỪ.data,expires_at=ĐÃ LOẠI TRỪ.expires_at;-
- ĐọcCHỌNdataTỪsessionsNƠIid=$1VÀexpires_at>BÂY GIỜ();

-
- Tìm tất cả phiên cho một người dùng cụ thểCHỌN*TỪsessionsNƠIdata->>'userId'='123';-
- Tìm phiên có vai trò cụ thểCHỌN*TỪsessionsNƠIdata->'user'->>'role'='admin';

đợidb.truy vấn('CHÈN VÀO posts ...');// 2msđợiredis.xóa('posts:latest');// 1ms (bước nhảy mạng)đợiredis.công bố('posts:new',data);// 1ms (bước nhảy mạng)// Tổng cộng:
~4ms

BẮT ĐẦU;CHÈN VÀObài_viết...;-
- 2msXÓA TỪbộ_nhớ_đệmWHEREkhóa='bài_viết:mới_nhất';-
- 0.1ms (cùng kết nối)THÔNG BÁObài_viết_mới,'...';-
- 0.1ms (cùng kết nối)CAM KẾT;-
- Tổng cộng:
~2.2ms

Redis: 100.000+ thao tác/giây (một phiên bản) PostgreSQL: 10.000-50.000 thao tác/giây 

-
- Bảng xếp hạng trong PostgreSQL (chậm hơn)SELECTuser_id,scoreFROMleaderboardORDERBYscoreDESCLIMIT10;-
- so với RedisZREVRANGEleaderboard09WITHSCORES

// Ghi vào cả haiawaitredis.set(key,value);awaitpg.query('INSERT INTO cache ...');// Đọc từ Redis (vẫn là chính)letdata=awaitredis.get(key);

// Thử PostgreSQL trướcletdata=awaitpg.query('SELECT value FROM cache WHERE key = $1',[key]);// Dự phòng bằng Redisif(!data){data=awaitredis.get(key);}

// Chỉ ghi vào PostgreSQLawaitpg.query('INSERT INTO cache ...');

# Tắt Redis# Theo dõi lỗi# Không có gì hỏng?
Thành công!

// cache.jsclassPostgresCache{constructor(pool){this.pool=pool;}asyncget(key){constresult=awaitthis.pool.query('SELECT value FROM cache WHERE key = $1 AND expires_at > NOW()',[key]);returnresult.rows[0]?.value;}asyncset(key,value,ttlSeconds=3600){awaitthis.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]);}asyncdelete(key){awaitthis.pool.query('DELETE FROM cache WHERE key = $1',[key]);}asynccleanup(){awaitthis.pool.query('DELETE FROM cache WHERE expires_at < NOW()');}}module.exports=PostgresCache;

// pubsub.jsclassPostgresPubSub{constructor(pool){this.pool=pool;this.listeners=newMap();}asyncpublish(channel,message){constpayload=JSON.stringify(message);awaitthis.pool.query('SELECT pg_notify($1, $2)',[channel,payload]);}asyncsubscribe(channel,callback){constclient=awaitthis.pool.connect();awaitclient.query(`LISTEN${channel}`);client.on('notification',(msg)=>{if(msg.channel===channel){callback(JSON.parse(msg.payload));}});this.listeners.set(channel,client);}asyncunsubscribe(channel){constclient=this.listeners.get(channel);if(client){awaitclient.query(`UNLISTEN${channel}`);client.release();this.listeners.delete(channel);}}}module.exports=PostgresPubSub;

// queue.jsclassPostgresQueue{constructor(pool){this.pool=pool;}asyncenqueue(queue,payload,scheduledAt=newDate()){awaitthis.pool.query('INSERT INTO jobs (queue, payload, scheduled_at) VALUES ($1, $2, $3)',[queue,payload,scheduledAt]);}asyncdequeue(queue){constresult=awaitthis.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]);returnresult.rows[0];}asynccomplete(jobId){awaitthis.pool.query('DELETE FROM jobs WHERE id = $1',[jobId]);}asyncfail(jobId,error){awaitthis.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');constpool=newPool({max:20,// Max connectionsidleTimeoutMillis:30000,connectionTimeoutMillis:2000,});

CREATEINDEXCONCURRENTLYidx_cache_keyONcache(key)WHEREexpires_at>NOW();CREATEINDEXCONCURRENTLYidx_jobs_pendingONjobs(queue,scheduled_at)WHEREattempts<max_attempts;

# 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 

-
- Run dailyVACUUMANALYZEcache;VACUUMANALYZEjobs;-
- Or enable autovacuum (recommended)ALTERTABLEcacheSET(autovacuum_vacuum_scale_factor=0.1);