#[cfg(feature = "redis")]
use ahash::AHashMap as HashMap;
#[cfg(feature = "redis")]
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
#[cfg(feature = "redis")]
use std::sync::Arc;
#[cfg(feature = "redis")]
use std::time::{Duration, Instant};
#[cfg(feature = "redis")]
use tokio::sync::RwLock;
#[cfg(feature = "redis")]
use tracing::{debug, error, info, trace, warn};
#[cfg(feature = "redis")]
use crate::cache::l2::L2Cache;
#[cfg(feature = "redis")]
use crate::error::StorageError;
#[cfg(all(feature = "redis", feature = "fallback"))]
use crate::fallback::{ComponentType, FallbackManager, FallbackStrategy};
#[cfg(feature = "redis")]
use crate::storage::Storage;
#[cfg(feature = "redis")]
use crate::redis_storage::{RedisConfig, RedisStorage};
#[cfg(feature = "redis")]
#[derive(Debug, Clone)]
pub struct L3CacheConfig {
pub redis_config: RedisConfig,
pub l2_capacity: usize,
pub l2_cleanup_interval: Duration,
pub l2_default_ttl: Option<Duration>,
pub l3_default_ttl: Option<Duration>,
pub degrade_check_interval: Duration,
pub enable_cache_penetration_protection: bool,
pub null_value_ttl: Duration,
}
#[cfg(feature = "redis")]
impl Default for L3CacheConfig {
fn default() -> Self {
Self {
redis_config: RedisConfig::default(),
l2_capacity: 10000,
l2_cleanup_interval: Duration::from_secs(60),
l2_default_ttl: Some(Duration::from_secs(300)),
l3_default_ttl: Some(Duration::from_secs(600)),
degrade_check_interval: Duration::from_secs(5),
enable_cache_penetration_protection: true,
null_value_ttl: Duration::from_secs(60),
}
}
}
#[cfg(feature = "redis")]
impl L3CacheConfig {
pub fn new(redis_url: impl Into<String>) -> Self {
Self {
redis_config: RedisConfig::new(redis_url),
..Default::default()
}
}
pub fn l2_capacity(mut self, capacity: usize) -> Self {
self.l2_capacity = capacity;
self
}
pub fn l2_default_ttl(mut self, ttl: Duration) -> Self {
self.l2_default_ttl = Some(ttl);
self
}
pub fn l3_default_ttl(mut self, ttl: Duration) -> Self {
self.l3_default_ttl = Some(ttl);
self
}
pub fn degrade_check_interval(mut self, interval: Duration) -> Self {
self.degrade_check_interval = interval;
self
}
pub fn enable_cache_penetration_protection(mut self, enable: bool) -> Self {
self.enable_cache_penetration_protection = enable;
self
}
}
#[cfg(feature = "redis")]
#[derive(Debug, Default)]
pub struct L3CacheStats {
l1_hits: AtomicU64,
l2_hits: AtomicU64,
l3_hits: AtomicU64,
misses: AtomicU64,
degradations: AtomicU64,
recoveries: AtomicU64,
penetration_protections: AtomicU64,
}
#[cfg(feature = "redis")]
impl L3CacheStats {
pub fn l1_hits(&self) -> u64 {
self.l1_hits.load(Ordering::Relaxed)
}
pub fn l2_hits(&self) -> u64 {
self.l2_hits.load(Ordering::Relaxed)
}
pub fn l3_hits(&self) -> u64 {
self.l3_hits.load(Ordering::Relaxed)
}
pub fn misses(&self) -> u64 {
self.misses.load(Ordering::Relaxed)
}
pub fn degradations(&self) -> u64 {
self.degradations.load(Ordering::Relaxed)
}
pub fn recoveries(&self) -> u64 {
self.recoveries.load(Ordering::Relaxed)
}
pub fn penetration_protections(&self) -> u64 {
self.penetration_protections.load(Ordering::Relaxed)
}
pub fn overall_hit_rate(&self) -> f64 {
let total = self.l1_hits() + self.l2_hits() + self.l3_hits() + self.misses();
if total == 0 {
0.0
} else {
(self.l1_hits() + self.l2_hits() + self.l3_hits()) as f64 / total as f64
}
}
pub fn l1_hit_rate(&self) -> f64 {
let total = self.l1_hits() + self.l2_hits() + self.l3_hits() + self.misses();
if total == 0 {
0.0
} else {
self.l1_hits() as f64 / total as f64
}
}
pub fn l2_hit_rate(&self) -> f64 {
let total = self.l1_hits() + self.l2_hits() + self.l3_hits() + self.misses();
if total == 0 {
0.0
} else {
self.l2_hits() as f64 / total as f64
}
}
pub fn l3_hit_rate(&self) -> f64 {
let total = self.l1_hits() + self.l2_hits() + self.l3_hits() + self.misses();
if total == 0 {
0.0
} else {
self.l3_hits() as f64 / total as f64
}
}
pub fn reset(&self) {
self.l1_hits.store(0, Ordering::Relaxed);
self.l2_hits.store(0, Ordering::Relaxed);
self.l3_hits.store(0, Ordering::Relaxed);
self.misses.store(0, Ordering::Relaxed);
self.degradations.store(0, Ordering::Relaxed);
self.recoveries.store(0, Ordering::Relaxed);
self.penetration_protections.store(0, Ordering::Relaxed);
}
}
#[cfg(feature = "redis")]
pub struct L3Cache {
l2_cache: Arc<L2Cache>,
l3_storage: Arc<RwLock<Option<Arc<RedisStorage>>>>,
config: L3CacheConfig,
stats: Arc<L3CacheStats>,
degraded: Arc<AtomicBool>,
last_degraded_at: Arc<RwLock<Option<Instant>>>,
health_check_handle: tokio::task::JoinHandle<()>,
fallback_manager: Arc<FallbackManager>,
}
#[cfg(feature = "redis")]
impl L3Cache {
pub async fn new(config: L3CacheConfig) -> Result<Self, StorageError> {
info!("创建L3缓存, Redis URL: {}", config.redis_config.url);
let l2_cache = Arc::new(L2Cache::with_config(crate::cache::l2::L2CacheConfig {
capacity: config.l2_capacity,
default_ttl: config.l2_default_ttl,
cleanup_interval: config.l2_cleanup_interval,
..Default::default()
}));
let l3_storage_arc = match RedisStorage::new(config.redis_config.clone()).await {
Ok(storage) => {
info!("L3存储(Redis)创建成功");
Some(Arc::new(storage))
}
Err(e) => {
warn!("L3存储(Redis)创建失败,将使用降级模式: {}", e);
None
}
};
let l3_storage = Arc::new(RwLock::new(l3_storage_arc));
let degraded = Arc::new(AtomicBool::new(l3_storage.read().await.is_none()));
let last_degraded_at = Arc::new(RwLock::new(if degraded.load(Ordering::Relaxed) {
Some(Instant::now())
} else {
None
}));
let stats = Arc::new(L3CacheStats::default());
let fallback_manager = Arc::new(FallbackManager::new(Arc::clone(&l2_cache)));
fallback_manager
.set_strategy(
ComponentType::L3Cache,
crate::fallback::FallbackConfig::new(
ComponentType::L3Cache,
FallbackStrategy::Degraded,
)
.enabled(true)
.timeout(Duration::from_secs(5))
.max_retries(3),
)
.await;
let health_check_handle = Self::start_health_check(
Arc::clone(&l3_storage),
Arc::clone(°raded),
Arc::clone(&last_degraded_at),
Arc::clone(&stats),
Arc::clone(&fallback_manager),
config.degrade_check_interval,
);
Ok(Self {
l2_cache,
l3_storage,
config,
stats,
degraded,
last_degraded_at,
health_check_handle,
fallback_manager,
})
}
fn start_health_check(
_l3_storage: Arc<RwLock<Option<Arc<RedisStorage>>>>,
_degraded: Arc<AtomicBool>,
_last_degraded_at: Arc<RwLock<Option<Instant>>>,
_stats: Arc<L3CacheStats>,
_fallback_manager: Arc<FallbackManager>,
interval: Duration,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let mut check_interval = tokio::time::interval(interval);
loop {
check_interval.tick().await;
if _degraded.load(Ordering::Relaxed) {
trace!("尝试恢复L3存储");
if _fallback_manager.is_failed(ComponentType::L3Cache).await {
if let Some(_storage) = _l3_storage.read().await.as_ref() {
_fallback_manager
.recover_failure(ComponentType::L3Cache)
.await;
}
}
}
}
})
}
pub async fn get(&self, key: &str) -> Option<String> {
if let Some(value) = self.l2_cache.get(key).await {
self.stats.l2_hits.fetch_add(1, Ordering::Relaxed);
trace!("L2缓存命中: key={}", key);
return Some(value);
}
if !self.degraded.load(Ordering::Relaxed) {
if let Some(storage) = self.l3_storage.read().await.as_ref().cloned() {
match storage.get(key).await {
Ok(Some(value)) => {
self.stats.l3_hits.fetch_add(1, Ordering::Relaxed);
trace!("L3缓存命中: key={}", key);
self.l2_cache
.set(key, &value, self.config.l2_default_ttl)
.await;
return Some(value);
}
Ok(None) => {
self.stats.misses.fetch_add(1, Ordering::Relaxed);
trace!("L3缓存未命中: key={}", key);
}
Err(e) => {
error!("L3缓存读取失败: key={}, error={}", key, e);
if e.is_permanent() {
self.set_degraded(true).await;
self.fallback_manager
.record_failure(ComponentType::L3Cache, &e.to_string())
.await;
}
else if e.is_transient() {
warn!("L3缓存临时错误: {}", e);
self.fallback_manager
.record_failure(ComponentType::L3Cache, &e.to_string())
.await;
let failure_count = self
.fallback_manager
.get_failure_count(ComponentType::L3Cache)
.await;
if failure_count >= 3 {
self.set_degraded(true).await;
}
}
}
}
}
} else {
debug!("L3缓存已降级,跳过L3查询");
}
None
}
pub async fn set(&self, key: &str, value: &str, ttl: Option<Duration>) {
self.l2_cache.set(key, value, ttl).await;
if !self.degraded.load(Ordering::Relaxed) {
if let Some(l3_storage) = self.l3_storage.read().await.as_ref() {
let l3_ttl = ttl.or(self.config.l3_default_ttl);
let l3_ttl_seconds = l3_ttl.map(|d| d.as_secs());
if let Err(e) = l3_storage.as_ref().set(key, value, l3_ttl_seconds).await {
error!("L3缓存写入失败: key={}, error={}", key, e);
self.set_degraded(true).await;
} else {
trace!("L3缓存写入成功: key={}", key);
}
}
}
}
pub async fn delete(&self, key: &str) {
self.l2_cache.delete(key).await;
if !self.degraded.load(Ordering::Relaxed) {
if let Some(l3_storage) = self.l3_storage.read().await.as_ref() {
if let Err(e) = l3_storage.as_ref().delete(key).await {
error!("L3缓存删除失败: key={}, error={}", key, e);
} else {
trace!("L3缓存删除成功: key={}", key);
}
}
}
}
pub async fn batch_get(&self, keys: &[String]) -> HashMap<String, String> {
let mut result = HashMap::new();
for key in keys {
if let Some(value) = self.get(key).await {
result.insert(key.clone(), value);
}
}
result
}
pub async fn batch_set(&self, items: &[(String, String, Option<Duration>)]) {
for (key, value, ttl) in items {
self.set(key, value, *ttl).await;
}
}
pub async fn batch_delete(&self, keys: &[String]) {
for key in keys {
self.delete(key).await;
}
}
pub async fn get_or_load<F, Fut>(&self, key: &str, loader: F) -> Result<String, StorageError>
where
F: Fn() -> Fut,
Fut: std::future::Future<Output = Result<String, StorageError>>,
{
if let Some(value) = self.get(key).await {
return Ok(value);
}
match loader().await {
Ok(value) => {
self.set(key, &value, None).await;
Ok(value)
}
Err(e) => {
if self.config.enable_cache_penetration_protection {
self.stats
.penetration_protections
.fetch_add(1, Ordering::Relaxed);
self.set(key, "__NULL__", Some(self.config.null_value_ttl))
.await;
}
Err(e)
}
}
}
async fn set_degraded(&self, degraded: bool) {
let current = self.degraded.load(Ordering::Relaxed);
if current != degraded {
self.degraded.store(degraded, Ordering::Relaxed);
if degraded {
*self.last_degraded_at.write().await = Some(Instant::now());
self.stats.degradations.fetch_add(1, Ordering::Relaxed);
warn!("L3缓存已降级");
} else {
self.stats.recoveries.fetch_add(1, Ordering::Relaxed);
info!("L3缓存已恢复");
}
}
}
pub async fn is_degraded(&self) -> bool {
self.degraded.load(Ordering::Relaxed)
}
pub fn stats(&self) -> &L3CacheStats {
&self.stats
}
pub fn l2_cache(&self) -> &Arc<L2Cache> {
&self.l2_cache
}
pub async fn l3_storage(&self) -> Option<Arc<RedisStorage>> {
self.l3_storage.read().await.as_ref().cloned()
}
pub fn fallback_manager(&self) -> &Arc<FallbackManager> {
&self.fallback_manager
}
pub async fn clear(&self) {
self.l2_cache.clear().await;
if !self.degraded.load(Ordering::Relaxed) {
if let Some(_l3_storage) = self.l3_storage.read().await.as_ref() {
debug!("L3缓存清空(仅清空L2,Redis需要手动处理)");
}
}
}
pub async fn shutdown(&self) {
self.health_check_handle.abort();
self.l2_cache.shutdown().await;
}
}
#[cfg(feature = "redis")]
impl Drop for L3Cache {
fn drop(&mut self) {
self.health_check_handle.abort();
}
}
#[cfg(all(test, feature = "redis"))]
mod tests {
use super::*;
#[test]
fn test_l3_cache_config_default() {
let config = L3CacheConfig::default();
assert_eq!(config.l2_capacity, 10000);
assert_eq!(config.l2_default_ttl, Some(Duration::from_secs(300)));
assert_eq!(config.l3_default_ttl, Some(Duration::from_secs(600)));
assert!(config.enable_cache_penetration_protection);
}
#[test]
fn test_l3_cache_config_builder() {
let config = L3CacheConfig::new("redis://localhost:6379")
.l2_capacity(5000)
.l2_default_ttl(Duration::from_secs(600))
.l3_default_ttl(Duration::from_secs(1200))
.degrade_check_interval(Duration::from_secs(10))
.enable_cache_penetration_protection(false);
assert_eq!(config.l2_capacity, 5000);
assert_eq!(config.l2_default_ttl, Some(Duration::from_secs(600)));
assert_eq!(config.l3_default_ttl, Some(Duration::from_secs(1200)));
assert_eq!(config.degrade_check_interval, Duration::from_secs(10));
assert!(!config.enable_cache_penetration_protection);
}
#[test]
fn test_l3_cache_stats() {
let stats = L3CacheStats::default();
assert_eq!(stats.l1_hits(), 0);
assert_eq!(stats.l2_hits(), 0);
assert_eq!(stats.l3_hits(), 0);
assert_eq!(stats.misses(), 0);
assert_eq!(stats.overall_hit_rate(), 0.0);
stats.l2_hits.fetch_add(1, Ordering::Relaxed);
stats.l3_hits.fetch_add(1, Ordering::Relaxed);
stats.misses.fetch_add(1, Ordering::Relaxed);
assert_eq!(stats.l2_hits(), 1);
assert_eq!(stats.l3_hits(), 1);
assert_eq!(stats.misses(), 1);
assert_eq!(stats.overall_hit_rate(), 2.0 / 3.0);
stats.reset();
assert_eq!(stats.l2_hits(), 0);
assert_eq!(stats.l3_hits(), 0);
assert_eq!(stats.misses(), 0);
}
#[tokio::test]
async fn test_l3_cache_l2_only() {
let config = L3CacheConfig {
redis_config: RedisConfig::new("redis://invalid:6379"),
l2_capacity: 100,
l2_cleanup_interval: Duration::from_secs(60),
l2_default_ttl: Some(Duration::from_secs(300)),
l3_default_ttl: None,
degrade_check_interval: Duration::from_secs(5),
enable_cache_penetration_protection: false,
null_value_ttl: Duration::from_secs(60),
};
let cache = L3Cache::new(config).await.unwrap();
assert!(cache.is_degraded().await);
cache.set("key1", "value1", None).await;
let value = cache.get("key1").await;
assert_eq!(value, Some("value1".to_string()));
assert_eq!(cache.stats().l2_hits(), 1);
cache.shutdown().await;
}
#[tokio::test]
async fn test_l3_cache_batch_operations() {
let config = L3CacheConfig {
redis_config: RedisConfig::new("redis://invalid:6379"),
l2_capacity: 100,
l2_cleanup_interval: Duration::from_secs(60),
l2_default_ttl: Some(Duration::from_secs(300)),
l3_default_ttl: None,
degrade_check_interval: Duration::from_secs(5),
enable_cache_penetration_protection: false,
null_value_ttl: Duration::from_secs(60),
};
let cache = L3Cache::new(config).await.unwrap();
let items = vec![
("key1".to_string(), "value1".to_string(), None),
("key2".to_string(), "value2".to_string(), None),
("key3".to_string(), "value3".to_string(), None),
];
cache.batch_set(&items).await;
let keys = vec!["key1".to_string(), "key2".to_string(), "key3".to_string()];
let result = cache.batch_get(&keys).await;
assert_eq!(result.len(), 3);
let delete_keys = vec!["key1".to_string(), "key2".to_string()];
cache.batch_delete(&delete_keys).await;
let remaining = cache.batch_get(&keys).await;
assert_eq!(remaining.len(), 1);
cache.shutdown().await;
}
#[tokio::test]
async fn test_l3_cache_get_or_load() {
let config = L3CacheConfig {
redis_config: RedisConfig::new("redis://invalid:6379"),
l2_capacity: 100,
l2_cleanup_interval: Duration::from_secs(60),
l2_default_ttl: Some(Duration::from_secs(300)),
l3_default_ttl: None,
degrade_check_interval: Duration::from_secs(5),
enable_cache_penetration_protection: false,
null_value_ttl: Duration::from_secs(60),
};
let cache = L3Cache::new(config).await.unwrap();
let value1 = cache
.get_or_load("key1", || async { Ok("loaded_value".to_string()) })
.await
.unwrap();
assert_eq!(value1, "loaded_value".to_string());
let value2 = cache
.get_or_load("key1", || async { Ok("should_not_be_called".to_string()) })
.await
.unwrap();
assert_eq!(value2, "loaded_value".to_string());
assert_eq!(cache.stats().l2_hits(), 1);
cache.shutdown().await;
}
#[tokio::test]
async fn test_l3_cache_clear() {
let config = L3CacheConfig {
redis_config: RedisConfig::new("redis://invalid:6379"),
l2_capacity: 100,
l2_cleanup_interval: Duration::from_secs(60),
l2_default_ttl: Some(Duration::from_secs(300)),
l3_default_ttl: None,
degrade_check_interval: Duration::from_secs(5),
enable_cache_penetration_protection: false,
null_value_ttl: Duration::from_secs(60),
};
let cache = L3Cache::new(config).await.unwrap();
cache.set("key1", "value1", None).await;
cache.set("key2", "value2", None).await;
assert_eq!(cache.l2_cache().len().await, 2);
cache.clear().await;
assert_eq!(cache.l2_cache().len().await, 0);
cache.shutdown().await;
}
}