cassandra java 例子_cassandra(三)--编写java代码对分布式数据库进行操作

本文主要关注两个部分，

1. 怎么写一个最简单cassandra的sample

2. 通过代码,了解cassandra的数据模型及隐藏在后面的交互逻辑

步骤一：

首先我们创建一个工程，然后将cassandra/lib目录下的包，导入到我们的工程中。

步骤二：

创建一个类，内容如下：

Java代码  

importorg.apache.cassandra.thrift.Cassandra;

importorg.apache.cassandra.thrift.Column;

importorg.apache.cassandra.thrift.ColumnPath;

importorg.apache.cassandra.thrift.ConsistencyLevel;

importorg.apache.cassandra.thrift.InvalidRequestException;

importorg.apache.cassandra.thrift.NotFoundException;

importorg.apache.cassandra.thrift.TimedOutException;

importorg.apache.cassandra.thrift.UnavailableException;

importorg.apache.thrift.TException;

importorg.apache.thrift.protocol.TBinaryProtocol;

importorg.apache.thrift.transport.TSocket;

importorg.apache.thrift.transport.TTransport;

importorg.apache.thrift.transport.TTransportException;

publicclassSampleOne {

staticCassandra.Client cassandraClient;

staticTTransport socket;

privatestaticvoidinit()throwsTTransportException {

String server ="192.168.1.129";

//       String server = "localhost";

intport =9160;

/* 首先指定cassandra server的地址 */

socket =newTSocket(server, port);

System.out.println(" connected to "+ server +":"+ port +".");

/* 指定通信协议为二进制流协议 */

TBinaryProtocol binaryProtocol =newTBinaryProtocol(socket,false,false);

cassandraClient =newCassandra.Client(binaryProtocol);

/* 建立通信连接 */

socket.open();

}

publicstaticvoidmain(String[] args)throwsTException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException {

/* 初始化连接 */

init();

/* 选择需要操作的Keyspaces， 可以理解成数据库的表 */

String keyspace="Keyspace1";

String row ="employee";

/* 创建一个Table Name */

String tableName ="Standard2";

/* 插入一条记录 */

insertOrUpdate(keyspace,tableName,row,"name","happy birthday!",System.currentTimeMillis());

/* 删除一条记录 */

//delete(keyspace,tableName,row,"name",System.currentTimeMillis());

/* 获取一条记录 (由于插入和删除是同一条记录,有可能会检索不到哦!请大家主意!*/

Column column = getByColumn(keyspace,tableName,row,"name", System.currentTimeMillis());

System.out.println("read row "+ row);

System.out.println("column name "+":"+newString(column.name));

System.out.println("column value"+":"+newString(column.value));

System.out.println("column timestamp"+":"+ (column.timestamp));

close();

}

/**

* 插入记录

*/

publicstaticvoidinsertOrUpdate(String tableSpace,String tableName, String rowParam,String ColumnName,String ColumnValue,longtimeStamp)

throwsTException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException{

/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */

String keyspace= tableSpace;

/* 数据所在的行标 */

String row = rowParam;

/* 创建一个column path */

ColumnPath col =newColumnPath(tableName);

col.setColumn(ColumnName.getBytes());

/* 执行插入操作，指定keysapce, row, col, 和数据内容， 后面两个参数一个是timestamp， 另外一个是consistency_level

* timestamp是用来做数据一致性保证的， 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo

*/

cassandraClient.insert(keyspace, row, col,"i don't know".getBytes(), System.currentTimeMillis(), ConsistencyLevel.ONE);

}

/**

* 删除记录

*/

publicstaticvoiddelete(String tableSpace,String tableName, String rowParam,String ColumnName,longtimeStamp)

throwsTException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException{

/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */

String keyspace= tableSpace;

/* 数据所在的行标 */

String row = rowParam;

/* 创建一个column path */

ColumnPath col =newColumnPath(tableName);

col.setColumn(ColumnName.getBytes());

/* 执行删除操作，指定keysapce, row, col， 后面两个参数一个是timestamp， 另外一个是consistency_level

* timestamp是用来做数据一致性保证的， 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo

*/

cassandraClient.remove(keyspace, row, col, System.currentTimeMillis(), ConsistencyLevel.ONE);

}

/**

* 获取数据

*/

publicstaticColumn getByColumn(String tableSpace,String tableName, String rowParam,String ColumnName,longtimeStamp)

throwsTException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException{

/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */

String keyspace= tableSpace;

/* 数据所在的行标 */

String row = rowParam;

/* 创建一个column path */

ColumnPath col =newColumnPath(tableName);

col.setColumn(ColumnName.getBytes());

/* 执行查询操作，指定keysapce, row, col， timestamp

* timestamp是用来做数据一致性保证的， 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo

*/

Column column = cassandraClient.get(keyspace, row, col, ConsistencyLevel.ONE).column;

returncolumn;

}

/**

* 关闭当前的远程访问连接

*/

publicstaticvoidclose() {

socket.close();

}

}

import org.apache.cassandra.thrift.Cassandra;

import org.apache.cassandra.thrift.Column;

import org.apache.cassandra.thrift.ColumnPath;

import org.apache.cassandra.thrift.ConsistencyLevel;

import org.apache.cassandra.thrift.InvalidRequestException;

import org.apache.cassandra.thrift.NotFoundException;

import org.apache.cassandra.thrift.TimedOutException;

import org.apache.cassandra.thrift.UnavailableException;

