書接上文 微服務(wù)實戰(zhàn)之 Cassandra 之一， Cassandra 的優(yōu)點說了不少霞捡，它的缺點大家也有所耳聞坐漏。

作為一個 NoSQL 存儲系統(tǒng)，它不支持多表連接碧信，不支持外鍵約束赊琳，所以也不需要遵循數(shù)據(jù)庫的經(jīng)典范式。

雖然 CQL 看起來很象 SQL砰碴，其實還是有點差別的躏筏，它的 Insert/Update 其實都是 Set, 它也不支持行級鎖，不支持事務(wù)（它也有一種輕量級的事務(wù)呈枉，但由于它的分布系統(tǒng)特點趁尼，與傳統(tǒng)的事務(wù)大相迥異）。

它的客戶端驅(qū)動是很豐富的猖辫， 下面我們以 Java 舉一個實際應(yīng)用實例:

創(chuàng)建 Keyspace 和 Tables

• 創(chuàng)建一個 key space: walter_apjc
  采用網(wǎng)絡(luò)拓撲策略酥泞，兩個DC: HF1 和 SZ1

CREATE KEYSPACE walter_apjc WITH replication = 
{'class': 'NetworkTopologyStrategy', 'HF1': '3', 'SZ1': '3'}  AND durable_writes = true; 

• 創(chuàng)建兩張表 inventory 和 person

CREATE TABLE inventory ( user_id uuid, 
inventory_id uuid, 
inventory_name text, 
name text, tags text, 
create_time timestamp, 
last_modified_time timestamp, 
PRIMARY KEY(user_id,inventory_id)); 

CREATE TABLE person (
id text PRIMARY KEY,
name text,
age int);

• 創(chuàng)建一個索引

create index on inventory(name);

使用 Cassandra 的 Java Driver 來存取數(shù)據(jù)

創(chuàng)建一個基礎(chǔ)類CassandraClient 來連接 Cassandra Cluster, 這里對于 Cluster 有很多關(guān)鍵屬性需要設(shè)置，在 Cassandra Java Driver 4.x 版本中包裝成 CqlSession

package com.github.walterfan.hellocassandra;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.policies.*;

import lombok.Builder;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;

@Data
@Builder
public class CassandraClient {

    private String contactPoints;
    private int port;
    private String localDC;
    private String username;
    private String password;
    private int maxConnectionsPerHost ;
    private int usedHostsPerRemote ;
    private long reconnectBaseDelayMs ;
    private long reconnectMaxDelayMs ;

    private volatile Cluster cluster;

    public synchronized void init() {

            DCAwareRoundRobinPolicy loadBanalcePolicy = DCAwareRoundRobinPolicy.builder()
                    .withLocalDc(localDC)
                    .withUsedHostsPerRemoteDc(usedHostsPerRemote)
                    .allowRemoteDCsForLocalConsistencyLevel()
                    .build();

            PoolingOptions poolingOptions =new PoolingOptions();
            poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, maxConnectionsPerHost);
            poolingOptions.setMaxConnectionsPerHost(HostDistance.REMOTE, maxConnectionsPerHost);

            Cluster.Builder clusterBuilder = Cluster.builder()
                    .withReconnectionPolicy(new ExponentialReconnectionPolicy(reconnectBaseDelayMs,reconnectMaxDelayMs))
                    .withRetryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE))
                    .withPoolingOptions(poolingOptions)
                    .withLoadBalancingPolicy(new TokenAwarePolicy(loadBanalcePolicy))
                    .withPort(port)
                    .addContactPoints(contactPoints.split(","))
                    .withoutJMXReporting();

            if(StringUtils.isNotEmpty(username) && StringUtils.isNotEmpty(password) ) {
                clusterBuilder.withCredentials(username, password);
            }

            cluster = clusterBuilder.build();


    }


    public Session connect(String keyspace) {
        if(null == cluster) {
            init();
        }
        return this.cluster.connect(keyspace);
    }

    public void close() {
        cluster.close();
    }
}

利用 Spring 提供的 CassandraTemplate 可以很方便來存取數(shù)據(jù)

在讀寫數(shù)據(jù)時啃憎，有幾個很重要的選項需要設(shè)置

• ConsistentLevel
• RetryPolicy
• FetchSize
• ReadTimeout

一致性水平是 Cassandra 所特有的芝囤，多數(shù)情況下我們會選擇 LOCAL_QUORUM, 也就是在本地的 DC 要符合 Quorum 法定節(jié)點數(shù)有成功響應(yīng)才可以。

代碼示例如下：

