SpringBoot中使用Cassandra分布式数据库

上一篇介绍了Cassandra分布式数据库的基本操作和分布式原理，这篇文章将介绍怎样在spring boot中使用Cassandra。

Cassandra jpa 依赖：

在SpringBoot中使用Cassandra，需要以下依赖：

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

实体定义

@Table
public class Person {

  @PrimaryKey
  private final String id;
  private final String name;
  private final int age;
  ...
}

使用 Spring Data Repositories

自定义的repository需要继承以下任一一个 repository：

public interface CrudRepository<T, ID extends Serializable> extends Repository<T, ID> {

  <S extends T> S save(S entity);      
  Optional<T> findById(ID primaryKey); 
  Iterable<T> findAll();           
  long count();                    
  void delete(T entity);               
  boolean existsById(ID primaryKey);   
}

public interface PagingAndSortingRepository<T, ID extends Serializable> extends CrudRepository<T, ID> {
  Iterable<T> findAll(Sort sort);
  Page<T> findAll(Pageable pageable);
}

public interface Repository<T, ID> {
}

具体示例如下：

interface UserRepository extends CrudRepository<User, Long> {

  long deleteByLastname(String lastname);

  List<User> removeByLastname(String lastname);

  long countByLastname(String lastname);
}

除了使用 Spring Data Repositories之外，还可以使用CassandraTemplate或者使用CqlTemplate，具体如下：

使用CassandraTemplate

CassandraOperations接口定义了操作Cassandra数据库的基本方法，而CassandraTemplate为它的一个实现类，所以，就可以注入CassandraTemplate来操作Cassandra数据库。示例如下：

/**
 * @author yawn
 * 2019/8/26 16:49
 */
@Repository
public class CassandraRepository<T extends CassandraEntity> {

    @Autowired
    protected CassandraTemplate cassandraTemplate;

    public boolean save(T entity) {
        List<Object> valueList = entity.valueList();
        List<String> columnList = new ArrayList<>();
        if (entity.lowerCase()) {
            for (String column : entity.columnList()) {
                columnList.add(column.toLowerCase());
            }
        } else {
            columnList = entity.columnList();
        }
        Insert insert = QueryBuilder.insertInto(entity.tableName()).values(columnList, valueList);
        cassandraTemplate.execute(insert);
        return true;
    }

    public List<T> listBy(Map<String, Object> params, Class<T> tClass) {
        Select select = QueryBuilder.select().from(tClass.getSimpleName().toLowerCase());
        params.forEach((key, value) ->
            select.where(QueryBuilder.eq(key.toLowerCase(), value))
        );
        return cassandraTemplate.select(select, tClass);
    }
}

更多操作如下：

import static org.springframework.data.cassandra.core.query.Criteria.where;
import static org.springframework.data.cassandra.core.query.Query.query;
…

Person bob = new Person("Bob", 33);
cassandraTemplate.insert(bob);

Person queriedBob = cassandraTemplate.selectOneById(query(where("age").is(33)), Person.class);

import static org.springframework.data.cassandra.core.query.Criteria.where;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.cassandra.core.query.Update;

...

boolean applied = cassandraTemplate.update(Query.query(where("id").is("foo")),
  Update.create().increment("balance", 50.00), Account.class);

import static org.springframework.data.cassandra.core.query.Criteria.where;
import static org.springframework.data.cassandra.core.query.Query.query;

…

List<Person> result = cassandraTemplate.select(query(where("age").is(50))
  .and(where("balance").gt(1000.00d)).withAllowFiltering(), Person.class);

使用CqlTemplate

CqlTemplate类似于jdbctemplate，可以直接执行Cql语句，使用示例如下：

// 统计
int rowCount = cqlTemplate.queryForObject("SELECT COUNT(*) FROM t_actor", Integer.class);
// 参数传递
int countOfActorsNamedJoe = cqlTemplate.queryForObject(
		"SELECT COUNT(*) FROM t_actor WHERE first_name = ?", Integer.class, "Joe");
// 插入数据
cqlTemplate.execute(
		"INSERT INTO t_actor (first_name, last_name) VALUES (?, ?)",
		"Leonor", "Watling");
// 创建表
cqlOperations.execute("CREATE TABLE test_table (id uuid primary key, event text)");
// 删除表
DropTableSpecification dropper = DropTableSpecification.dropTable("test_table");
String cql = DropTableCqlGenerator.toCql(dropper);
cqlTemplate.execute(cql);

// 查询
Actor actor = cqlTemplate.queryForObject(
		"SELECT first_name, last_name FROM t_actor WHERE id = ?",
		new RowMapper<Actor>() {
			public Actor mapRow(Row row, int rowNum) {
				Actor actor = new Actor();
				actor.setFirstName(row.getString("first_name"));
				actor.setLastName(row.getString("last_name"));
				return actor;
			},
      new Object[]{1212L},
		});

// 查询
public List<Actor> findAllActors() {
	return cqlTemplate.query("SELECT first_name, last_name FROM t_actor", ActorMapper.INSTANCE);
}
// 可以这样定义RowMapper
enum ActorMapper implements RowMapper<Actor> {

    INSTANCE;

	public Actor mapRow(Row row, int rowNum) {
		Actor actor = new Actor();
		actor.setFirstName(row.getString("first_name"));
		actor.setLastName(row.getString("last_name"));
		return actor;
	}
}

在springboot中，可以通过以下3中方式操作Cassandra数据库：

1. 使用 Spring Data Repositories
2. 使用 CassandraTemplate
3. 使用 CqlTemplate
4. 使用 ReactiveCassandraTemplate