import org.apache.thrift.TException;

import org.apache.thrift.protocol.TBinaryProtocol;

import org.apache.thrift.transport.TSocket;

import org.apache.thrift.transport.TTransport;

import org.apache.thrift.transport.TTransportException;

public class SampleOne {

static Cassandra.Client cassandraClient;

static TTransport socket;

private static void init() throws TTransportException {

String server = "192.168.1.129";

// String server = "localhost";

int port = 9160;

/* 首先指定cassandra server的地址 */

socket = new TSocket(server, port);

System.out.println(" connected to " + server + ":" + port + ".");

/* 指定通信协议为二进制流协议 */

TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false);

cassandraClient = new Cassandra.Client(binaryProtocol);

/* 建立通信连接 */

socket.open();

}

public static void main(String[] args) throws TException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException {

/* 初始化连接 */

init();

/* 选择需要操作的Keyspaces， 可以理解成数据库的表 */

String keyspace= "Keyspace1";

String row = "employee";

/* 创建一个Table Name */

String tableName = "Standard2";

/* 插入一条记录 */

insertOrUpdate(keyspace,tableName,row,"name","happy birthday!",System.currentTimeMillis());

/* 删除一条记录 */

//delete(keyspace,tableName,row,"name",System.currentTimeMillis());

/* 获取一条记录 (由于插入和删除是同一条记录,有可能会检索不到哦!请大家主意!*/

Column column = getByColumn(keyspace,tableName,row,"name", System.currentTimeMillis());

System.out.println("read row " + row);

System.out.println("column name " + ":" + new String(column.name));

System.out.println("column value" + ":" + new String(column.value));

System.out.println("column timestamp" + ":" + (column.timestamp));

close();

}

/**

* 插入记录

*/

public static void insertOrUpdate(String tableSpace,String tableName, String rowParam,String ColumnName,String ColumnValue,long timeStamp)

throws TException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException{

/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */

String keyspace= tableSpace;

/* 数据所在的行标 */

String row = rowParam;

/* 创建一个column path */

ColumnPath col = new ColumnPath(tableName);

col.setColumn(ColumnName.getBytes());

/* 执行插入操作，指定keysapce, row, col, 和数据内容， 后面两个参数一个是timestamp， 另外一个是consistency_level

* timestamp是用来做数据一致性保证的， 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo

*/

cassandraClient.insert(keyspace, row, col,"i don't know".getBytes(), System.currentTimeMillis(), ConsistencyLevel.ONE);

}

/**

* 删除记录

*/

public static void delete(String tableSpace,String tableName, String rowParam,String ColumnName,long timeStamp)

throws TException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException{

/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */

String keyspace= tableSpace;

/* 数据所在的行标 */

String row = rowParam;

/* 创建一个column path */

ColumnPath col = new ColumnPath(tableName);

col.setColumn(ColumnName.getBytes());

/* 执行删除操作，指定keysapce, row, col， 后面两个参数一个是timestamp， 另外一个是consistency_level

* timestamp是用来做数据一致性保证的， 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo

*/

cassandraClient.remove(keyspace, row, col, System.currentTimeMillis(), ConsistencyLevel.ONE);

}

/**

* 获取数据

*/

public static Column getByColumn(String tableSpace,String tableName, String rowParam,String ColumnName,long timeStamp)

throws TException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException{

/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */

String keyspace= tableSpace;

/* 数据所在的行标 */

String row = rowParam;

/* 创建一个column path */

ColumnPath col = new ColumnPath(tableName);

col.setColumn(ColumnName.getBytes());

/* 执行查询操作，指定keysapce, row, col， timestamp

* timestamp是用来做数据一致性保证的， 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo

*/

Column column = cassandraClient.get(keyspace, row, col, ConsistencyLevel.ONE).column;

return column;

}

/**

*关闭当前的远程访问连接

*/

public static void close() {

socket.close();

}

}

为了比较好的理解这些名词解释,我们先看看cassandra的数据模型:

Cassandra 的数据模型的基本概念：

keyspace:

用于存放 ColumnFamily 的容器，相当于关系数据库中的 Schema 或 database,

ColumnFamily :

用于存放 Column 的容器，类似关系数据库中的 table 的概念。

SuperColumn ：

它是一个特列殊的 Column, 它的 Value 值可以包函多个 Column

Java代码  

{// 这是一个SuperColumn

name:"李明杰",

// 包含一系列的Columns

value: {

street: {name:"street", value:"1234 x street", timestamp:123456789},

city: {name:"city", value:"san francisco", timestamp:123456789},

zip: {name:"zip", value:"94107", timestamp:123456789},

}

}

{ // 这是一个SuperColumn

name: "李明杰",

// 包含一系列的Columns

value: {

street: {name: "street", value: "1234 x street", timestamp: 123456789},

city: {name: "city", value: "san francisco", timestamp: 123456789},

zip: {name: "zip", value: "94107", timestamp: 123456789},

}

}

Columns：

Cassandra 的最基本单位。由 name , value , timestamp 组成

Java代码  

{// 这是一个column

name:"李明杰",

value:"mydream.limj@gmali.com",

timestamp:123456789

}

{ // 这是一个column

name: "李明杰",

value: "mydream.limj@gmali.com",

timestamp: 123456789

}

cassandra的数据模型主要就是由上述几种模型构建而成的,很简单吧,的确是这样,最大的好处就是读写数据的API非常简单.