GemFire CacheWriter 中的并发线程

Question

我们目前将 Cassandra 用作 NoSQL 数据库，将 GemFire 用作内存数据库。我们一直在使用 GemFire CacheWriter 在 Cassandra 中插入记录。我希望您能就在 CacheWriter 中使用并发线程来插入/更新记录是否是一种良好的工程实践提供反馈。我们将不胜感激您对此的反馈。

public class GenericWriter<K, V> extends CacheWriterAdapter<K, V> implements Declarable {

    private static Logger log = LoggerFactory.getLogger(GenericWriter.class);

    @Autowired
    private CassandraOperations cassandraOperations;

    ExecutorService executor = null;

    @Override
    public void beforeCreate(EntryEvent<K, V> e) {

    executor = Executors.newSingleThreadExecutor();

        executor.submit(() -> {
            if (eventOperation.equals("CREATE") || eventOperation.equalsIgnoreCase("PUTALL_CREATE")) {
                try {
                    cassandraOperations.insert(e.getNewValue());
                } catch (CassandraConnectionFailureException | CassandraWriteTimeoutException
                        | CassandraInternalException cassException) {
                } catch (Exception ex) {
                    log.error("Exception in GenericCacheWriter->" + ExceptionUtils.getStackTrace(ex));
                    throw ex;
                }
            }
        });
        executor.shutdown();
    }

    @Override
    public void init(Properties arg0) {
        // TODO Auto-generated method stub

    }
}

Answer 1

CacheWriter 处理程序是同步调用的，因此应用程序在处理程序返回之前不会继续。因此，不建议在此侦听器内执行长时间运行的操作。如果需要长时间运行的操作，请考虑通过 AsyncEventListener 异步处理该操作。

使用ExecutorService 将执行委托给不同的线程是可能的，但这是一种反模式，因为它不再实现快速失败属性，并且事件的处理不再是同步的，因此它的相对于应用程序完成事件的时间将无法保证。

您可以在 Geode Wiki 中阅读有关此主题的更多信息，特别是在 CacheWrite and CacheListener Best Practices。

希望这会有所帮助。

最好的问候。

Answer 2

是的，这是一个很好的模式，但删除 Executor 并对数据进行分区，以便 GemFire 的所有更新都转到一个且仅一个节点。以同样的方式对 Cassandra 进行分区。在 Cassandra 更新周围放置一个写锁。仅当您的吞吐量较低时才使用此选项。

如果您需要高吞吐量，请使用 AsyncEventListener 并为您的用户保证最终的一致性。如果您必须在 AEL 中使用 Executor，请以在主线程中引发异常的方式使用它们。如果在多次尝试后更新失败，您将失败的条目写入另一个区域，有效期为几秒或一分钟。到期后，重试该操作。继续这样做直到成功，然后才删除过期的条目。

如果更新顺序对您来说是否重要，您将需要跟踪版本号以及正在更新的内容以查看旧值/新值。