Cassandra：java客户端操作

java客户端介绍

Cassandra有众多的JAVA客户端，目前比较流程的都是不同公司开源的客户端，如：Netfix的astyanax，datastax的java-driver，hector，以及Spring Data for Apache Cassandra。

本文使用datastax的java-driver作为连接cassandra的客户端。

datastax的java-driver

介绍

是由DataStax公司，开源的用来操作Cassandra的工具包，官网：

https://docs.datastax.com/en/landing_page/doc/landing_page/current.html

在github的搜索页面点击 datastax/java-driver

源码地址：

https://github.com/datastax/java-driver

在页面上可以看到使用java-driver的简单介绍，包含Maven依赖内容，环境兼容要求。

可以看到需要jdk8或更版本，支持Cassandra2.1获取更高版本

入门使用

引入相关依赖：

        
        
            com.datastax.cassandra
            cassandra-driver-core
            3.9.0
        
        
            com.datastax.cassandra
            cassandra-driver-mapping
            3.9.0
        
        
        
            junit
            junit
            4.12
        

键空间操作

1.查询键空间

    Session session;

    @Before
    public void init() {
        String host = "192.168.1.43";
        int port = 9042;

        //连接服务端，获取会话
        Cluster cluster = Cluster.builder()
                .addContactPoint(() -> new InetSocketAddress(host, port))
                .build();
        session = cluster.connect();
    }

    /**
     * 查询键空间
     * @author wen.jie
     * @date 2022/3/16 15:07
     */
    @Test
    public void selectKeySpace() {
        List keyspaces =
                session.getCluster().getMetadata().getKeyspaces();
        for (KeyspaceMetadata keyspace : keyspaces) {
            System.out.println(keyspace);
        }
    }

2.创建键空间

/**
 * 创建键空间
 * @author wen.jie
 * @date 2022/3/16 15:08
 */
    @Test
    public void createKeySpace() {
        //执行cql语句
        //session.execute("CREATE KEYSPACE school_test WITH replication = {'class':'SimpleStrategy', 'replication_factor' : 3};");

        //面向对象方式
        HashMap map = new HashMap<>();
        map.put("class", "SimpleStrategy");
        map.put("replication_factor", 1);
        KeyspaceOptions opts = SchemaBuilder.createKeyspace("school_test")
                .ifNotExists()
                .with()
                .replication(map);
        session.execute(opts);
    }

效果：

3.删除键空间

    @Test
    public void dropKeySpace() {
        DropKeyspace dropKeyspace = SchemaBuilder.dropKeyspace("school_test").ifExists();
        session.execute(dropKeyspace);
    }

4.修改键空间

    @Test
    public void alterKeySpace() {
        HashMap map = new HashMap<>();
        map.put("class", "SimpleStrategy");
        map.put("replication_factor", 2);
        KeyspaceOptions options = SchemaBuilder.alterKeyspace("school")
                .with().replication(map);
        session.execute(options);
    }

表操作

1.创建表

    @Test
    public void createTable() {
        Create create = SchemaBuilder.createTable("school", "students")
                .addPartitionKey("id", DataType.bigint())
                .addColumn("address", DataType.varchar())
                .addColumn("name", DataType.varchar())
                .addColumn("phone", DataType.CollectionType.list(DataType.varchar()))
                .addColumn("interest", DataType.CollectionType.set(DataType.varchar()))
                .addColumn("education", DataType.CollectionType.map(DataType.varchar(), DataType.varchar()))
                .ifNotExists();
        session.execute(create);
    }

2.修改表

    @Test
    public void alterTable() {
        SchemaStatement statement = SchemaBuilder.alterTable("school", "students")
                .addColumn("email").type(DataType.text());
        session.execute(statement);

        SchemaStatement statement1 = SchemaBuilder.alterTable("school", "students")
                .dropColumn("address");
        session.execute(statement1);
    }

3.删除表

    @Test
    public void dropTable() {
        Drop drop = SchemaBuilder.dropTable("school", "students");
        session.execute(drop);
    }

数据操作

使用Mapper和Bean对象形式保存数据:

