Apache Cassandra Learning Step by Step (5): 实战性的JTwissandra项目

glcui

　　在完成了Apache Cassandra的四个基本学习步骤之后，可以尝试下实战性的编码了。
　　如有必要，建议再简单回顾一下：

Apache Cassandra Learning Step by Step (1)

Apache Cassandra Learning Step by Step (2): Core Concepts

Apache Cassandra Learning Step by Step (3): Samples ABC

Apache Cassandra Learning Step by Step (4): Data Modeling
　　基于第四点的建模思路，接下来我们要做的，就是搭建一个叫做JTwissandra的实战性项目，就是所谓的Java版本的Twissandra了。
　　其目的是为了以Twitter为假想对象，使用最简约（或者直接说简陋得了）的建模和实现，表达采用Apache Cassandra作为NoSQL平台的基本实现过程。
　　JTwissandra的基本编码环境：
　　1. Maven来管理
　　2. JUnit来测试
　　3. 基于hector client来作为Apache Cassandra的Java 客户端
　　大家可以通过下面的Github链接，直接clone出来最新的代码:
　　JTwissandra: https://github.com/itstarting/jtwissandra
　　也欢迎大家Fork或在这里直接拍砖——反正咱在NoSQL也是新手，脸皮厚点不要紧啦：）
　　1. 首先需要一个HFactoryHelper来初始化并建立Cassandra的客户端连接池和必要的对象：

import java.io.IOException;
import java.util.Properties;
import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Helper for Cassandra initialization
*
* @author bright_zheng
*
*/
public class HFactoryHelper {
private static Logger logger = LoggerFactory.getLogger(HFactoryHelper.class);
private static Cluster cluster;
private static Keyspace keyspace = initKeyspace();
private static Properties properties;
private HFactoryHelper(){}
public static Keyspace getKeyspace(){
return keyspace;
}
private static Keyspace initKeyspace() {
properties = new Properties();
try {
properties.load(HFactoryHelper.class.getResourceAsStream("/config.properties"));
} catch (IOException ioe) {
ioe.printStackTrace();
}
String cluster_name = properties.getProperty("cluster.name", "Test Cluster");
logger.debug("cluster.name={}", cluster_name);
String cluster_hosts = properties.getProperty("cluster.hosts", "127.0.0.1:9160");
logger.debug("cluster.hosts={}", cluster_hosts);
String active_keyspace = properties.getProperty("keyspace", "JTWISSANDRA");
logger.debug("keyspace={}", active_keyspace);
cluster = HFactory.getOrCreateCluster(cluster_name, cluster_hosts);
ConfigurableConsistencyLevel ccl = new ConfigurableConsistencyLevel();
ccl.setDefaultReadConsistencyLevel(HConsistencyLevel.ONE);
return HFactory.createKeyspace(
active_keyspace,
cluster,
ccl);
}
}
　　2. 建立各项业务服务的基类BaseService。
　　import java.util.UUID;

import me.prettyprint.cassandra.serializers.LongSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.clock.MicrosecondsClockResolution;
import me.prettyprint.cassandra.utils.TimeUUIDUtils;
import me.prettyprint.hector.api.ClockResolution;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.factory.HFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import bright.zheng.jtwissandra.HFactoryHelper;
/**
* Base service which all business services should extend
*
* @author bright_zheng
*
*/
public class BaseService {
protected Logger logger = LoggerFactory.getLogger(getClass());
protected static Keyspace KEYSPACE = HFactoryHelper.getKeyspace();
protected static final String CF_USER = "USER";
protected static final String CF_FRIEND = "FRIEND";
protected static final String CF_FOLLOWER = "FOLLOWER";
protected static final String CF_TWEET = "TWEET";
protected static final String CF_TIMELINE = "TIMELINE";
protected static final StringSerializer SERIALIZER_STRING
= StringSerializer.get();
protected static final LongSerializer SERIALIZER_LONG
= LongSerializer.get();
protected static final int TWEETS_LIMIT_DEFAULT = 10;
protected static final int TWEETS_LIMIT_MAX = 50;
protected HColumn<String, String> createColumn(String name, String value) {
return HFactory.createColumn(name, value, SERIALIZER_STRING, SERIALIZER_STRING);
}
protected HColumn<String, Long> createColumn(String name, Long value) {
return HFactory.createColumn(name, value, SERIALIZER_STRING, SERIALIZER_LONG);
}
protected HColumn<Long, String> createColumn(Long name, String value) {
return HFactory.createColumn(name, value, SERIALIZER_LONG, SERIALIZER_STRING);
}
/**
* REF: http://wiki.apache.org/cassandra/FAQ#working_with_timeuuid_in_java
*
* @return UUID
*/
public UUID getUUID(){
//TODO: which UUID should we use to make sure it's unique?
ClockResolution clock = new MicrosecondsClockResolution();
return TimeUUIDUtils.getTimeUUID(clock);
//return TimeUUIDUtils.getUniqueTimeUUIDinMillis();
}
protected Long getTimestamp(UUID uuid){
//return uuid.timestamp();
return TimeUUIDUtils.getTimeFromUUID(uuid);
}
protected Long generateTimestamp(){
return getTimestamp(getUUID());
}
}
 　　3. 下面是各项业务服务代码：
