// 格式：業(yè)務(wù)前綴:appId:deviceId:ip:時間窗口
String key = "flow:count:app123:device456:127.0.0.1:2023080117";

public void incrementCount(String key, int delta) {
    redisTemplate.opsForValue().increment(key, delta);
}

public void incrementWithExpire(String key, int delta, long ttlSeconds) {
    redisTemplate.opsForValue().increment(key, delta);
    redisTemplate.expire(key, ttlSeconds, TimeUnit.SECONDS);
}

private static final String LUA_SCRIPT =
    "local current = redis.call('INCRBY', KEYS[1], ARGV[1])\n" +
    "if current == tonumber(ARGV[1]) then\n" +
    "   redis.call('EXPIRE', KEYS[1], ARGV[2])\n" +
    "end\n" +
    "return current";

public Long incrementAtomically(String key, int delta, long ttl) {
    return redisTemplate.execute(
        new DefaultRedisScript<>(LUA_SCRIPT, Long.class),
        Collections.singletonList(key),
        String.valueOf(delta), String.valueOf(ttl)
    );
}

public void incrementWithNX(String key, int delta, long ttl) {
    redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
        StringRedisConnection conn = (StringRedisConnection) connection;
        conn.setNX(key, "0"); // 嘗試初始化
        conn.incrBy(key, delta);
        if (conn.setNX(key + ":lock", "1")) { // 簡易鎖判斷首次
            conn.expire(key, ttl);
            conn.expire(key + ":lock", 10);
        }
        return null;
    });
}

public long calculateTtlToNextHour() {
    LocalDateTime now = LocalDateTime.now();
    LocalDateTime nextHour = now.plusHours(1).truncatedTo(ChronoUnit.HOURS);
    return ChronoUnit.SECONDS.between(now, nextHour);
}

@Component
@RequiredArgsConstructor
public class FlowCounter {
    private final RedisTemplate<String, String> redisTemplate;
    private static final String KEY_PREFIX = "flow:count:";

    @KafkaListener(topics = "${kafka.topic}")
    public void handleMessages(List<Message> messages) {
        Map<String, Integer> countMap = messages.stream()
            .collect(Collectors.toMap(
                this::buildKey,
                msg -> 1,
                Integer::sum
            ));
        
        countMap.forEach((k, v) -> 
            incrementAtomically(k, v, calculateTtlToNextHour())
        );
    }

???????    private String buildKey(Message msg) {
        return String.format("%s%s:%s:%s:%s", 
            KEY_PREFIX,
            msg.getAppId(),
            msg.getDeviceId(),
            msg.getIp(),
            LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHH"))
        );
    }
}

public long getCurrentCount(String appId, String deviceId, String ip) {
    String key = buildKey(appId, deviceId, ip);
    String val = redisTemplate.opsForValue().get(key);
    return val != null ? Long.parseLong(val) : 0L;
}

redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
    StringRedisConnection conn = (StringRedisConnection) connection;
    countMap.forEach((k, v) -> {
        conn.incrBy(k, v);
        // 可結(jié)合Lua腳本進一步優(yōu)化
    });
    return null;
});

// 在內(nèi)存中先合并相同Key的計數(shù)
Map<String, Integer> localCount = messages.stream()
    .collect(Collectors.toMap(
        this::buildKey,
        m -> 1,
        Integer::sum
    ));

@Retryable(maxAttempts = 3, backoff = @Backoff(delay = 100))
public void safeIncrement(String key, int delta) {
    // 業(yè)務(wù)邏輯
}

# TYPE redis_operations_total counter
redis_operations_total{operation="incr"} 12345
redis_operations_total{operation="expire"} 678

@Scheduled(fixedRate = 3600000)
public void checkDataConsistency() {
    // 對比DB與Redis計數(shù)差異
}