首页 文章

如何在Cassandra中使用datastax java驱动程序有效地使用预准备语句?

提问于
浏览
7

我需要使用Datastax Java驱动程序查询Cassandra中的一个表 . 以下是我的代码,它的工作正常 -

public class TestCassandra {

        private Session session = null;
        private Cluster cluster = null;

        private static class ConnectionHolder {
            static final TestCassandra connection = new TestCassandra();
        }

        public static TestCassandra getInstance() {
            return ConnectionHolder.connection;
        }

        private TestCassandra() {
            Builder builder = Cluster.builder();
            builder.addContactPoints("127.0.0.1");

            PoolingOptions opts = new PoolingOptions();
            opts.setCoreConnectionsPerHost(HostDistance.LOCAL, opts.getCoreConnectionsPerHost(HostDistance.LOCAL));

            cluster = builder.withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE).withPoolingOptions(opts)
                    .withLoadBalancingPolicy(new TokenAwarePolicy(new DCAwareRoundRobinPolicy("DC2")))
                    .withReconnectionPolicy(new ConstantReconnectionPolicy(100L))
                    .build();
            session = cluster.connect();
        }

    private Set<String> getRandomUsers() {
        Set<String> userList = new HashSet<String>();

        for (int table = 0; table < 14; table++) {
            String sql = "select * from testkeyspace.test_table_" + table + ";";

            try {
                SimpleStatement query = new SimpleStatement(sql);
                query.setConsistencyLevel(ConsistencyLevel.QUORUM);
                ResultSet res = session.execute(query);

                Iterator<Row> rows = res.iterator();
                while (rows.hasNext()) {
                    Row r = rows.next();

                    String user_id = r.getString("user_id");
                    userList.add(user_id);
                }
            } catch (Exception e) {
                System.out.println("error= " + ExceptionUtils.getStackTrace(e));
            }
        }

        return userList;
    }
}

我在我的主要应用程序中使用上面这样的类 -

TestCassandra.getInstance().getRandomUsers();

有没有什么办法可以有效地使用 PreparedStatementgetRandomUsers ?我想我需要确保我只创建一次 PreparedStatement 而不是多次创建它 . 在我当前的架构中,最好的设计是什么?如何使用它?

2 回答

  • 14

    您可以创建所需语句的缓存(这是一个相当基本的例子来提供您的想法) . 让我们从创建将用作缓存的类开始 .

    private class StatementCache {
        Map<String, PreparedStatement> statementCache = new HashMap<>();
        public BoundStatement getStatement(String cql) {
            PreparedStatement ps = statementCache.get(cql);
            // no statement cached, create one and cache it now.
            if (ps == null) {
                ps = session.prepare(cql);
                statementCache.put(cql, ps);
            }
            return ps.bind();
        }
    }
    

    然后在您的单例中添加一个实例:

    public class TestCassandra {
        private Session session = null;
        private Cluster cluster = null;
        private StatementCache psCache = new StatementCache();
        // rest of class...
    

    最后使用你的函数缓存:

    private Set<String> getRandomUsers(PreparedStatement ps) {
    // lots of code.    
            try {
                SimpleStatement query = new SimpleStatement(sql);
                query.setConsistencyLevel(ConsistencyLevel.QUORUM);
                // abstract the handling of the cache to it's own class.
                // this will need some work to make sure it's thread safe
                // as currently it's not.
                ResultSet res = session.execute(psCache.getStatement(sql));
    
  • 1

    我的实现与上面分享的实现大致相同,但是通过性能检查和实现来处理竞争条件 . 在我的思考过程中查看代码的内联注释 .

    import com.datastax.driver.core.PreparedStatement;
     import com.datastax.driver.core.Session;
     import nl.ing.creditcards.commons.activity.ActivityException;
    
     import java.util.Map;
     import java.util.concurrent.ConcurrentHashMap;
    
    public class StatementCache {
    
    /* prevent cache incoherence issues*/
    private static volatile StatementCache sCacheInstance;
    private static final Map<String, PreparedStatement> holder = new ConcurrentHashMap<>();
    private static final String NOT_PERMITTED = "Operation not permitted";
    
    private StatementCache() {
        /*Prevent access through reflection api.*/
        if (sCacheInstance != null) {
            throw new ActivityException(NOT_PERMITTED, "Use getInstance() to retrieve the instance of this class");
        }
    }
    
    /**
     * Double check locking pattern usage for singleton classes
     *
     * @return
     */
    public static StatementCache getInstance() {
        if (sCacheInstance == null) { //Check for the first time
            synchronized (StatementCache.class) { // second check in order to keep the operation atomic
                if (sCacheInstance == null) sCacheInstance = new StatementCache();
            }
        }
        return sCacheInstance;
    }
    
    /**
     * If {@link StatementCache#getStatement#prepared_statement} is already present in cache,
     * then we don't have to synchronize and make threads wait, otherwise, we synchronize the caching bit.
     *
     * @param session
     * @param cql
     * @return
     */
    public PreparedStatement getStatement(Session session, String cql) {
        PreparedStatement prepared_statement = holder.get(cql);
        if (prepared_statement == null) {
            synchronized (this) {
                prepared_statement = holder.get(cql);
                if (prepared_statement == null) {
                    prepared_statement = session.prepare(cql);
                    holder.put(cql, prepared_statement);
                }
            }
        }
        return prepared_statement;
      }
    }
    

    使用此缓存单例类将如下所示:

    public class CacheConsumer{
    
        private static Session session;
    
        CacheConsumer(Session session){
         this.session=session;
       }
    
        public void someMethod(){
          String cqlstatement = "SELECT * FROM SOME_TABLE";
          PreparedStatement statement= 
           StatementCache.getInstance().getStatement(session,cqlstatement);
             // You can now use the prepared statement however you wish.
       }
    }
    

    很简单;)

相关问题