　　3.1 UserService

package bright.zheng.jtwissandra.service;
import java.util.UUID;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import bright.zheng.jtwissandra.bean.User;
/**
* User service
*
* @author bright_zheng
*
*/
public class UserService extends BaseService{
/**
* Sample CLI cmd:
* set USER['550e8400-e29b-41d4-a716-446655440000']['user_name'] = 'itstarting';
* set USER['550e8400-e29b-41d4-a716-446655440000']['password'] = '111222';
* set USER['550e8400-e29b-41d4-a716-446655440000']['create_timestamp'] = 1329836819890000;
*
* @param user
*/
public String addUser(User user) {
Mutator<String> mutator = HFactory.createMutator(
KEYSPACE, SERIALIZER_STRING);
UUID uuid = this.getUUID();
String user_uuid = uuid.toString();   
Long create_timestamp = this.getTimestamp(uuid);
logger.debug("user_uuid={}", user_uuid);  
logger.debug("user_name={}", user.getUser_name());
logger.debug("password={}", user.getUser_password());
logger.debug("create_timestamp={}", create_timestamp);
mutator.addInsertion(user_uuid, CF_USER,
this.createColumn("user_name", user.getUser_name()));
mutator.addInsertion(user_uuid, CF_USER,
this.createColumn("password", user.getUser_password()));
mutator.addInsertion(user_uuid, CF_USER,
this.createColumn("create_timestamp", create_timestamp));
mutator.execute();
//return the generated uuid
return user_uuid;
}   
}

 　　3.2 FriendService

package bright.zheng.jtwissandra.service;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.MutationResult;
import me.prettyprint.hector.api.mutation.Mutator;
/**
* Friend service
*
* @author bright_zheng
*
*/
public class FriendService extends BaseService{
/**
* Adding a friend has two business logic:
* 1. Add the friend's uuid to the Friend CF under my uuid
* 2. Add my uuid to the friend's uuid as follower
*
* set FRIEND['550e8400-e29b-41d4-a716-446655440000']['1329836819859000']
* = '550e8400-e29b-41d4-a716-446655440001;
*
* set FOLLOWER['550e8400-e29b-41d4-a716-446655440001']['1329836819859000'']
* = '550e8400-e29b-41d4-a716-446655440000;
*
* @param me
* @param friend
*/
public MutationResult followFriend(String me, String friend) {
Mutator<String> mutator = HFactory.createMutator(
KEYSPACE, SERIALIZER_STRING);
Long timestamp = this.generateTimestamp();
logger.debug("timestamp={}", timestamp);
mutator.addInsertion(me, CF_FRIEND,
this.createColumn(timestamp, friend));
mutator.addInsertion(friend, CF_FOLLOWER,
this.createColumn(timestamp, me));
return mutator.execute();
}   
}

 　　3.3 TimelineService

