编程导航会员会员
登录/注册

面试刷题平台反爬虫计数扩展

技术栈与思路：

  • 本地缓存 + 异步消息（Pulsar） + 分布式缓存（Redis）
  • Redis Lua 脚本 保证点赞原子性与幂等
  • 消息驱动：计数事件先落本地，再推送到 Pulsar，后台消费者批量缓存到Redis

1.本地计数与事件推送

▼
text
复制代码
public long incrAndGetCounter(String key, int timeInterval, TimeUnit timeUnit, int expirationSec) {
    // 1. 生成分时 Redis Key
    String redisKey = generateRedisKey(key, timeInterval, timeUnit);
    // 2. 本地 Caffeine 缓存自增
    counterCache.asMap().compute(redisKey, (k,v) -> v==null?1:v+1);
    // 3. 标记为脏数据，等待批量推送
    dirtyKeys.add(redisKey);
    // 4. 本地读取优先返回
    Integer newVal = counterCache.getIfPresent(redisKey);
    if (newVal!=null) return newVal.longValue();
    // 5. 回退读取 Redis
    long redisCount = redissonClient.getAtomicLong(redisKey).get();
    counterCache.put(redisKey,(int)redisCount);
    return redisCount;
}

Caffeine 本地缓存：60s 后过期或显式清除时，会触发 RemovalListener，将 (key,value) 封装为 CounterEvent 通过 Pulsar 推送。

▼
text
复制代码
 // 本地缓存，过期时通过 Pulsar 事件发送同步命令
    private final Cache<String, Integer> counterCache = Caffeine.newBuilder()
            .expireAfterWrite(60, TimeUnit.SECONDS)
            .removalListener((key, value, cause) -> {
                if (cause.wasEvicted() || cause == RemovalCause.EXPLICIT) {
                    counterProducer.sendAsync(new CounterEvent((String) key, (Integer) value))
                            .exceptionally(ex -> {
                                log.error("发送计数事件失败 key={} value={}", key, value, ex);
                                return null;
                            });
                }
            })
            .build();

▼
text
复制代码
@Data
@NoArgsConstructor      // 生成无参构造器，供 Jackson 反序列化使用
@AllArgsConstructor
public class CounterEvent implements Serializable {
    private String key;
    private Integer value;
}

定时批量刷新：@Scheduled(fixedRate = 5000) 批量将 dirtyKeys 中的计数事件异步发送至 counter-sync-topic。

▼
text
复制代码
// 新增：每5秒批量同步计数到 Redis
    @Scheduled(fixedRate = 5000)
    public void flushCounterEvents() {
        if (dirtyKeys.isEmpty()) {
            return;
        }
        // 批量发送脏数据，并移除标记
        Set<String> keysToFlush = new HashSet<>(dirtyKeys);
        for (String k : keysToFlush) {
            Integer v = counterCache.getIfPresent(k);
            if (v != null) {
                counterProducer.sendAsync(new CounterEvent(k, v))
                    .exceptionally(ex -> {
                        log.error("定时发送计数事件失败 key={} value={}", k, v, ex);
                        return null;
                    });
            }
            dirtyKeys.remove(k);
        }
    }

▼
text
复制代码
@PostConstruct
    public void init() throws PulsarClientException {
        // 创建 Pulsar 生产者，用于发送 CounterEvent
        counterProducer = pulsarClient
                .newProducer(Schema.JSON(CounterEvent.class))
                .topic("counter-sync-topic")
                .create();
    }

2. 消费与同步到 Redis 从 Pulsar 批量拉取 CounterEvent，并执行 Lua 脚本将本地缓存中的最新计数原子同步到 Redis。

