cassandra使用之添加更新和删除

import java.nio.ByteBuffer;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
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.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

public class CassandraTest {
	static Cassandra.Client cassandraClient;
	static TTransport socket;

	private static void init(String keySpace) throws InvalidRequestException, TException {
		String server = "192.168.100.108";
		int port = 9160;
		/* 首先指定的地址 */
		socket = new TSocket(server, port);
		System.out.println(" connected to " + server + ":" + port + ".");
		TFramedTransport transport = new TFramedTransport(socket);
		/* 指定通信协议为二进制流协议 */
		TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport);
		cassandraClient = new Cassandra.Client(binaryProtocol);
		/* 建立通信连接 */
		socket.open();
		cassandraClient.set_keyspace(keySpace);
	}

	public static void main(String[] args) throws TException, TimedOutException,
			InvalidRequestException, UnavailableException, NotFoundException {
		/* 选择需要操作的,可以理解成数据库 */
		String keyspace = "wyqTest";

		/* 初始化连接 */
		init(keyspace);

		/* 创建一个表名*/
		String columnFamily = "userprofile";
		//KEY的名字
		String tablename = "myinfo";

		/* 插入一条记录 */
		insertOrUpdate(columnFamily, tablename, "fengye", "枫叶", System.currentTimeMillis());
		/* 删除一条记录 */
		//delete(columnFamily,tablename,"fengye",System.currentTimeMillis());
		/* 获取一条记录 (由于插入和删除是同一条记录,有可能会检索不到哦!请大家主意! */
		Column column = getByColumn(columnFamily, tablename, "fengye", System.currentTimeMillis());

		System.out.println("read Table " + columnFamily);
		System.out.println("read column name " + ":" + toString(column.name));
		System.out.println("read column value" + ":" + toString(column.value));
		System.out.println("read column timestamp" + ":" + (column.timestamp));
		close();
	}

	/**
	 * 插入记录
	 */
	public static void insertOrUpdate(	String columnFamily,
										String tableName,
										String ColumnName,
										String ColumnValue,
										long timeStamp) throws TException, TimedOutException,
			InvalidRequestException, UnavailableException, NotFoundException {
		/* 创建一个column path */
		ColumnParent parent = new ColumnParent(columnFamily);
		Column col = new Column(CassandraTest.toByteBuffer(ColumnName));
		col.setValue(CassandraTest.toByteBuffer(ColumnValue));
		col.setTimestamp(System.currentTimeMillis());
		try{
			/*
			 * 执行插入操作，指定keysapce, row, col, 和数据内容， 后面两个参数一个是timestamp，
			 * 另外一个是consistency_level timestamp是用来做数据一致性保证的，
			 * 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo
			 */
			cassandraClient.insert(	CassandraTest.toByteBuffer(tableName),
									parent,
									col,
									ConsistencyLevel.ONE);
		}catch(Exception e){
			e.printStackTrace();
		}
	}

	/**
	 * 删除记录
	 */
	public static void delete(	String columnFamily,
								String tablename,
								String ColumnName,
								long timeStamp) throws TException, TimedOutException,
			InvalidRequestException, UnavailableException, NotFoundException {
		/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */
		/* 数据所在的行标 */
		/* 创建一个column path */
		ColumnPath col = new ColumnPath(columnFamily);
		col.setColumn(CassandraTest.toByteBuffer(ColumnName));
		try{
			/*
			 * 执行删除操作，指定keysapce, row, col， 后面两个参数一个是timestamp，
			 * 另外一个是consistency_level timestamp是用来做数据一致性保证的，
			 * 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo
			 */
			cassandraClient.remove(	CassandraTest.toByteBuffer(tablename),
									col,
									System.currentTimeMillis(),
									ConsistencyLevel.ONE);
		}catch(Exception e){
			e.printStackTrace();
		}
	}

	/**
	 * 获取数据
	 */
	public static Column getByColumn(	String columnFamily,
										String tablename,
										String ColumnName,
										long timeStamp) throws TException, TimedOutException,
			InvalidRequestException, UnavailableException, NotFoundException {
		try{
			/*
			 * 执行查询操作，指定keysapce, row, col， timestamp timestamp是用来做数据一致性保证的，
			 * 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo
			 */
			/* 创建一个columnFamily */
			ColumnPath col = new ColumnPath(columnFamily);
			col.setColumn(CassandraTest.toByteBuffer(ColumnName));
			System.out.println(tablename);
			System.out.println(ColumnName);
			System.out.println(cassandraClient.get(toByteBuffer(tablename), col, ConsistencyLevel.ONE));
			ColumnOrSuperColumn superColumn = cassandraClient.get(	CassandraTest.toByteBuffer(tablename),
																	col,
																	ConsistencyLevel.ONE);
			System.out.println(">>>>>>>>>>>>>>>>" + superColumn);
	
			Column column = cassandraClient.get(CassandraTest.toByteBuffer(tablename),
												col,
												ConsistencyLevel.ONE).column;
			return column;
		}catch(Exception e){
			e.printStackTrace();
			return null;
		}
	}

	/**
	 * 关闭当前的远程访问连接
	 */
	public static void close() {
		socket.close();
	}
	
	//转这Byte
    public static ByteBuffer toByteBuffer(String value)
    {
    	try{
    		return ByteBuffer.wrap(value.getBytes("UTF-8"));
    	}catch(Exception e){
    		e.printStackTrace();
    		return null;
    	}
    }
    //得到字符串
    public static String toString(ByteBuffer buffer)
    {
    	try{
	        byte[] bytes = new byte[buffer.remaining()];
	        buffer.get(bytes);
	        return new String(bytes, "UTF-8");
    	}catch(Exception e){
    		e.printStackTrace();
    		return null;
    	}
    }
}