package bright.zheng.jtwissandra.service;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SliceQuery;
import bright.zheng.jtwissandra.bean.Timeline;
/**
* Timeline service
*
* @author bright_zheng
*
*/
public class TimelineService extends BaseService{
/**
* get specified user's first Timeline
*
* @param user_uuid
* @return
*/
public TimelineWrapper getTimeline(String user_uuid){
return getTimeline(user_uuid, 0L, TWEETS_LIMIT_DEFAULT);
}
/**
* get specified user's Timeline with start point
*
* @param user_uuid
* @param start
* @return
*/
public TimelineWrapper getTimeline(String user_uuid, long start){
return getTimeline(user_uuid, start, TWEETS_LIMIT_DEFAULT);
}
/**
* get specified user's Timeline with start point and limit
*
* @param user_uuid
* @param start
* @param limit
* @return
*/
public TimelineWrapper getTimeline(String user_uuid, long start, int limit){
if (start<0) start = 0;
if (limit<0) limit = TWEETS_LIMIT_DEFAULT;
if (limit>TWEETS_LIMIT_MAX) limit = TWEETS_LIMIT_MAX;
SliceQuery<String, Long, String> sliceQuery =
HFactory.createSliceQuery(KEYSPACE, SERIALIZER_STRING, SERIALIZER_LONG, SERIALIZER_STRING);        
sliceQuery.setColumnFamily(CF_TIMELINE);
sliceQuery.setKey(user_uuid);
sliceQuery.setRange(start, Long.MAX_VALUE, false, limit+1);
QueryResult<ColumnSlice<Long, String>> result = sliceQuery.execute();
List<HColumn<Long, String>> list = result.get().getColumns();
long next = 0L;
if(list==null){
return new TimelineWrapper(null, next);
}else if (list.size()<=limit){        
return new TimelineWrapper(convertToTimeline(list), 0L);
}else{
HColumn<Long,String> last = list.get(list.size()-1);
next = last.getName(); //the name is the timestamp as the "next" start
list.remove(list.size()-1);
return new TimelineWrapper(convertToTimeline(list), next);
}
}
private List<Timeline> convertToTimeline(List<HColumn<Long,String>> cols){
Iterator<HColumn<Long,String>> it = cols.iterator();
List<Timeline> result = new ArrayList<Timeline>();
while(it.hasNext()){
HColumn<Long,String> col = it.next();
result.add(new Timeline(col.getValue(), col.getName()));
}
return result;
}
public class TimelineWrapper{
private List<Timeline> timelines;
private long nextTimeline;
public TimelineWrapper(List<Timeline> timelines, long nextTimeline){
this.timelines = timelines;
this.nextTimeline = nextTimeline;
}
public long getNextTimeline() {
return nextTimeline;
}
public List<Timeline> getTimelines() {
return timelines;
}
}
}

 　　3.4 TweetService

package bright.zheng.jtwissandra.service;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.beans.Rows;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.MultigetSliceQuery;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SliceQuery;
import bright.zheng.jtwissandra.bean.Tweet;
/**
* Tweet service
*
* @author bright_zheng
*
*/
public class TweetService extends BaseService{
/**
* Adding a tweet has following logic:
* 1. Save the tweet to CF of TWEET
* 2. Add the new tweet to my TIMELINE
* 3. Add the new tweet to all my followers' TIMELINE
*
* @param me
* @param friend
*/
public String addTweet(String user_uuid, String tweet_content) {
Mutator<String> mutator = HFactory.createMutator(
KEYSPACE, SERIALIZER_STRING);
//the tweet uuid
UUID uuid = this.getUUID();
String tweet_uuid = uuid.toString();
logger.debug("tweet_uuid={}", tweet_uuid);
//the timestamp to build the timeline
Long timestamp = this.getTimestamp(uuid);
logger.debug("timestamp={}", timestamp);
mutator.addInsertion(tweet_uuid, CF_TWEET,
this.createColumn("user_uuid", user_uuid));
mutator.addInsertion(tweet_uuid, CF_TWEET,
this.createColumn("tweet_content", tweet_content));
mutator.addInsertion(user_uuid, CF_TIMELINE,
this.createColumn(timestamp, tweet_uuid));
// get back all my follower and insert the tweet to his/her TIMELINE
SliceQuery<String, Long, String> sliceQuery =
HFactory.createSliceQuery(KEYSPACE, SERIALIZER_STRING, SERIALIZER_LONG, SERIALIZER_STRING);        
sliceQuery.setColumnFamily(CF_FOLLOWER);
sliceQuery.setKey(user_uuid);
sliceQuery.setRange(Long.MIN_VALUE, Long.MAX_VALUE, false, 500); //TODO: 500 followers hard code here?
QueryResult<ColumnSlice<Long, String>> result = sliceQuery.execute();
Iterator<HColumn<Long, String>> followers = result.get().getColumns().iterator();
while(followers.hasNext()) {
HColumn<Long, String> follower = followers.next();
String follower_uuid = follower.getValue();
logger.debug("follower's uuid={}", follower_uuid);
logger.debug("timestamp={}", follower.getName());
//insert the tweet to the follower's TIMELINE
mutator.addInsertion(follower_uuid, CF_TIMELINE,
this.createColumn(timestamp, tweet_uuid));
}
mutator.execute();
//return the new generated tweet's uuid
return tweet_uuid;
}
/**
* Should we add this service?
*
* @param tweet_uuid
* @return
*/
public Tweet getTweet(String tweet_uuid){
return null;
}
public List<Tweet> getTweets(List<String> tweet_uuids){
MultigetSliceQuery<String, String, String> multigetSlicesQuery =
HFactory.createMultigetSliceQuery(KEYSPACE, SERIALIZER_STRING, SERIALIZER_STRING, SERIALIZER_STRING);
multigetSlicesQuery.setColumnFamily(CF_TWEET);
multigetSlicesQuery.setColumnNames("user_uuid","tweet_content");        
multigetSlicesQuery.setKeys(tweet_uuids);
QueryResult<Rows<String, String, String>> results = multigetSlicesQuery.execute();
return convertRowsToTweets(results.get());
}
private List<Tweet> convertRowsToTweets(Rows<String, String, String> rows){
List<Tweet> list = new ArrayList<Tweet>();
Iterator<Row<String, String, String>> iterator = rows.iterator();
while(iterator.hasNext()){
Row<String, String, String> row = iterator.next();
ColumnSlice<String, String> cs = row.getColumnSlice();
list.add(new Tweet(row.getKey(),
cs.getColumnByName("tweet_content").getValue(),
cs.getColumnByName("user_uuid").getValue()));
}
return list;
}
}

 　　4. 当然少不了JUnit测试用例了：


