Bối cảnh & Tại sao chọn Redis Streams
Sau khi làm việc với MySQL, PostgreSQL và MongoDB qua nhiều dự án khác nhau, tôi cứ gặp đi gặp lại cùng một vấn đề: database rất giỏi lưu trữ trạng thái, nhưng lại rất tệ trong việc truyền sự kiện giữa các service. Khi cần xử lý đơn hàng real-time, tracking hoạt động người dùng hay thông báo cross-service, chúng tôi thường gắn thêm polling loop hoặc task queue đơn giản — và lúc nào cũng có cảm giác đang dùng sai công cụ.
Redis Streams, ra mắt trong Redis 5.0, đã thay đổi điều đó. Pub/sub cơ bản sẽ mất message nếu subscriber offline. List đơn giản thiếu ngữ nghĩa consumer group. Streams cho bạn một log persistent, append-only với acknowledgment và delivery tracking tích hợp sẵn — không còn những đánh đổi đó nữa.
Đây là những trường hợp Redis Streams phù hợp hơn Kafka hay RabbitMQ:
- Bạn đang chạy Redis sẵn và muốn tránh thêm một dependency infrastructure
- Lượng message ở mức vừa phải (hàng triệu mỗi ngày, không phải hàng tỷ)
- Bạn cần setup nhanh với chi phí vận hành thấp
- Bạn muốn ngữ nghĩa consumer group mà không phải đối mặt với độ phức tạp quản lý partition của Kafka
Cài đặt
Cài Redis 7+ với hỗ trợ Streams
Redis Streams đã ổn định từ Redis 5.0, nhưng Redis 7 mang lại những cải tiến quan trọng cho XAUTOCLAIM và quản lý consumer group. Tôi khuyến nghị dùng Redis 7.x cho mọi setup mới.
Trên Ubuntu/Debian:
# Thêm repository Redis chính thức
curl -fsSL https://packages.redis.io/gpg | sudo gpg --dearmor -o /usr/share/keyrings/redis-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/redis-archive-keyring.gpg] https://packages.redis.io/deb $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/redis.list
sudo apt update
sudo apt install -y redis-server
# Kiểm tra phiên bản — phải là 7.x
redis-server --versionTrên CentOS/RHEL:
sudo dnf install -y epel-release
sudo dnf install -y redis
sudo systemctl enable --now redisKiểm tra các lệnh Streams khả dụng:
redis-cli XADD test_stream '*' key value
# Nên trả về dạng: "1728390000000-0"
redis-cli DEL test_streamThiết lập Python Client
pip install "redis>=5.0.0"Cấu hình
Thêm message với XADD
XADD thêm một message vào stream. Mỗi message nhận một ID tự động — timestamp millisecond cộng với số thứ tự — trừ khi bạn chỉ định ID cụ thể.
# XADD cơ bản: tự động tạo ID
XADD orders '*' user_id 42 product_id 1001 quantity 2 status pending
# Dùng MAXLEN để giới hạn kích thước stream (trim xấp xỉ với ~)
XADD orders MAXLEN ~ 10000 '*' user_id 43 product_id 1002 quantity 1 status pendingTôi luôn thêm MAXLEN ~ 10000 trong môi trường production. Nếu không, stream sẽ tăng trưởng vô tận và Redis cuối cùng sẽ hết bộ nhớ. Dấu ~ báo Redis trim một cách lười biếng — chỉ khi có ý nghĩa về mặt cấu trúc — điều này rẻ hơn nhiều so với trim chính xác trên mỗi lần ghi.
Trong Python:
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# Thêm một sự kiện đơn hàng
msg_id = r.xadd(
'orders',
{
'user_id': '42',
'product_id': '1001',
'quantity': '2',
'status': 'pending'
},
maxlen=10000,
approximate=True
)
print(f"Đã thêm message: {msg_id}")Tạo Consumer Groups
Consumer group cho phép nhiều worker đọc từ cùng một stream mà không bị trùng lặp. Mỗi message chỉ đến đúng một consumer trong group — không broadcast đến tất cả.
# Tạo group bắt đầu từ đầu stream
# 0 = đọc tất cả message hiện có
# $ = chỉ message mới từ bây giờ
# MKSTREAM tạo stream nếu chưa tồn tại
XGROUP CREATE orders order-processors 0 MKSTREAM# Tạo consumer group, bỏ qua nếu đã tồn tại
try:
r.xgroup_create('orders', 'order-processors', id='0', mkstream=True)
print("Đã tạo consumer group")
except redis.exceptions.ResponseError as e:
if 'BUSYGROUP' in str(e):
print("Group đã tồn tại, bỏ qua")
else:
raiseNhận message với XREADGROUP
XREADGROUP đọc message mới và chuyển chúng vào Pending Entry List (PEL) của consumer. Message nằm trong PEL cho đến khi bạn gọi XACK — đó là đảm bảo at-least-once delivery.
# Đọc tối đa 10 message mới cho consumer "worker-1"
# Dấu ">" có nghĩa là "cho tôi các message chưa được giao"
XREADGROUP GROUP order-processors worker-1 COUNT 10 BLOCK 2000 STREAMS orders >Thay > bằng 0 để đọc lại các message pending (chưa được xác nhận) — rất hữu ích khi khôi phục sau khi crash.
import time
import socket
consumer_name = f"worker-{socket.gethostname()}"
def process_orders():
while True:
# Đọc message mới, block tối đa 2 giây nếu stream trống
messages = r.xreadgroup(
groupname='order-processors',
consumername=consumer_name,
streams={'orders': '>'},
count=10,
block=2000
)
if not messages:
continue
for stream_name, stream_messages in messages:
for msg_id, fields in stream_messages:
try:
process_single_order(msg_id, fields)
# Acknowledge: xóa khỏi PEL
r.xack('orders', 'order-processors', msg_id)
except Exception as e:
print(f"Xử lý thất bại {msg_id}: {e}")
# Không ack — message ở lại PEL để retry
def process_single_order(msg_id, fields):
print(f"Đang xử lý đơn hàng {msg_id}: {fields}")
time.sleep(0.1) # Giả lập xử lýDead Letter Queue với XPENDING và XCLAIM
Hầu hết hướng dẫn chỉ đề cập XREADGROUP rồi dừng lại. Pattern reaper mới là thứ thực sự làm cho Streams đáng tin cậy trong production. Khi một consumer crash giữa chừng xử lý, các message của nó bị kẹt trong PEL cho đến khi có gì đó đến claim chúng. Bạn cần một reaper process chuyên dụng: quét các PEL entry idle, retry chúng, và sau N lần thất bại, chuyển vào dead letter queue (DLQ).
# Kiểm tra pending messages (tất cả consumer, tất cả ID)
XPENDING orders order-processors - + 100DLQ_STREAM = 'orders:dead-letter'
MAX_RETRIES = 3
CLAIM_IDLE_MS = 5 * 60 * 1000 # 5 phút
def reaper_loop():
"""Thu hồi message bị treo và chuyển các lỗi dai dẳng vào DLQ."""
while True:
# Tìm message idle lâu hơn CLAIM_IDLE_MS
pending = r.xpending_range(
'orders',
'order-processors',
min='-',
max='+',
count=50,
idle=CLAIM_IDLE_MS
)
for entry in pending:
msg_id = entry['message_id']
delivery_count = entry['times_delivered']
if delivery_count >= MAX_RETRIES:
# Lấy nội dung message trước khi xóa khỏi PEL
msgs = r.xrange('orders', min=msg_id, max=msg_id, count=1)
if msgs:
_, fields = msgs[0]
fields['original_id'] = msg_id
fields['failure_reason'] = f'vượt quá {MAX_RETRIES} lần retry'
# Ghi vào DLQ
r.xadd(DLQ_STREAM, fields, maxlen=5000, approximate=True)
# Ack để xóa khỏi PEL — dừng retry
r.xack('orders', 'order-processors', msg_id)
print(f"Đã chuyển {msg_id} vào DLQ sau {delivery_count} lần thử")
else:
# Giao lại cho reaper consumer này để retry
r.xclaim(
'orders',
'order-processors',
consumer_name,
min_idle_time=CLAIM_IDLE_MS,
message_ids=[msg_id]
)
print(f"Đã thu hồi {msg_id} để retry (lần thử {delivery_count + 1})")
time.sleep(30) # Chạy mỗi 30 giâyMột điều tôi học được theo cách khó: luôn đặt MAXLEN cho DLQ của bạn nữa. Một bug liên tục sinh ra lỗi sẽ âm thầm lấp đầy bộ nhớ nếu DLQ tăng trưởng không kiểm soát.
Nếu bạn đang dùng Redis 6.2+, XAUTOCLAIM kết hợp scan và claim thành một lệnh atomic duy nhất — gọn hơn pattern hai bước XPENDING + XCLAIM ở trên. Đáng chuyển sang dùng khi setup mới.
Kiểm tra & Giám sát
Kiểm tra sức khỏe Stream và Group
# Độ dài stream
XLEN orders
# Thông tin đầy đủ stream (entry đầu/cuối, số group, độ dài)
XINFO STREAM orders
# Chi tiết consumer group (số pending, ID giao gần nhất)
XINFO GROUPS orders
# Kích thước PEL theo từng consumer
XINFO CONSUMERS orders order-processorsScript kiểm tra sức khỏe
Theo dõi số pending theo từng consumer. Nếu nó liên tục tăng, consumer của bạn đang quá chậm hoặc bị kẹt ngầm — cả hai vấn đề bạn muốn phát hiện sớm, không phải sau khi có 50.000 message chưa xử lý tồn đọng. Tích hợp số liệu này vào hệ thống observability giúp bạn tương quan độ trễ xử lý với các vấn đề ở tầng ứng dụng.
def stream_health_check():
groups = r.xinfo_groups('orders')
for group in groups:
name = group['name']
pending = group['pending']
last_delivered = group['last-delivered-id']
print(f"Group: {name}")
print(f" Message đang chờ: {pending}")
print(f" ID giao gần nhất: {last_delivered}")
if pending > 1000:
print(f" CẢNH BÁO: Số pending cao — consumer có thể đang bị chậm")
# Kiểm tra DLQ
dlq_len = r.xlen(DLQ_STREAM) if r.exists(DLQ_STREAM) else 0
print(f"\nKích thước Dead Letter Queue: {dlq_len}")
if dlq_len > 0:
print(" Cần xử lý — xem lại các message lỗi trong DLQ")
stream_health_check()Bật Redis Persistence
Redis mặc định lưu dữ liệu trong RAM. Với bất kỳ hàng đợi nào mà việc mất message là vấn đề, hãy bật persistence trong /etc/redis/redis.conf:
# AOF: ghi lại mọi lệnh ghi (độ bền tốt nhất)
appendonly yes
appendfsync everysec
# Hoặc dùng RDB snapshot (nhẹ hơn, có nguy cơ mất một ít dữ liệu)
save 900 1
save 300 10
save 60 10000Sau đó khởi động lại Redis: sudo systemctl restart redis
Mẹo từ kinh nghiệm vận hành thực tế
- Luôn đặt MAXLEN với
~— trim chính xác trên mỗi XADD rất tốn kém; trim xấp xỉ có hành vi gần như giống hệt - Dùng tên consumer dựa trên hostname — giúp dễ dàng truy vết xem pod nào đang sở hữu PEL entry nào bị treo
- Chạy reaper như một process riêng biệt — trộn vào consumer loop chính sẽ thêm độ phức tạp và các failure mode không cần thiết
- Đặt idle timeout bằng 2–3 lần thời gian xử lý tối đa dự kiến — quá ngắn gây reclaim nhầm; quá dài trì hoãn phục hồi sau crash
- Theo dõi kích thước PEL theo từng consumer — xuất dữ liệu
XINFO CONSUMERSsang Prometheus hoặc Grafana nếu bạn đang chạy ở quy mô lớn - Luôn đặt MAXLEN cho DLQ của bạn — một vòng lặp lỗi không kiểm soát sẽ lấp đầy bộ nhớ nếu không giới hạn
Redis Streams sẽ không thay thế Kafka ở mức hàng triệu message mỗi giây. Nhưng với hầu hết các team đang chạy Redis sẵn, nó giúp loại bỏ cả một lớp infrastructure. Model consumer group rất vững chắc. Các lệnh rất rõ ràng. Cấu hình thêm reaper là bạn đã có đảm bảo at-least-once delivery thực sự — không cần broker cluster riêng biệt.