package com.github.walterfan.hellocassandra;

import java.net.InetSocketAddress;

import java.time.Instant;
import java.util.Collections;
import java.util.List;

import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.*;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.lang3.StringUtils;

import org.springframework.cassandra.core.RowMapper;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.support.exception.CassandraTypeMismatchException;
import org.springframework.data.cassandra.core.CassandraTemplate;

import org.springframework.data.cassandra.mapping.Table;


import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;

import static org.assertj.core.api.Assertions.assertThat;

/**
 * Created by yafan on 15/11/2017.
 */

@Slf4j
@Setter
public class CassandraTemplateExample {

    static final String AGE_COLUMN_NAME = "age";
    static final String ID_COLUMN_NAME = "id";
    static final String NAME_COLUMN_NAME = "name";
    private String keyspace;
    private CassandraTemplate template;
    private CassandraClient client;
    private QueryOptions queryOptions;
    private WriteOptions writeOptions;


    public CassandraTemplateExample(String hostnames, int port, String localDC, String username, String password, String keysapce) {
        client = CassandraClient.builder()
                .contactPoints(hostnames)
                .port(port)
                .localDC(localDC)
                .username(username)
                .password(password)
                .maxConnectionsPerHost(2048)
                .reconnectBaseDelayMs(1000)
                .reconnectMaxDelayMs(600_000)
                .build();
        this.keyspace = keysapce;

        queryOptions = this.getQueryOptions();
        writeOptions = this.getWriteOptions(null);

    }

    protected QueryOptions getQueryOptions() {
        QueryOptions queryOptions = new QueryOptions();
        queryOptions.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
        return queryOptions;
    }

    protected WriteOptions getWriteOptions(Integer ttl) {
        WriteOptions writeOptions = new WriteOptions();
        writeOptions.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
        writeOptions.setTtl(ttl);
        return writeOptions;
    }

    private void testCql() {
        try(Session session = client.connect(keyspace)) {

            template = new CassandraTemplate(session);
            testCrud();
            testTransaction();
            testPagination();
        }


    }

    private void execute(String cql) {

        log.info("execute {}" , cql);
        if(cql.startsWith("select")) {
            List<?> aList = template.select(cql, List.class);
            aList.forEach(System.out::println);
        } else {
            template.execute(cql);
        }


    }

    private void testCrud() {
        log.info("--------- testCrud ----------------");
        Person thePerson = template.insert(Person.create("Walter Fan", 37));


        log.info("Inserted [{}]", thePerson);

        Person queriedPerson = queryPersonById(thePerson.getId());
        assertThat(queriedPerson).isNotSameAs(thePerson);
        assertThat(queriedPerson).isEqualTo(thePerson);
    }


    private Person queryPersonById(String id) {
        Select personQuery = selectPerson(id);


        log.info("CQL SELECT [{}]", personQuery);

        Person queriedPerson = template.queryForObject(personQuery, personRowMapper());

        log.info("* Query Result [{}]", queriedPerson);

        return queriedPerson;
    }

    private void testTransaction() {
        System.out.println("--------- testTransaction ----------------");
        template.execute("insert into person(id, name, age) values ('a', 'alice', 20)");
        template.execute("update person set name = 'ada' where id='a' ");
        template.execute("update person set name = 'adam' where id='a' IF name ='alice'");

        template.execute("insert into person(id, name, age) values ('b', 'bob', 21)");
        template.execute("insert into person(id, name, age) values ('c', 'carl', 22)");

        template.execute("insert into person(id, name, age) values ('d', 'david', 31)");
        template.execute("insert into person(id, name, age) values ('d', 'dean', 32) ");
        template.execute("insert into person(id, name, age) values ('d', 'dog', 33)  IF NOT EXISTS");


        Person a = queryPersonById( "a");
        Person d = queryPersonById( "d");

        assertThat(a.getName().equals("ada"));
        assertThat(a.getAge() == 20);
        assertThat(a.getName().equals("dean"));

        template.execute("delete from person where id in ('a','b','c','d')");


    }


