具有响应式Cassandra 的Spring Data
Posted SpringForAll社区
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了具有响应式Cassandra 的Spring Data相关的知识，希望对你有一定的参考价值。
原文链接：https://www.baeldung.com/spring-data-cassandra-reactive
译者：xuli
1.简介
在本教程中，我们将学习如何使用Spring Data Cassandra的响应式数据访问功能。
特别地，这是Spring Data Cassandra文章系列的第三篇文章。在本文中，我们将使用REST API暴露Cassandra数据库。
我们可以在本系列的第一篇和第二篇文章中阅读有关Spring Data Cassandra的更多信息 。
2.Maven依赖
事实上，Spring Data Cassandra支持Project Reactor和RxJava反应类型。为了演示，我们将在本教程中使用Project reactor的反应类型Flux和Mono 。
首先，让我们添加教程所需的依赖项：
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-cassandra</artifactId>
<version>2.1.2.RELEASE</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
最新版本的 spring-data-cassandra 可以在这里找到。
现在，我们将通过REST API从数据库中暴露SELECT操作。所以，我们也为RestController添加依赖项：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
3.实施我们的应用程序
由于我们将持久化数据，让我们首先定义我们的实体对象：
@Table
public class Employee {
@PrimaryKey
private int id;
private String name;
private String address;
private String email;
private int age;
}
接下来，是时候创建一个从ReactiveCassandraRepository扩展的EmployeeRepository 。重要的是要注意此接口支持响应类型：
public interface EmployeeRepository extends ReactiveCassandraRepository<Employee, Integer> {
@AllowFiltering
Flux<Employee> findByAgeGreaterThan(int age);
}
3.1.用于CRUD操作的Rest控制器
为了便于说明，我们将使用一个简单的Rest Controller 公开一些基本的 SELECT操作：
@RestController
@RequestMapping("employee")
public class EmployeeController {
@Autowired
EmployeeService employeeService;
@PostConstruct
public void saveEmployees() {
List<Employee> employees = new ArrayList<>();
employees.add(new Employee(123, "John Doe", "Delaware", "jdoe@xyz.com", 31));
employees.add(new Employee(324, "Adam Smith", "North Carolina", "asmith@xyz.com", 43));
employees.add(new Employee(355, "Kevin Dunner", "Virginia", "kdunner@xyz.com", 24));
employees.add(new Employee(643, "Mike Lauren", "New York", "mlauren@xyz.com", 41));
employeeService.initializeEmployees(employees);
}
@GetMapping("/list")
public Flux<Employee> getAllEmployees() {
Flux<Employee> employees = employeeService.getAllEmployees();
return employees;
}
@GetMapping("/{id}")
public Mono<Employee> getEmployeeById(@PathVariable int id) {
return employeeService.getEmployeeById(id);
}
@GetMapping("/filterByAge/{age}")
public Flux<Employee> getEmployeesFilterByAge(@PathVariable int age) {
return employeeService.getEmployeesFilterByAge(age);
}
}
最后，让我们添加一个简单的EmployeeService：
@Service
public class EmployeeService {
@Autowired
EmployeeRepository employeeRepository;
public void initializeEmployees(List<Employee> employees) {
Flux<Employee> savedEmployees = employeeRepository.saveAll(employees);
savedEmployees.subscribe();
}
public Flux<Employee> getAllEmployees() {
Flux<Employee> employees = employeeRepository.findAll();
return employees;
}
public Flux<Employee> getEmployeesFilterByAge(int age) {
return employeeRepository.findByAgeGreaterThan(age);
}
public Mono<Employee> getEmployeeById(int id) {
return employeeRepository.findById(id);
}
}
3.2.数据库配置
然后，让我们在application.properties中指定用于连接Cassandra的密钥空间和端口：
spring.data.cassandra.keyspace-name=practice
spring.data.cassandra.port=9042
4.测试端点
最后，是时候测试我们的API端点了。
4.1.手动测试
首先，让我们从数据库中获取员工记录：
curl localhost:8080/employee/list
结果，我们得到了所有员工：
[
{
"id": 324,
"name": "Adam Smith",
"address": "North Carolina",
"email": "asmith@xyz.com",
"age": 43
},
{
"id": 123,
"name": "John Doe",
"address": "Delaware",
"email": "jdoe@xyz.com",
"age": 31
},
{
"id": 355,
"name": "Kevin Dunner",
"address": "Virginia",
"email": "kdunner@xyz.com",
"age": 24
},
{
"id": 643,
"name": "Mike Lauren",
"address": "New York",
"email": "mlauren@xyz.com",
"age": 41
}
]
继续，让我们尝试通过他的id找到一个特定的员工：
curl localhost:8080/employee/643
结果，我们让Mike Lauren先生回来了：
{
"id": 643,
"name": "Mike Lauren",
"address": "New York",
"email": "mlauren@xyz.com",
"age": 41
}
最后，让我们看看我们的年龄过滤器是否有效：
curl localhost:8080/employee/filterByAge/35
正如预期的那样，我们得到所有年龄超过35岁的员工：
[
{
"id": 324,
"name": "Adam Smith",
"address": "North Carolina",
"email": "asmith@xyz.com",
"age": 43
},
{
"id": 643,
"name": "Mike Lauren",
"address": "New York",
"email": "mlauren@xyz.com",
"age": 41
}
]
4.2.集成测试
另外，让我们通过编写测试用例来测试相同的功能：
@RunWith(SpringRunner.class)
@SpringBootTest
public class ReactiveEmployeeRepositoryIntegrationTest {
@Autowired
EmployeeRepository repository;
@Before
public void setUp() {
Flux<Employee> deleteAndInsert = repository.deleteAll()
.thenMany(repository.saveAll(Flux.just(
new Employee(111, "John Doe", "Delaware", "jdoe@xyz.com", 31),
new Employee(222, "Adam Smith", "North Carolina", "asmith@xyz.com", 43),
new Employee(333, "Kevin Dunner", "Virginia", "kdunner@xyz.com", 24),
new Employee(444, "Mike Lauren", "New York", "mlauren@xyz.com", 41))));
StepVerifier
.create(deleteAndInsert)
.expectNextCount(4)
.verifyComplete();
}
@Test
public void givenRecordsAreInserted_whenDbIsQueried_thenShouldIncludeNewRecords() {
Mono<Long> saveAndCount = repository.count()
.doOnNext(System.out::println)
.thenMany(repository
.saveAll(Flux.just(
new Employee(325, "Kim Jones", "Florida", "kjones@xyz.com", 42),
new Employee(654, "Tom Moody", "New Hampshire", "tmoody@xyz.com", 44))))
.last()
.flatMap(v -> repository.count())
.doOnNext(System.out::println);
StepVerifier
.create(saveAndCount)
.expectNext(6L)
.verifyComplete();
}
@Test
public void givenAgeForFilter_whenDbIsQueried_thenShouldReturnFilteredRecords() {
StepVerifier
.create(repository.findByAgeGreaterThan(35))
.expectNextCount(2)
.verifyComplete();
}
}
5.结论
概括一下，我们学习了如何使用Spring Data Cassandra来构建非阻塞应用程序。与往常一样，可在GitHub上查看本教程的源代码。
推荐：
上一篇：
关注公众号
点击原文阅读更多
以上是关于具有响应式Cassandra 的Spring Data的主要内容，如果未能解决你的问题，请参考以下文章
使用 Spring-data-cassandra 查询具有复合主键的表
使用Spring-data-cassandra查询具有复合主键的表
@PrimaryKeyColumn 注释必须具有 PARTITIONED 类型，用于 scala Cassandra Spring Data 应用程序