利用Redis键空间通知实现缓存过期时的数据库同步更新

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

@Configuration
public class RedisListenerConfig {

    /**
     * 配置Redis消息监听容器
     * 负责管理Redis的订阅连接，并将接收到的消息分发给注册的监听器。
     */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(
            RedisConnectionFactory connectionFactory,
            CompanyAccountCacheExpirationListener companyAccountCacheExpirationListener) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        // 注册监听器，监听所有键的过期事件
        // __keyevent@*:expired 是Redis键空间通知的特定频道模式，
        // 用于接收所有数据库（*）中键的过期事件。
        container.addMessageListener(
            new MessageListenerAdapter(companyAccountCacheExpirationListener), 
            new PatternTopic("__keyevent@*:expired")
        );

        return container;
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Component
public class CompanyAccountCacheExpirationListener implements MessageListener {

    // 假设有一个服务层来处理数据库更新
    @Autowired
    private CompanyService companyService; 

    // Redis序列化器，用于将接收到的字节消息转换为字符串
    private final RedisSerializer<String> stringSerializer = new StringRedisSerializer();

    /**
     * 当接收到Redis消息时触发此方法
     * 消息体是过期键的名称。
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 解析过期键的名称
        String expiredKey = stringSerializer.deserialize(message.getBody());
        String channel = stringSerializer.deserialize(message.getChannel());

        System.out.println("Received expiration event from channel: " + channel + ", key: " + expiredKey);

        // 根据过期键的命名规范，提取所需信息并触发数据库更新
        // 假设缓存键的格式是 "company:account:ID"，例如 "company:account:123"
        if (expiredKey != null && expiredKey.startsWith("company:account:")) {
            try {
                String accountIdStr = expiredKey.substring("company:account:".length());
                Long accountId = Long.parseLong(accountIdStr);

                // 调用服务层方法更新数据库
                companyService.updateCompanyLastAccessedDate(accountId);
                System.out.println("Cache for company account ID " + accountId + " expired. Database updated successfully.");

            } catch (NumberFormatException e) {
                System.err.println("Error parsing account ID from expired key: " + expiredKey + ". " + e.getMessage());
            } catch (Exception e) {
                System.err.println("Error updating database for expired key " + expiredKey + ": " + e.getMessage());
                // 可以在这里添加更复杂的错误处理，例如记录日志、发送警报或重试机制
            }
        }
    }
}

// CompanyService.java (接口)
public interface CompanyService {
    void updateCompanyLastAccessedDate(Long accountId);
}

// CompanyServiceImpl.java (实现)
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;

@Service
public class CompanyServiceImpl implements CompanyService {

    // 假设有一个JPA Repository或MyBatis Mapper来与数据库交互
    // @Autowired
    // private CompanyRepository companyRepository; 

    @Override
    @Transactional // 确保数据库操作的事务性
    public void updateCompanyLastAccessedDate(Long accountId) {
        // 实际的数据库更新逻辑
        // 例如：
        // Company company = companyRepository.findById(accountId).orElse(null);
        // if (company != null) {
        //     company.setLastAccessedDate(LocalDateTime.now());
        //     companyRepository.save(company);
        // }
        System.out.println("Updating database for company account ID: " + accountId + " with current timestamp.");
        // 这里只是一个模拟，实际应调用DAO层进行数据库更新
    }
}