    private void testPagination() {
        System.out.println("--------- testPagination ----------------");
        String user_id = "e7d6038e-7a07-4dca-a98f-939428ded582";

        String[] inventoryIDs = {"01786fd5-92ef-491c-b64e-7d83a624b95d",
                "12a7cd81-d384-44d4-8919-5fa092f6427f",
                "1cd52315-fba7-4461-99b3-8e44b7b589e8",
                "2a7a3b39-f080-40c1-b5c8-08b743d66076",
                "2b2b458e-75e4-4709-9853-8491cfeb13e9"};

        int i = 0;
        for(String inventory_id: inventoryIDs) {

            String cql = String.format("insert into inventory(user_id, inventory_id, inventory_name, name, tags, create_time, last_modified_time) " +
                            "values (%s, %s, '%s','%s', '%s', '%s', '%s')",
                    user_id, inventory_id, "book", "posa" + (++i), "tech", Instant.now().toString(), Instant.now().toString());
            log.info("execute {}", cql);
            template.execute(cql);

        }


        List<Inventory> inventories0 = template.select(String.format("select * from inventory WHERE user_id=%s", user_id), Inventory.class);

        List<Inventory> inventories2 = template.select(String.format("select * from inventory WHERE user_id=%s and inventory_id > %s ALLOW FILTERING", user_id, inventoryIDs[2]), Inventory.class);


        System.out.println("---------all inventories----------------");
        inventories0.forEach(System.out::println);

        System.out.println("---------filterd inventories----------------");

        inventories2.forEach(System.out::println);

    }

    public InetSocketAddress newSocketAddress(String hostname, int port) {
        return new InetSocketAddress(hostname, port);
    }



    public void close() {
        this.client.close();
    }

    protected static RowMapper<Person> personRowMapper() {
        return new RowMapper<Person>() {
            public Person mapRow(Row row, int rowNum) throws DriverException {
                try {

                    log.debug("row [{}] @ index [{}]", row, rowNum);


                    Person person = Person.create(row.getString(ID_COLUMN_NAME),
                            row.getString(NAME_COLUMN_NAME), row.getInt(AGE_COLUMN_NAME));


                    log.debug("person [{}]", person);


                    return person;
                }
                catch (Exception e) {
                    throw new CassandraTypeMismatchException(String.format(
                            "failed to map row [%1$] @ index [%2$d] to object of type [%3$s]",
                            row, rowNum, Person.class.getName()), e);
                }
            }
        };
    }

    protected static Select selectPerson(String personId) {
        Select selectStatement = QueryBuilder.select().from(toTableName(Person.class));
        selectStatement.where(QueryBuilder.eq(ID_COLUMN_NAME, personId));
        return selectStatement;
    }

    @SuppressWarnings("unused")
    protected static String toTableName(Object obj) {
        return toTableName(obj.getClass());
    }

    protected static String toTableName(Class<?> type) {
        Table tableAnnotation = type.getAnnotation(Table.class);

        return (tableAnnotation != null && StringUtils.isNotEmpty(tableAnnotation.value())
                ? tableAnnotation.value() : type.getSimpleName());
    }

    public static void main(String[] args) throws Exception {

        CassandraTemplateExample exam = new CassandraTemplateExample("10.224.38.139", 9042, "HF1","test", "pass","walter_apjc");
        exam.testCql();
        exam.close();
    }

}

Spring Data Cassadra 項目

Spring Data 項目為 Cassandra 也創(chuàng)建了一個子項目辛萍，我們可以用它來大大簡化我們的代碼悯姊。

1） 先加入相應(yīng)的 dependency

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.github.walterfan</groupId>
    <artifactId>hellocassandra</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>hellocassandra</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.8.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-cassandra</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.assertj</groupId>
            <artifactId>assertj-core</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.5</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>


</project>

再分別為這兩張表創(chuàng)建相應(yīng)的實體對象

• 清單類 Inventory

package com.github.walterfan.hellocassandra;


import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.cassandra.core.PrimaryKeyType;
import org.springframework.data.cassandra.mapping.Column;

import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
import org.springframework.data.cassandra.mapping.Table;

import java.time.Instant;
import java.util.UUID;

@Data
@NoArgsConstructor
@Table
public class Inventory {
    @PrimaryKeyColumn(name = "user_id", ordinal = 0, type = PrimaryKeyType.PARTITIONED)
    private UUID userId;

    @PrimaryKeyColumn(name = "inventory_id", ordinal = 1, type = PrimaryKeyType.PARTITIONED)
    private UUID inventoryId;

    @Column("inventory_name")
    private String inventoryName;

    @Column("name")
    private String name;

    @Column("tags")
    private String tags;

    @Column("create_time")
    private Instant createTime;

    @Column("last_modified_time")
    private Instant lastmodifiedTime;

    public Inventory(UUID userId, UUID inventoryId, String inventoryName, String name, String tags) {
        this.userId = userId;
        this.inventoryId = inventoryId;
        this.inventoryName = inventoryName;
        this.name = name;
        this.tags = tags;
        this.createTime = Instant.now();
        this.lastmodifiedTime = Instant.now();
    }
}

