Spring Reactive Cassandra是Spring Data Cassandra的响应式扩展,支持以非阻塞的方式操作Cassandra数据库,在需要处理高并发、低延迟的场景中应用广泛。当Cassandra表使用复合主键时,访问方式和单主键表存在一定差异,需要开发者掌握对应的实体定义与操作逻辑。

复合主键基础概念
Cassandra的复合主键由分区键和聚类键组成,分区键用于确定数据存储在哪个节点,聚类键用于确定同一分区内数据的排序规则。复合主键的定义格式如下:
CREATE TABLE user_action (
user_id text,
action_time timestamp,
action_type text,
action_content text,
PRIMARY KEY ((user_id), action_time)
);
上述表中,user_id是分区键,action_time是聚类键,查询时如果只指定分区键可以获取该分区下的所有数据,同时指定分区键和聚类键可以精准定位单条数据。
定义复合主键实体
在Spring Reactive Cassandra中,需要单独定义主键类,然后在实体类中引用该主键类。首先定义复合主键类:
import org.springframework.data.cassandra.core.mapping.PrimaryKeyClass;
import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;
import org.springframework.data.cassandra.core.cql.PrimaryKeyType;
import java.time.Instant;
@PrimaryKeyClass
public class UserActionPrimaryKey {
@PrimaryKeyColumn(name = "user_id", type = PrimaryKeyType.PARTITIONED)
private String userId;
@PrimaryKeyColumn(name = "action_time", type = PrimaryKeyType.CLUSTERED)
private Instant actionTime;
// 构造方法、getter、setter、equals、hashCode方法
public UserActionPrimaryKey() {
}
public UserActionPrimaryKey(String userId, Instant actionTime) {
this.userId = userId;
this.actionTime = actionTime;
}
public String getUserId() {
return userId;
}
public void setUserId(String userId) {
this.userId = userId;
}
public Instant getActionTime() {
return actionTime;
}
public void setActionTime(Instant actionTime) {
this.actionTime = actionTime;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
UserActionPrimaryKey that = (UserActionPrimaryKey) o;
return java.util.Objects.equals(userId, that.userId) && java.util.Objects.equals(actionTime, that.actionTime);
}
@Override
public int hashCode() {
return java.util.Objects.hash(userId, actionTime);
}
}
然后定义对应的实体类,使用@PrimaryKey注解引用主键类:
import org.springframework.data.cassandra.core.mapping.Table;
import org.springframework.data.cassandra.core.mapping.PrimaryKey;
@Table("user_action")
public class UserAction {
@PrimaryKey
private UserActionPrimaryKey key;
private String actionType;
private String actionContent;
// 构造方法、getter、setter
public UserAction() {
}
public UserAction(UserActionPrimaryKey key, String actionType, String actionContent) {
this.key = key;
this.actionType = actionType;
this.actionContent = actionContent;
}
public UserActionPrimaryKey getKey() {
return key;
}
public void setKey(UserActionPrimaryKey key) {
this.key = key;
}
public String getActionType() {
return actionType;
}
public void setActionType(String actionType) {
this.actionType = actionType;
}
public String getActionContent() {
return actionContent;
}
public void setActionContent(String actionContent) {
this.actionContent = actionContent;
}
}
使用ReactiveCassandraTemplate操作复合主键表
ReactiveCassandraTemplate提供了灵活的响应式操作方式,首先需要注入该模板:
import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Instant;
@Component
public class UserActionService {
@Autowired
private ReactiveCassandraTemplate cassandraTemplate;
// 插入数据
public Mono<UserAction> insertUserAction(String userId, Instant actionTime, String actionType, String actionContent) {
UserActionPrimaryKey key = new UserActionPrimaryKey(userId, actionTime);
UserAction userAction = new UserAction(key, actionType, actionContent);
return cassandraTemplate.insert(userAction);
}
// 根据完整主键查询单条数据
public Mono<UserAction> getUserActionByFullKey(String userId, Instant actionTime) {
UserActionPrimaryKey key = new UserActionPrimaryKey(userId, actionTime);
return cassandraTemplate.selectOneById(key, UserAction.class);
}
// 根据分区键查询该分区下所有数据
public Flux<UserAction> getUserActionsByPartitionKey(String userId) {
return cassandraTemplate.select(
"SELECT * FROM user_action WHERE user_id = ?",
UserAction.class,
userId
);
}
// 根据分区键和聚类键范围查询
public Flux<UserAction> getUserActionsByTimeRange(String userId, Instant startTime, Instant endTime) {
return cassandraTemplate.select(
"SELECT * FROM user_action WHERE user_id = ? AND action_time >= ? AND action_time <= ?",
UserAction.class,
userId, startTime, endTime
);
}
// 根据完整主键删除数据
public Mono<Boolean> deleteUserAction(String userId, Instant actionTime) {
UserActionPrimaryKey key = new UserActionPrimaryKey(userId, actionTime);
return cassandraTemplate.deleteById(key, UserAction.class);
}
}
使用ReactiveCassandraRepository操作复合主键表
除了使用模板类,还可以通过继承ReactiveCassandraRepository来简化操作,首先定义Repository接口:
import org.springframework.data.cassandra.repository.ReactiveCassandraRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
@Repository
public interface UserActionRepository extends ReactiveCassandraRepository<UserAction, UserActionPrimaryKey> {
// 根据分区键查询,方法名遵循Spring Data命名规范
Flux<UserAction> findByKeyUserId(String userId);
// 根据分区键和聚类键范围查询
Flux<UserAction> findByKeyUserIdAndKeyActionTimeBetween(String userId, Instant startTime, Instant endTime);
}
然后在业务类中注入Repository使用:
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Instant;
@Service
public class UserActionRepositoryService {
@Autowired
private UserActionRepository userActionRepository;
// 保存数据
public Mono<UserAction> saveUserAction(UserAction userAction) {
return userActionRepository.save(userAction);
}
// 根据分区键查询
public Flux<UserAction> getByUserId(String userId) {
return userActionRepository.findByKeyUserId(userId);
}
// 根据时间范围查询
public Flux<UserAction> getByTimeRange(String userId, Instant startTime, Instant endTime) {
return userActionRepository.findByKeyUserIdAndKeyActionTimeBetween(userId, startTime, endTime);
}
}
注意事项
- 复合主键类必须实现
equals和hashCode方法,否则在查询和删除操作时可能出现定位不准确的问题 - 查询时如果只使用聚类键作为条件,会导致全表扫描,性能较差,尽量保证查询条件包含分区键
- 响应式操作返回的是
Mono或Flux类型,需要合理处理背压和异常,避免数据流中断 - Cassandra的聚类键默认是升序排序,如果需要降序可以在建表时指定,或者在查询时使用
ORDER BY子句
Spring_Reactive_CassandraCassandra复合主键ReactiveCassandraTemplate@PrimaryKey响应式数据访问修改时间:2026-07-18 11:06:38