redis

// Nén payload bằng Snappy trước khi đẩy vào Redis
byte[] compressedPayload = Snappy.compress(jsonBytes);
redisTemplate.opsForValue().set(key, compressedPayload, Duration.ofHours(1));
 
// Quá trình đọc và giải nén diễn ra trong suốt với business logic
byte[] rawBytes = redisTemplate.opsForValue().get(key);
byte[] decompressedPayload = Snappy.uncompress(rawBytes);

redis-cli INFO memory              # tổng quan memory usage
redis-cli --bigkeys                # tìm keys lớn nhất (sample-based)
redis-cli --memkeys                # sample memory per key
redis-cli MEMORY USAGE mykey       # memory 1 key cụ thể (bytes)
redis-cli MEMORY DOCTOR            # diagnostic suggestions
redis-cli DBSIZE                   # tổng số keys
redis-cli INFO keyspace            # keys per database + expires
 
# Object encoding check
redis-cli OBJECT ENCODING mykey    # ziplist? hashtable? embstr? int?

# Check fragmentation ratio
redis-cli INFO memory | grep mem_fragmentation_ratio

public User getUser(String id) {
   String cached = redis.opsForValue().get("user:" + id);
   if (cached != null) return deserialize(cached);
 
   User user = db.findById(id);
   redis.opsForValue().set("user:" + id, serialize(user), Duration.ofMinutes(30));
   return user;
}
 
public void updateUser(String id, User user) {
   db.save(user);
   redis.delete("user:" + id);  // invalidate, không update
}

// Dùng distributed lock khi cache miss để tránh thundering herd
public User getUserSafe(String id) {
   String cached = redis.opsForValue().get("user:" + id);
   if (cached != null) return deserialize(cached);
 
   String lockKey = "lock:user:" + id;
   boolean locked = redis.opsForValue().setIfAbsent(lockKey, "1", Duration.ofSeconds(5));
 
   if (locked) {
       try {
           User user = db.findById(id);
           redis.opsForValue().set("user:" + id, serialize(user), Duration.ofMinutes(30));
           return user;
       } finally {
           redis.delete(lockKey);
       }
   } else {
       // Wait and retry
       Thread.sleep(50);
       return getUserSafe(id);
   }
}

public boolean isAllowed(String userId, int limit, int windowSeconds) {
   String key = "rate:" + userId + ":" + (System.currentTimeMillis() / (windowSeconds * 1000));
   Long count = redis.opsForValue().increment(key);
   if (count == 1) {
       redis.expire(key, Duration.ofSeconds(windowSeconds));
   }
   return count <= limit;
}

public boolean isAllowed(String userId, int limit, int windowMs) {
   String key = "rate:" + userId;
   long now = System.currentTimeMillis();
   long windowStart = now - windowMs;
 
   redis.opsForZSet().removeRangeByScore(key, 0, windowStart);
   Long count = redis.opsForZSet().zCard(key);
 
   if (count < limit) {
       redis.opsForZSet().add(key, UUID.randomUUID().toString(), now);
       redis.expire(key, Duration.ofMillis(windowMs + 1000));
       return true;
   }
   return false;
}

-- KEYS[1] = bucket key
-- ARGV[1] = max tokens, ARGV[2] = refill rate (tokens/sec), ARGV[3] = now (ms)
local key = KEYS[1]
local max_tokens = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
 
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or max_tokens
local last_refill = tonumber(bucket[2]) or now
 
-- Refill
local elapsed = (now - last_refill) / 1000
local new_tokens = math.min(max_tokens, tokens + elapsed * refill_rate)
 
if new_tokens >= 1 then
   redis.call('HSET', key, 'tokens', new_tokens - 1, 'last_refill', now)
   redis.call('EXPIRE', key, math.ceil(max_tokens / refill_rate) + 1)
   return 1  -- allowed
end
 
redis.call('HSET', key, 'tokens', new_tokens, 'last_refill', now)
return 0  -- denied

public boolean acquireLock(String resource, String owner, int ttlSeconds) {
   Boolean acquired = redis.opsForValue()
       .setIfAbsent("lock:" + resource, owner, Duration.ofSeconds(ttlSeconds));
   return Boolean.TRUE.equals(acquired);
}
 
public boolean releaseLock(String resource, String owner) {
   // Lua script đảm bảo chỉ owner mới unlock được (atomic)
   String script = """
       if redis.call('get', KEYS[1]) == ARGV[1] then
           return redis.call('del', KEYS[1])
       end
       return 0
   """;
   Long result = redis.execute(
       RedisScript.of(script, Long.class),
       List.of("lock:" + resource),
       owner
   );
   return result != null && result == 1;
}

// Dùng Redisson library
RLock lock = redisson.getLock("resource:123");
try {
   // Wait tối đa 10s để acquire, auto-release sau 30s
   if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
       // critical section
   }
} finally {
   lock.unlock();
}

// Add/update score
redis.opsForZSet().add("leaderboard:weekly", "player_a", 1500);
redis.opsForZSet().add("leaderboard:weekly", "player_b", 2300);
 
// Increment score
redis.opsForZSet().incrementScore("leaderboard:weekly", "player_a", 50);
 
// Top 10 (descending)
Set<ZSetOperations.TypedTuple<String>> top10 =
   redis.opsForZSet().reverseRangeWithScores("leaderboard:weekly", 0, 9);
 
// Rank of specific player (0-based)
Long rank = redis.opsForZSet().reverseRank("leaderboard:weekly", "player_a");
 
// Score of specific player
Double score = redis.opsForZSet().score("leaderboard:weekly", "player_a");
 
// Players ranked 50-60
Set<String> page = redis.opsForZSet().reverseRange("leaderboard:weekly", 50, 60);