• 人員類 Person

package com.github.walterfan.hellocassandra;


import java.util.UUID;

import lombok.Data;
import org.springframework.data.annotation.PersistenceConstructor;
import org.springframework.data.cassandra.mapping.PrimaryKey;
import org.springframework.data.cassandra.mapping.Table;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@Data
@Table("person")
public class Person {

    @PrimaryKey
    private final String id;

    private final String name;

    private int age;

    public static Person create(String name, int age) {
        return create(UUID.randomUUID().toString(), name, age);
    }

    public static Person create(String id, String name, int age) {
        return new Person(id, name, age);
    }

    @PersistenceConstructor
    public Person(String id, String name, int age) {
        Assert.hasText(id, "'id' must be set");
        Assert.hasText(name, "'name' must be set");

        this.id = id;
        this.name = name;
        this.age = validateAge(age);
    }

    private int validateAge(int age) {
        Assert.isTrue(age > 0, "age must be greater than 0");
        return age;
    }


}

就象大多數(shù) Spring Data JPA 項目一樣，它的存取倉庫類也是聲明式贩毕，非常簡單

package com.github.walterfan.hellocassandra;


import org.springframework.data.cassandra.repository.Query;
import org.springframework.data.repository.CrudRepository;
import org.springframework.stereotype.Repository;

import java.util.List;
import java.util.UUID;


@Repository
public interface InventoryRepository extends CrudRepository<Inventory, String> {

    @Query(value="SELECT * FROM inventory WHERE name=?0")
    public List<Inventory> findByName(String name);

    @Query("SELECT * FROM inventory WHERE tags = ?0")
    public List<Inventory> findByTags(String tags);

    @Query("SELECT * FROM inventory WHERE user_id = ?0")
    public List<Inventory> findByUserId(UUID userId);


    public void deleteAllByUserId(UUID userId);


    @Query("SELECT * FROM inventory WHERE user_id = ?0 and inventory_id= ?1 ")
    public List<Inventory> findByUserAndInventoryId(UUID userId, UUID inventory_id);


}

通過 Spring Boot Command Line Application 來簡單演示一下

package com.github.walterfan.hellocassandra;


import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.List;
import java.util.UUID;

/**
 * Created by yafan on 14/11/2017.
 */

@SpringBootApplication
public class HelloCassandra implements CommandLineRunner {

    private UUID userId = UUID.randomUUID();


    @Autowired
    InventoryRepository inventoryRepository;

    @Override
    public void run(String... args) throws Exception {
        clearData();
        saveData();
        lookup();
    }

    public void clearData(){
        inventoryRepository.deleteAllByUserId(userId);
    }

    public void saveData(){

        System.out.println("===================Save Customers to Cassandra===================");
        Inventory inventory_1 = new Inventory(userId, UUID.randomUUID(), "book", "Web Scalability", "tech");
        Inventory inventory_2 = new Inventory(userId, UUID.randomUUID(), "book","Ansible", "tech");
        Inventory inventory_3 = new Inventory(userId, UUID.randomUUID(), "book", "Go in action", "tech");
        Inventory inventory_4 = new Inventory(userId, UUID.randomUUID(),"task","Write diary", "work");
        Inventory inventory_5 = new Inventory(userId, UUID.randomUUID(),"task","Write book", "work");
        Inventory inventory_6 = new Inventory(userId, UUID.randomUUID(),"task","Write reading notes", "work");

        // save customers to ElasticSearch
        inventoryRepository.save(inventory_1);
        inventoryRepository.save(inventory_2);
        inventoryRepository.save(inventory_3);
        inventoryRepository.save(inventory_4);
        inventoryRepository.save(inventory_5);
        inventoryRepository.save(inventory_6);
    }

    public void lookup(){

        System.out.println("===================Lookup Inventory from Cassandra by userId===================");
        List<Inventory> list1 = inventoryRepository.findByUserId(userId);
        list1.forEach(System.out::println);

        System.out.println("===================Lookup Inventory from Cassandra by name ===================");
        List<Inventory> list2 = inventoryRepository.findByName("Ansible");
        list2.forEach(System.out::println);


    }

    public static void main(String[] args) {
        SpringApplication.run(HelloCassandra.class, args).close();
    }
}

大家可以把源代碼下載下來運行并查看結(jié)果悯许，完整代碼參見 https://github.com/walterfan/helloworld/tree/master/hellocassandra

參考資料

• Cassandra java driver 參考文檔
• Cassandra java driver 源代碼
• spring data cassandra 參考文檔