@Data
@Table(keyspace = "school", name = "student")
public class Student implements Serializable {
    @PartitionKey //标识当前字段是分区键
    private Integer id;
    private String address;
    private Integer age;
    private Map education;
    private Byte gender;
    private Set interest;
    private String name;
    private List phone;
}

测试插入数据

    /**
     * 使用Mapper和Bean对象形式保存数据
     * @author wen.jie
     * @date 2022/3/16 15:51
     */
    @Test
    public void insert() {
        Student student = new Student();
        student.setId(1012);
        student.setAddress("南京");
        student.setAge(17);
        student.setEducation(Collections.singletonMap("大学", "abc大学"));
        student.setGender((byte)1);
        student.setInterest(new HashSet<>(Arrays.asList("看书", "音乐")));
        student.setName("tom");
        student.setPhone(Arrays.asList("12222222222","13333333333"));
        Mapper mapper = new MappingManager(session).mapper(Student.class);
        mapper.save(student);
    }

测试查询全部

    @Test
    public void findAll() {
        ResultSet resultSet = session.execute(QueryBuilder.select().from("school", "student"));
        Mapper mapper = new MappingManager(session).mapper(Student.class);
        List students = mapper.map(resultSet).all();
        for (Student student : students) {
            System.out.println(student);
        }
    }

根据id查询

    @Test
    public void findById() {
        Select.Where where = QueryBuilder.select().from("school", "student").where(QueryBuilder.eq("id", 1012));
        ResultSet resultSet = session.execute(where);
        Mapper mapper = new MappingManager(session).mapper(Student.class);
        Student student = mapper.map(resultSet).one();
        System.out.println(student);
    }

根据id删除

    @Test
    public void deleteById() {
        Mapper mapper = new MappingManager(session).mapper(Student.class);
        mapper.delete(1012);
    }

索引操作

创建索引

    @Test
    public void createIdx() {
        SchemaStatement statement = SchemaBuilder.createIndex("name_idx")
                .onTable("school", "student")
                .andColumn("name");
        session.execute(statement);
    }

删除索引

    @Test
    public void dropIndex() {
        Drop drop = SchemaBuilder.dropIndex("school", "name_idx");
        session.execute(drop);
    }

预编译操作

cassandra提供了类似jdbc preparedstatement使用预编译占位符。官方文档链接如下：

https://docs.datastax.com/en/developer/java-driver/3.0/manual/statements/prepared/

基本原理:

预编译statement的时候，Cassandra会解析query语句，缓存解析的结果并返回一个唯一的标志。当绑定并且执行预编译statement的时候，驱动只会发送这个标志，那么Cassandra就会跳过解析query语句的过程。

应保证query语句只应该被预编译一次，缓存PreparedStatement 到我们的应用中（PreparedStatement 是线程安全的）；如果我们对同一个query语句预编译了多次，那么驱动输出印警告日志；

如果一个query语句只执行一次，那么预编译不会提供性能上的提高，反而会降低性能，因为是两次请求，那么此时可以考虑用 simple statement 来代替

代码：

    @Test
    public void batchPrepare(){
        //先把语句预编译
        BatchStatement batch = new BatchStatement();
        PreparedStatement ps = session .prepare("INSERT INTO school.student (id,address,age,gender,name,interest, phone,education) VALUES (?,?,?,?,?,?,?,?)");
        //循环10次，构造不同的student对象
        for (int i = 0; i < 10; i++) {
            HashMap education = new HashMap<>();
            education.put("小学", "中心第"+i+"小学");
            education.put("中学", "第"+i+"中学");
            HashSet interest = new HashSet<>();
            interest.add("看书");
            interest.add("电影");
            List phones = new ArrayList<>();
            phones.add("0"+i+"0-66666666");
            phones.add("1"+i+"666666666");
            Student student = new Student(
                    1013 + i,
                    "北京市朝阳区10" + i + "号",
                    21 + i,
                    education,
                    (byte)1,
                    interest,
                    "学生"+i,
                    phones);
            BoundStatement bs = ps.bind(student.getId(),
                    student.getAddress(),
                    student.getAge(),
                    student.getGender(),
                    student.getName(),
                    student.getInterest(),
                    student.getPhone(),
                    student.getEducation());
            batch.add(bs);
        }
        session.execute(batch);
        batch.clear();
    }

效果：