package bright.zheng.jtwissandra;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import junit.framework.Assert;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import bright.zheng.jtwissandra.bean.Timeline;
import bright.zheng.jtwissandra.bean.Tweet;
import bright.zheng.jtwissandra.bean.User;
import bright.zheng.jtwissandra.service.FriendService;
import bright.zheng.jtwissandra.service.TimelineService;
import bright.zheng.jtwissandra.service.TimelineService.TimelineWrapper;
import bright.zheng.jtwissandra.service.TweetService;
import bright.zheng.jtwissandra.service.UserService;
/**
* Test cases for all services currently provided.
* Please drop and create schema first and then run all cases as one round
* The 'me' and 'friend' will be created each round dynamically for easier testing
*
* @author bright_zheng
*
*/
public class ServiceTest{
Logger logger = LoggerFactory.getLogger(ServiceTest.class);
private static UserService SERVICE_USER = new UserService();
private static FriendService SERVICE_FRIEND = new FriendService();
private static TweetService SERVICE_TWEET = new TweetService();
private static TimelineService SERVICE_TIMELINE = new TimelineService();
private static String me;
private static String friend;
private static long nextTimeline = 0L;
@BeforeClass
public static void setUp(){
//
}
@Test
public void addUser() {
logger.debug("=====================addUser{====================");
//add user 1
me = SERVICE_USER.addUser(new User("itstarting","1234"));
logger.debug("This round of tesing, ME={}", me);
Assert.assertNotNull(me);
//add user 2
friend = SERVICE_USER.addUser(new User("test1","1234"));
logger.debug("This round of tesing, FRIEND={}", friend);
Assert.assertNotNull(friend);
logger.debug("=====================}//addUser====================");
}   
/**
* I'm following a friend
*/
@Test
public void followFriend() {
logger.debug("=====================followFriend{====================");
SERVICE_FRIEND.followFriend(me, friend);
logger.debug("=====================}//followFriend====================");
}
/**
* I'm followed by a follower
*/
@Test
public void followedByFollower() {
logger.debug("=====================followedByFollower{====================");
SERVICE_FRIEND.followFriend(friend, me);
logger.debug("=====================}//followedByFollower====================");
}
/**
* I'm twittering
*/
@Test
public void addTweetByMe() {
logger.debug("=====================addTweetByMe{====================");
for(int i=0; i<100; i++){
String tweet_uuid = SERVICE_TWEET.addTweet(me, "Hellow JTWISSANDRA -- by itstarting:" + i);
Assert.assertNotNull(tweet_uuid);
}
logger.debug("=====================}//addTweetByMe====================");
}
/**
* My friend is twittering
*
*/
@Test
public void addTweetByFriend() {
logger.debug("=====================addTweetByFriend{====================");
for(int i=0; i<100; i++){
String tweet_uuid = SERVICE_TWEET.addTweet(friend, "Hellow JTWISSANDRA -- by test1:" + i);
Assert.assertNotNull(tweet_uuid);
}
logger.debug("=====================}//addTweetByFriend====================");
}
/**
* Get tweets for me
*/
@Test
public void getTweetsByMe(){
logger.debug("=====================getTweetsByMe{====================");
getTweets(me, 0);
logger.debug("=====================}//getTweetsByMe====================");
}
/**
* Get tweets at next Timeline (if any)
*/
@Test
public void getTweetsByMeForNextTimeline(){
logger.debug("=====================getTweetsByMeForNextTimeline{====================");
if(nextTimeline>0L){
getTweets(me, nextTimeline);
}
logger.debug("=====================}//getTweetsByMeForNextTimeline====================");
}
/**
* Get tweets for my friend
*/
@Test
public void getTweetsByMyFriend(){
logger.debug("=====================getTweetsByMyFriend{====================");
getTweets(friend, 0);
logger.debug("=====================}//getTweetsByMyFriend====================");
}
/**
*
*/
@Test
public void getTweetsByMyFriendForNextTimeline(){
logger.debug("=====================getTweetsByMyFriendForNextTimeline{====================");
getTweets(friend, nextTimeline);
logger.debug("=====================}//getTweetsByMyFriendForNextTimeline====================");
}
private void getTweets(String user_uuid, long start){
TimelineWrapper wrapper = SERVICE_TIMELINE.getTimeline(user_uuid, start);
Assert.assertNotNull(wrapper);
List<Timeline> list = wrapper.getTimelines();
List<String> tweet_uuids = new ArrayList<String>();
for(Timeline timeline: list){
String tweet_uuid = timeline.getTweet_uuid();
logger.debug("From Timeline: tweet_uuid={}, tweet_timestamp={}",
tweet_uuid, timeline.getTweet_timestamp());
tweet_uuids.add(tweet_uuid);
}
List<Tweet> tweets = SERVICE_TWEET.getTweets(tweet_uuids);
Iterator<Tweet> it = tweets.iterator();
while(it.hasNext()){
Tweet tweet = it.next();
logger.debug("From Tweet: tweet_uuid={}, tweet_content={}, user_uuid={}",
new Object[]{tweet.getTweet_uuid(),
tweet.getTweet_content(),
tweet.getUser_uuid()
});
}
if(wrapper.getNextTimeline() > 0L){
logger.debug("The start timeline of next page is: {}", wrapper.getNextTimeline());
nextTimeline = wrapper.getNextTimeline();
}else{
logger.debug("No next page available");
nextTimeline = 0L;
}
}
@AfterClass
public static void shutdown(){
//cluster.getConnectionManager().shutdown();
}
}