// Lưu session
redis.opsForHash().putAll("session:" + sessionId, Map.of(
   "userId", "123",
   "role", "admin",
   "lastAccess", String.valueOf(System.currentTimeMillis())
));
redis.expire("session:" + sessionId, Duration.ofMinutes(30));
 
// Mỗi request: refresh TTL (sliding expiration)
redis.expire("session:" + sessionId, Duration.ofMinutes(30));
 
// Đọc session
Map<Object, Object> session = redis.opsForHash().entries("session:" + sessionId);
 
// Logout: xóa session
redis.delete("session:" + sessionId);

// Publisher
redis.convertAndSend("notifications:user:123", jsonMessage);
 
// Subscriber (Spring)
@Component
public class NotificationListener implements MessageListener {
   @Override
   public void onMessage(Message message, byte[] pattern) {
       String payload = new String(message.getBody());
       // Push to WebSocket
       websocketService.send(extractUserId(message.getChannel()), payload);
   }
}
 
// Config
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory factory) {
   RedisMessageListenerContainer container = new RedisMessageListenerContainer();
   container.setConnectionFactory(factory);
   container.addMessageListener(listener, new PatternTopic("notifications:*"));
   return container;
}

// Producer: add job
redis.opsForStream().add("jobs:email", Map.of(
   "to", "user@example.com",
   "subject", "Welcome",
   "template", "onboarding"
));
 
// Consumer group setup (chạy 1 lần)
redis.opsForStream().createGroup("jobs:email", "email-workers");
 
// Consumer: read and process
List<MapRecord<String, Object, Object>> messages = redis.opsForStream().read(
   Consumer.from("email-workers", "worker-1"),
   StreamReadOptions.empty().count(10).block(Duration.ofSeconds(5)),
   StreamOffset.create("jobs:email", ReadOffset.lastConsumed())
);
 
for (MapRecord<String, Object, Object> msg : messages) {
   try {
       processEmail(msg.getValue());
       // Acknowledge
       redis.opsForStream().acknowledge("jobs:email", "email-workers", msg.getId());
   } catch (Exception e) {
       // Message stays in PEL (Pending Entries List), sẽ được retry
       log.error("Failed to process: {}", msg.getId(), e);
   }
}
 
// Check pending (failed/timeout messages)
PendingMessages pending = redis.opsForStream().pending("jobs:email", "email-workers");

// Add locations
redis.opsForGeo().add("stores", new Point(106.6297, 10.8231), "store_hcm");
redis.opsForGeo().add("stores", new Point(105.8342, 21.0278), "store_hn");
 
// Tìm stores trong bán kính 5km từ user location
GeoResults<RedisGeoCommands.GeoLocation<String>> results = redis.opsForGeo().radius(
   "stores",
   new Circle(new Point(106.63, 10.82), new Distance(5, Metrics.KILOMETERS)),
   RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs()
       .includeDistance()
       .sortAscending()
       .limit(10)
);
 
// Khoảng cách giữa 2 points
Distance dist = redis.opsForGeo().distance("stores", "store_hcm", "store_hn", Metrics.KILOMETERS);

// Count unique visitors (approximate, ±0.81% error)
redis.opsForHyperLogLog().add("visitors:2024-01-15", "user_1", "user_2", "user_3");
redis.opsForHyperLogLog().add("visitors:2024-01-15", "user_1"); // duplicate, không tăng
 
Long count = redis.opsForHyperLogLog().size("visitors:2024-01-15"); // ~3
 
// Merge multiple days
redis.opsForHyperLogLog().union("visitors:week",
   "visitors:2024-01-15", "visitors:2024-01-16", "visitors:2024-01-17");

// Mark user active today
int userId = 12345;
redis.opsForValue().setBit("dau:2024-01-15", userId, true);
 
// Check if user was active
Boolean active = redis.opsForValue().getBit("dau:2024-01-15", userId);
 
// Count total active users today
// (dùng BITCOUNT command)
Long activeCount = redis.execute((RedisCallback<Long>) conn ->
   conn.bitCount("dau:2024-01-15".getBytes()));
 
// Users active cả 3 ngày (AND operation)
redis.execute((RedisCallback<Long>) conn ->
   conn.bitOp(BitOperation.AND, "dau:3days".getBytes(),
       "dau:2024-01-15".getBytes(),
       "dau:2024-01-16".getBytes(),
       "dau:2024-01-17".getBytes()));

// Atomic increment
Long views = redis.opsForValue().increment("article:123:views");
 
// Increment by N
redis.opsForValue().increment("stats:downloads", 5);
 
// Hash-based counters (multiple metrics per entity)
redis.opsForHash().increment("article:123", "views", 1);
redis.opsForHash().increment("article:123", "likes", 1);
redis.opsForHash().increment("article:123", "shares", 1);
 
// Batch flush to DB periodically (không write DB mỗi increment)
// Scheduled job mỗi 5 phút: read counters → batch update DB → reset counters

// Track failures cho external service
public boolean isCircuitOpen(String service) {
   String key = "circuit:" + service;
   String state = redis.opsForValue().get(key + ":state");
 
   if ("open".equals(state)) {
       // Check if cooldown passed
       Long ttl = redis.getExpire(key + ":state");
       return ttl != null && ttl > 0;
   }
   return false;
}
 
public void recordFailure(String service, int threshold, int cooldownSeconds) {
   String key = "circuit:" + service + ":failures";
   Long failures = redis.opsForValue().increment(key);
   redis.expire(key, Duration.ofMinutes(1)); // reset window
 
   if (failures >= threshold) {
       redis.opsForValue().set("circuit:" + service + ":state", "open",
           Duration.ofSeconds(cooldownSeconds));
   }
}