cassandra新增、更新、删除数据。

来源:转载


package client;import java.io.UnsupportedEncodingException;import java.nio.ByteBuffer;import java.util.List;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.SlicePredicate;import org.apache.cassandra.thrift.SliceRange;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.protocol.TProtocol;import org.apache.thrift.transport.TFramedTransport;import org.apache.thrift.transport.TSocket;import org.apache.thrift.transport.TTransport;public class CClient{ public static void main(String[] args) throws TException, InvalidRequestException, UnavailableException, UnsupportedEncodingException, NotFoundException, TimedOutException { //建立连接 TTransport tr = new TFramedTransport(new TSocket("127.0.0.1", 9160)); TProtocol proto = new TBinaryProtocol(tr); Cassandra.Client client = new Cassandra.Client(proto); tr.open(); String key_user_id = "1"; // insert data long timestamp = System.currentTimeMillis(); //相当于DB NAME client.set_keyspace("DEMO"); //相当于DB Table ColumnParent parent = new ColumnParent("Users"); //字段名 Column nameColumn = new Column(toByteBuffer("name")); //字段值 nameColumn.setValue(toByteBuffer("scott")); //插入时间 nameColumn.setTimestamp(timestamp); //将数据添加到cassandra client.insert(toByteBuffer(key_user_id), parent, nameColumn, ConsistencyLevel.ONE); //字段名 Column ageColumn = new Column(toByteBuffer("password")); //字段值 ageColumn.setValue(toByteBuffer("tiger")); //插入时间 ageColumn.setTimestamp(timestamp); //将数据添加到cassandra client.insert(toByteBuffer(key_user_id), parent, ageColumn, ConsistencyLevel.ONE); //得到相当于DB Table ColumnPath path = new ColumnPath("Users"); // read single column path.setColumn(toByteBuffer("name")); System.out.println(client.get(toByteBuffer(key_user_id), path, ConsistencyLevel.ONE)); // read entire row SlicePredicate predicate = new SlicePredicate(); SliceRange sliceRange = new SliceRange(toByteBuffer(""), toByteBuffer(""), false, 10); predicate.setSlice_range(sliceRange); List<ColumnOrSuperColumn> results = client.get_slice(toByteBuffer(key_user_id), parent, predicate, ConsistencyLevel.ONE); for (ColumnOrSuperColumn result : results) { Column column = result.column; System.out.println(toString(column.name) + " -&gt; " + toString(column.value)); } tr.close(); } public static ByteBuffer toByteBuffer(String value) throws UnsupportedEncodingException { return ByteBuffer.wrap(value.getBytes("UTF-8")); } public static String toString(ByteBuffer buffer) throws UnsupportedEncodingException { byte[] bytes = new byte[buffer.remaining()]; buffer.get(bytes); return new String(bytes, "UTF-8"); }}

 

package result;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 = "127.0.0.1"; 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 = "DEMO"; /* 初始化连接 */ init(keyspace); /* 创建一个表名*/ String columnFamily = "Users"; //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; } }}

 

/** * */package result;import java.util.HashMap;import java.util.Map;import me.prettyprint.cassandra.model.AllOneConsistencyLevelPolicy;import me.prettyprint.cassandra.serializers.StringSerializer;import me.prettyprint.cassandra.service.FailoverPolicy;import me.prettyprint.cassandra.service.template.ColumnFamilyResult;import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;import me.prettyprint.cassandra.service.template.ColumnFamilyUpdater;import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;import me.prettyprint.hector.api.Cluster;import me.prettyprint.hector.api.Keyspace;import me.prettyprint.hector.api.exceptions.HectorException;import me.prettyprint.hector.api.factory.HFactory;public class GetResult { /** * @param args */ @SuppressWarnings("unchecked") public static void main(String[] args) { String keySpace = "DEMO";//相当于DB NAME String columnFamily = "Users";//相当于DB Table Cluster cluster = HFactory.getOrCreateCluster("Test Cluster", "127.0.0.1:9160"); Map accessMap = new HashMap(); accessMap.put("username", "wyq"); accessMap.put("password", "123456"); Keyspace ksp = HFactory.createKeyspace(keySpace, cluster, new AllOneConsistencyLevelPolicy(), FailoverPolicy.ON_FAIL_TRY_ALL_AVAILABLE, accessMap); ColumnFamilyTemplate<String, String> template = new ThriftColumnFamilyTemplate<String, String>( ksp, columnFamily, StringSerializer.get(), StringSerializer .get()); ColumnFamilyUpdater<String, String> updater = template.createUpdater("u_1"); // 以下name,email,time相当于字段 updater.setString("name", "wyqa"); //updater.setString("email", "[email protected]"); updater.setString("password", "123456"); //updater.setLong("time", System.currentTimeMillis()); try { template.update(updater); System.out.println("update ok."); } catch (HectorException e) { e.printStackTrace(); } try { ColumnFamilyResult<String, String> res = template .queryColumns("u_1"); ColumnFamilyResult<String, String> rest = template .queryColumns("1"); String name = res.getString("name"); String email = res.getString("email"); //long time = res.getLong("time"); System.out.println("read u_1 name:" + name); System.out.println("read u_1 email:" + email); //System.out.println("read u_1 time:" + time); System.out.println("age:" + rest.getString("age")); System.out.println("name:" + rest.getString("name")); } catch (HectorException e) { e.printStackTrace(); } }}

 



分享给朋友:
您可能感兴趣的文章:
随机阅读: