我正在使用DataStax Java Driver 3.1.0连接到Cassandra集群,我的Cassandra集群版本为2.0.10。我正在使用
QUORUM
一致性进行异步写入。 private final ExecutorService executorService = Executors.newFixedThreadPool(10);
private final Semaphore concurrentQueries = new Semaphore(1000);
public void save(String process, int clientid, long deviceid) {
String sql = "insert into storage (process, clientid, deviceid) values (?, ?, ?)";
try {
BoundStatement bs = CacheStatement.getInstance().getStatement(sql);
bs.setConsistencyLevel(ConsistencyLevel.QUORUM);
bs.setString(0, process);
bs.setInt(1, clientid);
bs.setLong(2, deviceid);
concurrentQueries.acquire();
ResultSetFuture future = session.executeAsync(bs);
Futures.addCallback(future, new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet result) {
concurrentQueries.release();
logger.logInfo("successfully written");
}
@Override
public void onFailure(Throwable t) {
concurrentQueries.release();
logger.logError("error= ", t);
}
}, executorService);
} catch (Exception ex) {
logger.logError("error= ", ex);
}
}
我的保存方法将会被多个线程以非常快的速度调用。如果我写入的速度比我的Cassandra集群处理能力快,那么它将开始抛出错误,并且我希望所有的写入都可以成功地进入Cassandra而没有任何损失。
问题:
我正在考虑使用某种类型的队列或缓冲区来排队请求(例如java.util.concurrent.ArrayBlockingQueue
)。 "缓冲区已满" 将意味着客户端应该等待。缓冲区还将用于重新排队失败的请求。但是为了更公平,失败的请求可能应该放在队列的前面,以便首先重试它们。此外,我们还应该解决队列已满并且有新的失败请求的情况。然后,单线程工作程序将从队列中选择请求并将其发送到Cassandra。由于它不应该做太多事情,因此它不太可能成为瓶颈。这个工作者可以应用自己的速率限制,例如基于时间的com.google.common.util.concurrent.RateLimiter
。
实现这个队列或缓冲区功能并在写入Cassandra时应用特定的guava速率限制的最佳方法是什么?如果有更好的方法,请告诉我。我希望以每秒2000个请求写入Cassandra(这应该是可配置的,以便我可以尝试不同的设置来查看最佳设置)。
如下面的评论中所述,如果内存持续增加,我们可以使用Guava Cache或CLHM来删除旧记录,以确保我的程序不会耗尽内存。我们将在盒子上拥有大约12GB的内存,并且这些记录非常小,因此我认为这不应该是一个问题。