　　这是个一锅端的测试用例，全部跑一次可以覆盖几乎所有的业务服务逻辑。
　　5. 最后，别忘了在跑之前创建必要的schema：


drop keyspace JTWISSANDRA;
create keyspace JTWISSANDRA
with placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy'
and strategy_options = [{replication_factor:1}];
use JTWISSANDRA;
create column family USER
with comparator = UTF8Type   
and key_validation_class = UTF8Type
and default_validation_class = UTF8Type
and column_metadata = [
{column_name: user_name, validation_class: UTF8Type,
index_name:user_name_idx, index_type:KEYS }
{column_name: user_password, validation_class: UTF8Type}
{column_name: create_timestamp, validation_class: LongType,
index_name:create_timestamp_idx, index_type:KEYS}
];
create column family FRIEND
with comparator = LongType   
and key_validation_class = UTF8Type
and default_validation_class = UTF8Type;
create column family FOLLOWER
with comparator = LongType   
and key_validation_class = UTF8Type
and default_validation_class = UTF8Type;
create column family TWEET
with comparator = UTF8Type   
and key_validation_class = UTF8Type
and default_validation_class = UTF8Type
and column_metadata = [
{column_name: user_uuid, validation_class: UTF8Type}
{column_name: tweet_content, validation_class: UTF8Type}
];
create column family TIMELINE
with comparator = LongType   
and key_validation_class = UTF8Type
and default_validation_class = UTF8Type;

　　6. 其他？
　　有些咋想，就不贴了吧，略过……有兴趣的可从Github clone下来跑跑看。
　　==========所发现的问题，接下来再谈再讨论，最最希望有丰富经验的高手前来助阵，去我谜团，谢谢！==========