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通过线程池检索数据

回答

相关问题