Hazelcast EventQueue 重载更新缓存区域 default-update-timestamps-region答案

【问题标题】：Hazelcast EventQueue overloaded updating cache region default-update-timestamps-regionHazelcast EventQueue 重载更新缓存区域 default-update-timestamps-region
【发布时间】：2021-05-16 12:34:28
【问题描述】：

我最近添加了一个新的工作线程，它将队列排入不同的表，然后退出。运行时，我看到很多后续日志消息；

2021-05-16 11:25:19.496  WARN 18 --- [Thread-1] com.hazelcast.spi.EventService           : [127.0.0.1]:5701 [dev] [3.12.6] EventQueue overloaded! TopicEvent{name='default-update-timestamps-region', publishTime=1621164319495, publisherAddress=[172.18.0.4]:5701} failed to publish to hz:impl:topicService:default-update-timestamps-region

我正在读取/写入的实体表都没有被缓存，所以我想知道为什么这个线程上的缓存会被刷新，更不用说它是如何打破这个 EventQueue 的限制的了？

我没有更改配置的默认值（使用 Hazelcast 3.12.6），所以我很困惑这怎么能这么快地补充这个缓存？

我的新服务的粗略伪代码见下文；

private void processForever() {
    threadRef = Thread.currentThread();
    synchronized (syncObject) {
        //notify init that we're good to continue
        syncObject.notifyAll();
    }
    while (threadRef == Thread.currentThread()) {
        boolean foundWork = false;
        try {
            foundWork = process();
        } catch (Exception e) {
            log.debug("stack", e);
        }

        long sleep = foundWork ? 1000 : 60000;
        try {
            Thread.sleep(sleep);
        } catch (InterruptedException e) {

        }
    }
}

private boolean process() {
    try {
        // N.B this attempts to grab a shared lock on the current tenant and skips of already taken
        return dataGrid.runExclusiveForCurrentTenantOrSkip(LockName.PROCESS, this::processInternal).orElse(true);
    } catch (Exception ex) {
        log.error("error", ex);
        return true;
    }
}

private boolean processInternal() {
    Long maxSid = sourceQueueRepo.findMaxSid();
    if (maxSid == null) {
        return false;
    }

    Set<Worker> agents = workerRepo.findAllWorkers();
    queueWork(maxSid, agents);

    return true;
}

public void queueWork(Long maxId, Set<Worker> workers) {

    sourceQueueRepo.dedupeByMaxSid(maxId);

    List<SourceQueue> batch = sourceQueueRepo.findAllBySidLessThanEqual(maxId);
    Map<Long, List<SourceQueue>> batched = // Redacted

    for (Worker worker : workers) {
        // Method 'batchInsert' calls a save query (transactional)
        batchInsert(worker, batched.getOrDefault(Type.TYPE_1, new HashMap<>()));
        batchInsert(worker, batched.getOrDefault(Type.TYPE_2, new HashMap<>()));
        batchInsert(worker, batched.getOrDefault(Type.TYPE_3, new HashMap<>()));
    }


    sourceQueueRepo.deleteByMaxId(maxId);
}

注意

每个查询都是事务性的，目的是保持数据库事务简短，因为目标表上的其他线程会争用。
插入此队列的代码在此新线程上调用中断，以确保其排空新队列。有多个线程调用它，因此在重负载下停机时间非常少。

【问题讨论】：

据我所见 - 运行上述代码似乎正在更新所有实体的缓存（偶尔会看到此日志消息中列出的一些实体），并且对于此操作，我不依赖任何缓存实体或查询

标签： hibernate hazelcast

【解决方案1】：

我发现问题是 HazelcastLocalCacheRegionFactory 的使用，我的新线程承受了很大的压力，并且缓存区域在传播事件时淹没了 EventQueue。

我选择不改用HazelcastCacheRegionFactory，因为在我的情况下，这对性能产生了不可接受的影响。

相反，对于 @Modifying 查询，我选择使用 NativeQuery 对象并指定我通过 addSynchronizedEntityClass 无效的缓存项。

例如;


return em.createNativeQuery("DELETE FROM dbo.SourceQueue WHERE id <= :id")
    .unwrap(NativeQuery.class)
    .setParameter("id", id)
    .addSynchronizedEntityClass(SourceQueue.class)
    .executeUpdate();

【讨论】：