Redis #
Redis is an in-memory data store that can serve as a cache, message broker, session store, rate limiter, and distributed lock all at once. Its strengths lie in extreme speed (hundreds of thousands of operations per second) and rich data structures far beyond plain key-value — strings, hashes, lists, sets, sorted sets, streams, and more. In Rust, the redis crate is the most widely used Redis driver, and deadpool-redis adds an async connection pool on top of it. This article covers all Redis data types, common caching patterns, distributed locks, rate limiting, and lightweight Pub/Sub as an alternative to a message broker.
Installation #
[dependencies]
redis = { version = "0.25", features = ["tokio-comp", "connection-manager"] }
deadpool-redis = "0.15"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
Connections and Connection Pools #
use deadpool_redis::{Config, Pool, Runtime};
use redis::AsyncCommands;
fn buat_pool(url: &str) -> Pool {
let cfg = Config::from_url(url);
cfg.create_pool(Some(Runtime::Tokio1))
.expect("Failed to create Redis pool")
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let pool = buat_pool("redis://localhost:6379");
// Get a connection from the pool
let mut conn = pool.get().await?;
// Ping to verify
let pong: String = redis::cmd("PING").query_async(&mut conn).await?;
println!("Redis: {}", pong); // PONG
Ok(())
}
Strings — The Basic Data Type #
use redis::AsyncCommands;
async fn contoh_string(conn: &mut deadpool_redis::Connection)
-> Result<(), Box<dyn std::error::Error>>
{
// Basic SET and GET
conn.set("kunci", "nilai").await?;
let nilai: String = conn.get("kunci").await?;
println!("GET: {}", nilai);
// SET with TTL (expire in seconds)
conn.set_ex("sesi:user-42", "data-sesi", 3600).await?; // 1 hour
// SET only if absent (SET NX) — for a simple distributed lock
let berhasil: bool = conn.set_nx("lock:resource-1", "owner-1").await?;
println!("Lock acquired: {}", berhasil);
// INCR / DECR — atomic without race conditions
conn.set("counter:halaman", 0i64).await?;
let baru: i64 = conn.incr("counter:halaman", 1).await?;
println!("Counter: {}", baru);
// MSET / MGET — batch operations
conn.set_multiple(&[
("user:1:nama", "Budi"),
("user:2:nama", "Sari"),
("user:3:nama", "Joko"),
]).await?;
let nama_list: Vec<Option<String>> = conn.get(vec![
"user:1:nama", "user:2:nama", "user:3:nama", "user:99:nama"
]).await?;
println!("{:?}", nama_list); // [Some("Budi"), Some("Sari"), Some("Joko"), None]
// TTL — check remaining key lifetime
let ttl: i64 = conn.ttl("sesi:user-42").await?;
println!("Session TTL: {} seconds", ttl);
// EXPIRE — set TTL on an existing key
conn.expire("kunci", 60).await?;
Ok(())
}
Hashes — Structured Objects #
Hashes are suitable for storing objects with many fields — more efficient than storing a JSON string because individual fields can be updated:
use redis::AsyncCommands;
async fn contoh_hash(conn: &mut deadpool_redis::Connection)
-> Result<(), Box<dyn std::error::Error>>
{
let kunci = "pengguna:1001";
// HSET — set one or many fields
conn.hset_multiple(kunci, &[
("nama", "Budi Santoso"),
("email", "[email protected]"),
("peran", "admin"),
("aktif", "true"),
]).await?;
// HGET — get one field
let nama: String = conn.hget(kunci, "nama").await?;
println!("Name: {}", nama);
// HMGET — get many fields at once
let data: Vec<Option<String>> = conn.hget(kunci, vec!["nama", "email", "peran"]).await?;
println!("{:?}", data);
// HGETALL — get all fields and values
let semua: std::collections::HashMap<String, String> = conn.hgetall(kunci).await?;
println!("All fields: {:?}", semua);
// HINCRBY — atomically increment a numeric field
conn.hset(kunci, "skor", "0").await?;
let skor: i64 = conn.hincr(kunci, "skor", 10).await?;
println!("Score: {}", skor);
// HEXISTS — check field existence
let ada: bool = conn.hexists(kunci, "email").await?;
println!("Email exists: {}", ada);
// HDEL — delete a field
conn.hdel(kunci, "aktif").await?;
// Set TTL on the hash
conn.expire(kunci, 3600).await?;
Ok(())
}
Lists — Queues and Stacks #
use redis::AsyncCommands;
async fn contoh_list(conn: &mut deadpool_redis::Connection)
-> Result<(), Box<dyn std::error::Error>>
{
let kunci = "antrian:email";
// RPUSH — add to the right (for FIFO queues)
conn.rpush(kunci, "email-1").await?;
conn.rpush(kunci, "email-2").await?;
conn.rpush(kunci, "email-3").await?;
// LPUSH — add to the left (for LIFO stacks)
conn.lpush("stack:undo", "aksi-1").await?;
conn.lpush("stack:undo", "aksi-2").await?;
// LLEN — list length
let panjang: i64 = conn.llen(kunci).await?;
println!("Queue length: {}", panjang);
// LPOP — take from the left (FIFO: take the first in)
let item: Option<String> = conn.lpop(kunci, None).await?;
println!("Popped: {:?}", item);
// RPOP — take from the right (LIFO/stack)
let top: Option<String> = conn.rpop("stack:undo", None).await?;
println!("Stack pop: {:?}", top);
// LRANGE — read all elements without removing
let semua: Vec<String> = conn.lrange(kunci, 0, -1).await?;
println!("All: {:?}", semua);
// BLPOP — blocking pop (wait until an element arrives)
// Useful for worker queues
// let (_, item): (String, String) = conn.blpop(kunci, 5.0).await?;
Ok(())
}
Sets and Sorted Sets #
use redis::AsyncCommands;
async fn contoh_set(conn: &mut deadpool_redis::Connection)
-> Result<(), Box<dyn std::error::Error>>
{
// Set: a collection of unique elements without duplicates
conn.sadd("tag:artikel-1", vec!["rust", "backend", "performance"]).await?;
conn.sadd("tag:artikel-2", vec!["rust", "webdev", "frontend"]).await?;
// SMEMBERS — all members
let tag: std::collections::HashSet<String> = conn.smembers("tag:artikel-1").await?;
println!("Tags: {:?}", tag);
// SISMEMBER — membership check
let ada: bool = conn.sismember("tag:artikel-1", "rust").await?;
println!("Has 'rust': {}", ada);
// Set operations
let irisan: Vec<String> = conn.sinter(vec!["tag:artikel-1", "tag:artikel-2"]).await?;
println!("Intersection: {:?}", irisan); // ["rust"]
let gabungan: Vec<String> = conn.sunion(vec!["tag:artikel-1", "tag:artikel-2"]).await?;
println!("Union: {:?}", gabungan);
// Sorted Set: a collection of elements with scores (floats)
// Great for leaderboards, rankings, sliding-window rate limiting
conn.zadd("leaderboard", "Alice", 1500.0_f64).await?;
conn.zadd("leaderboard", "Bob", 1200.0_f64).await?;
conn.zadd("leaderboard", "Charlie", 1800.0_f64).await?;
conn.zadd("leaderboard", "Diana", 950.0_f64).await?;
// ZRANGE — get by rank (ascending)
let top: Vec<String> = conn.zrange("leaderboard", 0, 2).await?;
println!("3 lowest: {:?}", top);
// ZREVRANGE — get by rank (descending)
let top3: Vec<String> = conn.zrevrange("leaderboard", 0, 2).await?;
println!("Top 3: {:?}", top3); // ["Charlie", "Alice", "Bob"]
// ZRANK — element position (0-based, ascending)
let rank: Option<i64> = conn.zrank("leaderboard", "Alice").await?;
println!("Alice's rank: {:?}", rank);
// ZSCORE — element score
let skor: Option<f64> = conn.zscore("leaderboard", "Charlie").await?;
println!("Charlie's score: {:?}", skor);
// ZINCRBY — atomically add to a score
let skor_baru: f64 = conn.zincr("leaderboard", "Bob", 300.0_f64).await?;
println!("Bob's new score: {}", skor_baru);
Ok(())
}
Pipelines — Batch Commands #
Pipelines reduce network round-trips by sending many commands at once:
use redis::{AsyncCommands, Pipeline};
async fn contoh_pipeline(
pool: &deadpool_redis::Pool,
) -> Result<(), Box<dyn std::error::Error>> {
let mut conn = pool.get().await?;
// ANTI-PATTERN: many separate round-trips
// for i in 0..100 {
// conn.set(format!("key:{}", i), i).await?; // 100 round-trips!
// }
// CORRECT: a pipeline — one round-trip for many commands
let mut pipe = redis::pipe();
for i in 0..100 {
pipe.set(format!("key:{}", i), i).ignore();
}
// Add several GETs at once
pipe.get("key:0").get("key:50").get("key:99");
let hasil: (String, String, String) = pipe
.query_async(&mut conn)
.await?;
println!("key:0={}, key:50={}, key:99={}", hasil.0, hasil.1, hasil.2);
Ok(())
}
Common Caching Patterns #
Cache-Aside (Lazy Loading) #
use redis::AsyncCommands;
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize, Clone)]
struct Produk {
id: u64,
nama: String,
harga: f64,
}
// Simulate a slow database query
async fn ambil_dari_db(id: u64) -> Produk {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
Produk { id, nama: format!("Produk {}", id), harga: id as f64 * 1000.0 }
}
async fn ambil_produk(
pool: &deadpool_redis::Pool,
id: u64,
) -> Result<Produk, Box<dyn std::error::Error>> {
let kunci = format!("produk:{}", id);
let mut conn = pool.get().await?;
// 1. Check the cache
let cached: Option<String> = conn.get(&kunci).await?;
if let Some(data) = cached {
println!("Cache HIT for product {}", id);
let produk: Produk = serde_json::from_str(&data)?;
return Ok(produk);
}
// 2. Cache MISS — get from the database
println!("Cache MISS for product {}", id);
let produk = ambil_dari_db(id).await;
// 3. Store in the cache with a TTL
let json = serde_json::to_string(&produk)?;
conn.set_ex(&kunci, &json, 300).await?; // 5 minute TTL
Ok(produk)
}
// Cache invalidation when data changes
async fn perbarui_produk(
pool: &deadpool_redis::Pool,
id: u64,
harga_baru: f64,
) -> Result<(), Box<dyn std::error::Error>> {
// Update the database...
// Delete the cache — it will be refreshed on the next request
let mut conn = pool.get().await?;
conn.del(format!("produk:{}", id)).await?;
println!("Cache for product {} invalidated", id);
Ok(())
}
Distributed Locks #
use redis::AsyncCommands;
use uuid::Uuid;
struct RedisLock {
pool: deadpool_redis::Pool,
kunci: String,
token: String,
}
impl RedisLock {
async fn acquire(
pool: &deadpool_redis::Pool,
resource: &str,
ttl_ms: u64,
) -> Option<RedisLock> {
let kunci = format!("lock:{}", resource);
let token = Uuid::new_v4().to_string();
let mut conn = pool.get().await.ok()?;
// SET NX PX — atomic: set if absent, with TTL in milliseconds
let result: Option<String> = redis::cmd("SET")
.arg(&kunci)
.arg(&token)
.arg("NX")
.arg("PX")
.arg(ttl_ms)
.query_async(&mut conn)
.await
.ok()?;
if result.is_some() {
Some(RedisLock {
pool: pool.clone(),
kunci,
token,
})
} else {
None // The lock is already held by someone else
}
}
async fn release(&self) -> bool {
// Lua script for atomic release — only delete if the token matches
// Prevents deleting someone else's lock if our TTL already expired
let script = redis::Script::new(r"
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('DEL', KEYS[1])
else
return 0
end
");
let mut conn = match self.pool.get().await {
Ok(c) => c,
Err(_) => return false,
};
let result: i64 = script
.key(&self.kunci)
.arg(&self.token)
.invoke_async(&mut conn)
.await
.unwrap_or(0);
result == 1
}
}
async fn dengan_lock<F, Fut, T>(
pool: &deadpool_redis::Pool,
resource: &str,
ttl_ms: u64,
f: F,
) -> Result<T, String>
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = T>,
{
let lock = RedisLock::acquire(pool, resource, ttl_ms)
.await
.ok_or_else(|| format!("Failed to acquire lock for '{}'", resource))?;
let hasil = f().await;
lock.release().await;
Ok(hasil)
}
Rate Limiting with a Sliding Window #
use redis::AsyncCommands;
async fn cek_rate_limit(
pool: &deadpool_redis::Pool,
user_id: u64,
maks_request: u64,
window_detik: u64,
) -> Result<bool, Box<dyn std::error::Error>> {
let kunci = format!("rate:{}:{}", user_id, chrono::Utc::now().timestamp() / window_detik as i64);
let mut conn = pool.get().await?;
// Atomic INCR — count requests in the current window
let jumlah: u64 = conn.incr(&kunci, 1u64).await?;
// Set TTL only on the first request in the window
if jumlah == 1 {
conn.expire(&kunci, window_detik as i64 * 2).await?;
}
let diizinkan = jumlah <= maks_request;
if !diizinkan {
println!("Rate limit exceeded for user {}: {}/{}", user_id, jumlah, maks_request);
}
Ok(diizinkan)
}
// Rate limiting with a Sorted Set (more accurate sliding window)
async fn rate_limit_sliding(
pool: &deadpool_redis::Pool,
identifier: &str,
maks: u64,
window_ms: u64,
) -> Result<bool, Box<dyn std::error::Error>> {
let kunci = format!("sliding:{}", identifier);
let sekarang = chrono::Utc::now().timestamp_millis() as f64;
let window_mulai = sekarang - window_ms as f64;
let mut conn = pool.get().await?;
let mut pipe = redis::pipe();
pipe.atomic()
// Remove entries that have left the window
.cmd("ZREMRANGEBYSCORE").arg(&kunci).arg(0).arg(window_mulai).ignore()
// Add the current request
.cmd("ZADD").arg(&kunci).arg(sekarang).arg(sekarang).ignore()
// Count requests in the window
.cmd("ZCARD").arg(&kunci)
// Set TTL
.cmd("EXPIRE").arg(&kunci).arg((window_ms / 1000 + 1) as i64).ignore();
let (jumlah,): (u64,) = pipe.query_async(&mut conn).await?;
Ok(jumlah <= maks)
}
Pub/Sub — Lightweight Messaging #
Redis Pub/Sub is suitable for real-time notifications within a single datacenter — simpler than Kafka/RabbitMQ but without persistence:
use redis::AsyncCommands;
async fn redis_pubsub(
pool: &deadpool_redis::Pool,
) -> Result<(), Box<dyn std::error::Error>> {
// Publisher — use a separate connection
let publisher_pool = pool.clone();
tokio::spawn(async move {
let mut conn = publisher_pool.get().await.unwrap();
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
for i in 1..=5 {
conn.publish("channel:notifikasi",
serde_json::json!({"id": i, "pesan": format!("Notifikasi {}", i)}).to_string()
).await.unwrap();
println!("Notification {} sent", i);
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
});
// Subscriber — needs a dedicated connection (can't run other commands)
let mut conn = pool.get().await?;
let mut pubsub = conn.as_pubsub();
pubsub.subscribe("channel:notifikasi").await?;
println!("Subscriber active, waiting for notifications...");
for _ in 0..5 {
let pesan = pubsub.on_message().next_message().await?;
let payload: String = pesan.get_payload()?;
println!("Received: {}", payload);
}
Ok(())
}
Summary #
- Always use a connection pool —
deadpool-redismanages an async connection pool. One connection per request is very expensive; a pool shares connections between requests.- Key naming convention with
:— use namespaces likepengguna:1001,sesi:abc123,rate:user-42for clear organization that can be managed withSCAN.set_exfor caching with a TTL — always set a TTL on cache keys so memory doesn’t fill up and data doesn’t stay stale forever.- Hashes are more efficient than JSON strings for objects — you can update one field without re-serializing/deserializing the whole object. Good for data that’s frequently partially updated.
- Pipelines for batch operations — sending many commands in one round-trip is far more efficient than separate commands. Saves up to 10x latency for bulk operations.
- Distributed locks with
SET NX PX— atomic and auto-expiring. Always use a unique token (UUID) and a Lua script for release so you don’t delete someone else’s lock.- Sorted Sets for leaderboards and rate limiting —
ZADD+ZRANK+ZINCRBYfor real-time leaderboards;ZREMRANGEBYSCORE+ZCARDfor sliding-window rate limiting.- Redis Pub/Sub for in-process notifications — simpler than Kafka/RabbitMQ, but without persistence. Good for cache invalidation between instances or WebSocket notifications.
- Lua scripts for complex atomic operations — several Redis commands that need to execute atomically can be wrapped in a Lua script without needing a full transaction.