▼
text
复制代码
private void receiveMessages() {
        while (running) {
            try {
                Messages<CounterEvent> messages = consumer.batchReceive();
                if (messages != null) {
                    for (Message<CounterEvent> msg : messages) {
                        try {
                            CounterEvent evt = msg.getValue();
                            syncToRedis(evt.getKey(), evt.getValue());
                            consumer.acknowledge(msg);
                            log.info("处理计数同步消息成功");
                        } catch (Exception ex) {
                            log.error("处理计数同步消息失败", ex);
                            consumer.negativeAcknowledge(msg);
                        }
                    }
                }
            } catch (AlreadyClosedException e) {
                log.info("Pulsar Consumer 已关闭，退出接收循环");
                break;
            } catch (Exception e) {
                log.error("接收 CounterEvent 消息失败", e);
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

▼
text
复制代码
private void syncToRedis(String key, Integer value) {
        try {
            RScript script = redissonClient.getScript(IntegerCodec.INSTANCE);
            script.eval(RScript.Mode.READ_WRITE,
                    CounterManager.LUA_SCRIPT,
                    RScript.ReturnType.INTEGER,
                    Collections.singletonList(key),
                    value,
                    60);
            log.info("Synced key: {}, value: {} to Redis", key, value);
        } catch (Exception e) {
            log.error("执行 Redis 脚本失败 key={} value={}", key, value, e);
            throw e;
        }
    }

代码

▼
text
复制代码
/**
 * Pulsar 事件：用于本地计数同步
 */
@Data
@NoArgsConstructor      // 生成无参构造器，供 Jackson 反序列化使用
@AllArgsConstructor
public class CounterEvent implements Serializable {
    private String key;
    private Integer value;
}


▼
text
复制代码
/**
 * 本地计数管理器——使用 Pulsar 事件驱动将过期的计数同步到 Redis
 */
@Slf4j
@Service
public class CounterManager {

    @Resource
    private RedissonClient redissonClient;
    @Resource
    private PulsarClient   pulsarClient;

    private Producer<CounterEvent> counterProducer;

    // 跟踪需要批量刷新的 key
    private final Set<String> dirtyKeys = ConcurrentHashMap.newKeySet();

    public static final String LUA_SCRIPT =
            "local exists = redis.call('exists', KEYS[1])\n" +
                    "if exists == 1 then\n" +
                    "    redis.call('set', KEYS[1], ARGV[1])\n" +
                    "else\n" +
                    "    redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2])\n" +
                    "end\n" +
                    "return ARGV[1]";

    // 本地缓存，过期时通过 Pulsar 事件发送同步命令
    private final Cache<String, Integer> counterCache = Caffeine.newBuilder()
            .expireAfterWrite(60, TimeUnit.SECONDS)
            .removalListener((key, value, cause) -> {
                if (cause.wasEvicted() || cause == RemovalCause.EXPLICIT) {
                    counterProducer.sendAsync(new CounterEvent((String) key, (Integer) value))
                            .exceptionally(ex -> {
                                log.error("发送计数事件失败 key={} value={}", key, value, ex);
                                return null;
                            });
                }
            })
            .build();

    public long incrAndGetCounter(String key) {
        return incrAndGetCounter(key, 1, TimeUnit.MINUTES);
    }

    public long incrAndGetCounter(String key, int timeInterval, TimeUnit timeUnit) {
        int expireSec;
        switch (timeUnit) {
            case SECONDS: expireSec = timeInterval; break;
            case MINUTES: expireSec = timeInterval * 60; break;
            case HOURS:   expireSec = timeInterval * 3600; break;
            default: throw new IllegalArgumentException("Unsupported TimeUnit");
        }
        return incrAndGetCounter(key, timeInterval, timeUnit, expireSec);
    }

    public long incrAndGetCounter(String key, int timeInterval, TimeUnit timeUnit, int expirationSec) {
        if (StrUtil.isBlank(key)) return 0;
        String redisKey = generateRedisKey(key, timeInterval, timeUnit);
        // 更新本地缓存中的计数
        counterCache.asMap().compute(redisKey, (k, v) -> v == null ? 1 : v + 1);
        // 标记为脏数据，待定时刷新
        dirtyKeys.add(redisKey);
        // 优先读取本地缓存
        Integer newValue = counterCache.getIfPresent(redisKey);
        if (newValue != null) {
            return newValue.longValue();
        }
        // 如果本地缓存中没有，再回退到 Redis
        long redisCount = redissonClient.getAtomicLong(redisKey).get();
        counterCache.put(redisKey, (int) redisCount);
        return redisCount;
    }
     //生成key
    private String generateRedisKey(String key, int interval, TimeUnit tu) {
        long factor;
        switch (tu) {
            case SECONDS: factor = Instant.now().getEpochSecond() / interval; break;
            case MINUTES: factor = Instant.now().getEpochSecond() / 60 / interval; break;
            case HOURS:   factor = Instant.now().getEpochSecond() / 3600 / interval; break;
            default: throw new IllegalArgumentException("Unsupported TimeUnit");
        }
        return key + ":" + factor;
    }

    @PostConstruct
    public void init() throws PulsarClientException {
        // 创建 Pulsar 生产者，用于发送 CounterEvent
        counterProducer = pulsarClient
                .newProducer(Schema.JSON(CounterEvent.class))
                .topic("counter-sync-topic")
                .create();
    }

    @PreDestroy
    public void destroy() {
        if (counterProducer != null) {
            try {
                counterProducer.close();
            } catch (PulsarClientException e) {
                log.error("关闭 CounterEvent 生产者失败", e);
            }
        }
    }

    // 新增：每5秒批量同步计数到 Redis
    @Scheduled(fixedRate = 5000)
    public void flushCounterEvents() {
        if (dirtyKeys.isEmpty()) {
            return;
        }
        // 批量发送脏数据，并移除标记
        Set<String> keysToFlush = new HashSet<>(dirtyKeys);
        for (String k : keysToFlush) {
            Integer v = counterCache.getIfPresent(k);
            if (v != null) {
                counterProducer.sendAsync(new CounterEvent(k, v))
                    .exceptionally(ex -> {
                        log.error("定时发送计数事件失败 key={} value={}", k, v, ex);
                        return null;
                    });
            }
            dirtyKeys.remove(k);
        }
    }
}


▼
text
复制代码
@Service
@Slf4j
public class CounterSyncConsumer {

    private final PulsarClient pulsarClient;
    private final RedissonClient redissonClient;
    private Consumer<CounterEvent> consumer;
    private volatile boolean running = true;

    public CounterSyncConsumer(PulsarClient pulsarClient, RedissonClient redissonClient) {
        this.pulsarClient = pulsarClient;
        this.redissonClient = redissonClient;
    }

    @PostConstruct
    public void init() {
        try {
            consumer = pulsarClient.newConsumer(Schema.JSON(CounterEvent.class))
                    .topic("counter-sync-topic")
                    .subscriptionName("counter-sync-subscription")
                    .subscriptionType(SubscriptionType.Shared)
                    .batchReceivePolicy(BatchReceivePolicy.builder()
                            .maxNumMessages(500)
                            .timeout(5000, TimeUnit.MILLISECONDS)
                            .build())
                    .subscribe();
            new Thread(this::receiveMessages).start();
        } catch (PulsarClientException e) {
            log.error("初始化 CounterSyncConsumer 失败", e);
        }
    }

    private void receiveMessages() {
        while (running) {
            try {
                Messages<CounterEvent> messages = consumer.batchReceive();
                if (messages != null) {
                    for (Message<CounterEvent> msg : messages) {
                        try {
                            CounterEvent evt = msg.getValue();
                            syncToRedis(evt.getKey(), evt.getValue());
                            consumer.acknowledge(msg);
                            log.info("处理计数同步消息成功");
                        } catch (Exception ex) {
                            log.error("处理计数同步消息失败", ex);
                            consumer.negativeAcknowledge(msg);
                        }
                    }
                }
            } catch (AlreadyClosedException e) {
                log.info("Pulsar Consumer 已关闭，退出接收循环");
                break;
            } catch (Exception e) {
                log.error("接收 CounterEvent 消息失败", e);
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    private void syncToRedis(String key, Integer value) {
        try {
            RScript script = redissonClient.getScript(IntegerCodec.INSTANCE);
            script.eval(RScript.Mode.READ_WRITE,
                    CounterManager.LUA_SCRIPT,
                    RScript.ReturnType.INTEGER,
                    Collections.singletonList(key),
                    value,
                    60);
            log.info("Synced key: {}, value: {} to Redis", key, value);
        } catch (Exception e) {
            log.error("执行 Redis 脚本失败 key={} value={}", key, value, e);
            throw e;
        }
    }

    @PreDestroy
    public void close() {
        running = false;
        if (consumer != null) {
            try {
                consumer.close();
            } catch (PulsarClientException e) {
                log.error("关闭 CounterSyncConsumer 失败", e);
            }
        }
    }
}
面试刷题项目
0个评论
全部评论
点击登录，快来和大家讨论吧～
表情
图片
暂无评论
鱼友9639
鱼友9639
等级
内容推荐
更多
🚀 编程导航成品项目（上百个真实项目）
41
SpringBoot3 Druid 无法打开控制台解决方案
2
通过注解实现自动使用 HotKey 探测并缓存
3
在面试刷题平台中，我使用 nacos 来配置 sentinel 的 rules 的时候，经常遇到修改了配置却监听不到情况，直到我重启 nacos 这种情况才得到解决，但可能在一会又会有这种情况，想知道大家又遇到这种情况吗？又是怎么解决的？
2
智能面试刷题平台学习笔记（前端）(1)
4
作者分享
更多
分别用Sentinel和HotKey统计访问数实现反爬虫
6
面试平台基于注解的反爬虫机制实现
8
下载 APP