Cassandra通過線程池檢索數據

我對Cassandra相對比較陌生，我試圖使用通過執行程序池執行的預準備語句檢索數據。看起來我收到的數據不一致。Cassandra通過線程池檢索數據

我有這個user_connections表，其中user_id是行鍵，friends_id列表作爲set列。我有另一個表，friends_info表，其中friend_id是行鍵，所有其他信息爲列。

當我試圖檢索用戶AA的朋友列表時，我正在檢索朋友列表BBB，CCC，DDD。這非常好。

當試圖通過使用準備語句的執行程序池檢索BBB，CCC，DDD時。數據不一致。有時候，所有三個記錄都是BBB，有時候所有三個記錄都是，有時候兩個記錄是BBB，一個是CCC等...

我已經提供了我正在使用的方法和相關類，請你幫忙我與此。我知道準備好的聲明是線程安全的，並且可以按預期工作。

public Set<User> listUserConnections(String userId) { 
    Session session = client.getSession(); 

    Set<String> listUserConnectionIds = listUserConnections(userId, session); 

    if (listUserConnectionIds.size() == 0) 
     return new HashSet<User>(); 



    Set<User> listConnectionUserDetails = retrieveUserConnectionProfileInfo(
      listUserConnectionIds, session); 
    return listConnectionUserDetails; 

} 



private Set<User> retrieveUserConnectionProfileInfo(Set<String> listUserConnectionIds, 
     Session session) { 


    Set<Callable<User>> callables = new HashSet<Callable<User>>(); 


    for(String key:listUserConnectionIds){ 



     logger.info("about to add callable" + key); 

     Callable<User> callable = new QueryTask(key); 

     callables.add(callable); 

    } 
    // invoke in parallel 
    List<Future<User>> futures; 
    Set<User> users = new HashSet<User>(); 

    // TODO Revisit this 
    ExecutorService executorPool = Executors.newFixedThreadPool(100); 

    try { 
     futures = executorPool.invokeAll(callables); 
     for (Future<User> f : futures) { 

      User user = f.get(); 
      users.add(user); 

      logger.info("User from future"+user.getUserId()); 

     } 
    } catch (ExecutionException e) { 
     logger.error("Error in retrieving the stores in parallel ", e); 
    } catch (InterruptedException e) { 
     logger.error("Error in retrieving the stores in parallel as it was interrupted ", e); 
    } finally { 
     executorPool.shutdown(); 
    } 

    return users; 
}

//執行人池類

class QueryTask implements Callable<User> { 

    private String userName; 

    // final PreparedStatement statement = 
    // client.getSession().prepare(QueryConstants.GET_ALL_STORE_BRANDS); 

    QueryTask(String name) { 
     this.userName = name; 

    } 

    @Override 
    public User call() throws Exception { 
     // -------------I am seeing the userName is correct------------- for example BBB 
     logger.info("inside call processing queries for " + userName); 

     //------------This is a prepared statement, userPreparedStatement.getbStUserInfo() 
     BoundStatement bStUserInfo = userPreparedStatement.getbStUserInfo(); 
     bStUserInfo.bind(userName); 
     Session session = client.getSession(); 
     ResultSet rs = session.execute(bStUserInfo); 

     User user = new User(); 

     Iterator<Row> rowIterator = rs.iterator(); 
     while (rowIterator.hasNext()) 
     { 
     Row row = rowIterator.next(); 

     //-------This user id is not right 
     logger.info("Inside the callable after retrieval"+row.getString(TableConstants.Users.COLUMN_NAME_USER_ID)); 
     user.setUserId(row.getString(TableConstants.Users.COLUMN_NAME_USER_ID)); 



     return user; 
     } 

     logger.info("outside the while loop"); 
     return user; 

    } 

}

來源

2016-04-10 fsheriff

我沒有看到你的'BoundStatement'實例在哪裏。 – Ralf

作爲一個方面說明，驅動程序已經爲您提供了一個異步API，因此您可以擺脫幾乎所有的QueryTask/Executor代碼並專注於您的測試。 –

謝謝你這麼多@Ralf和亞歷克斯·波佩斯庫的又回到了我這一點。 Datastax有一個深入探討Aync調用如何工作的文檔。

@ Alex Popescu。謝謝我嘗試了他們的異步調用，它對我來說工作得很好。

來源

2016-04-13 21:00:07 fsheriff

Cassandra通過線程池檢索數據

回答

相關問題