通过stream去重_stream去重

引言Spring Boot 2.0最近去了GA，所以我决定写我关于Spring的第一篇文章很长一段时间引言Spring Boot 2.0最近去了GA，所以我决定写我关于Spring的第一篇文章很长一段时间。自发布以来，我一直在看到越来越多的Spring WebFlux以及如何使用它的教程。但是在阅读完它们并尝试让它自己工作之后，我发现从包含在我阅读的文章和教程中的代码跳转到编写实际上比返回字符串更有趣的事情从后端。现在，我希望我不会在自己的脚下说自己可能会对我在这篇文章中使用的代码做出同样的批评，但这里是我试图给Spring WebFlux的教程，它实际上类似于你可能会在野外使用的东西。项目结构:在我继续之前，在提及WebFlux之后，究竟是什么呢？Spring WebFlux是Spring MVC的完全非阻塞反应式替代方案。它允许更好的垂直缩放而不增加硬件资源。被动反应它现在使用Reactive Streams来允许从调用返回到服务器的数据的异步处理。这意味着我们将看到更少的Lists，Collection甚至单个对象，而不是他们的反应等价物，例如Flux和Mono(来自Reactor)。我不会深入研究Reactive Streams是什么，诚实地说，在我尝试向任何人解释它之前，我需要更加深入地研究它。相反，让我们回过头来关注WebFlux。像往常一样，我使用Spring Boot在本教程中编写代码。以下是我在这篇文章中使用的依赖关系。&ltdependencies>

&ltdependency>

&ltgroupId>org.springframework.boot&lt/groupId>

&ltartifactId>spring-boot-starter-webflux&lt/artifactId>

&lt/dependency>

&ltdependency>

&ltgroupId>org.springframework.boot&lt/groupId>

&ltartifactId>spring-boot-starter-data-cassandra-reactive&lt/artifactId>

&ltversion>2.0.0.RELEASE&lt/version>

&lt/dependency>

&lt/dependencies>尽管我没有将它包含在上面的依赖代码片段中，但是它spring-boot-starter-parent被使用了，最终可以将其提升到版本2.0.0.RELEASE。本教程是关于WebFlux的，包括这spring-boot-starter-webflux显然是一个好主意。spring-boot-starter-data-cassandra-reactive也被包括在内，因为我们将用它作为示例应用程序的数据库，因为它是少数几个有反应支持的数据库之一(在编写本文时)。通过一起使用这些依赖关系，我们的应用程序可以从前到后完全反应。WebFlux引入了一种不同的方式来处理请求，而不是使用Spring MVC 中使用的@Controller或@RestController编程模型。但是，它并没有取代它。相反，它已被更新以允许使用被动类型。这使您可以保持与使用Spring编写相同的格式，但对返回类型进行一些更改，以便返回Fluxs或Monos。下面是一个非常人为的例子。@RestController

public class PersonController {

private final PersonRepository personRepository;

public PersonController(PersonRepository personRepository) {

this.personRepository = personRepository;

}

@GetMapping("/people")

public Flux&ltPerson> all() {

return personRepository.findAll();

}

@GetMapping("/people/{id}")

Mono&ltPerson> findById(@PathVariable String id) {

return personRepository.findOne(id);

}

}对我来说，这看起来非常熟悉，并且从一眼就可以看出它与标准的Spring MVC控制器没有任何区别，但通过阅读方法后，我们可以看到不同的返回类型。在这个例子中PersonRepository必须是一个被动库，因为我们已经能够直接返回他们的搜索查询的结果供参考，被动库会返回一个Flux集合和一个Mono单一的实体。注释方法不是我想在这篇文章中关注的内容。这对我们来说不够酷，时髦。没有足够的lambda表达式来满足我们以更有效的方式编写Java的渴望。但Spring WebFlux有我们的支持。它提供了一种替代方法来路由和处理请求到我们的服务器，轻轻地使用lambdas编写路由器功能。我们来看一个例子。@Configuration

public class PersonRouter {

@Bean

public RouterFunction&ltServerResponse> route(PersonHandler personHandler) {

return RouterFunctions.route(GET("/people/{id}").and(accept(APPLICATION_JSON)), personHandler::get)

.andRoute(GET("/people").and(accept(APPLICATION_JSON)), personHandler::all)

.andRoute(POST("/people").and(accept(APPLICATION_JSON)).and(contentType(APPLICATION_JSON)), personHandler::post)

.andRoute(PUT("/people/{id}").and(accept(APPLICATION_JSON)).and(contentType(APPLICATION_JSON)), personHandler::put)

.andRoute(DELETE("/people/{id}"), personHandler::delete)

.andRoute(GET("/people/country/{country}").and(accept(APPLICATION_JSON)), personHandler::getByCountry);

}

} 这些都是PersonHandler我们稍后会看到的方法的所有路线。我们创建了一个将处理我们路由的bean。为了设置路由功能，我们使用了名为的RouterFunctions类为我们提供了一个静态方法，但现在我们只关心它的route方法。以下是该route方法的签名。public static &ltT extends ServerResponse> RouterFunction&ltT> route(

RequestPredicate predicate, HandlerFunction&ltT> handlerFunction) {

// stuff

} 该方法表明，它与a RequestPredicate一起HandlerFunction并输出a RouterFunction。这RequestPredicate是我们用来指定路由的行为，比如我们处理函数的路径，它是什么类型的请求以及它可以接受的输入类型。由于我使用静态导入将所有内容读得更清晰，所以一些重要信息已经隐藏起来。要创建一个RequestPredicate我们应该使用RequestPredicates(复数)，一个静态帮助类为我们提供我们需要的所有方法。就个人而言，我建议静态导入，RequestPredicates否则由于使用RequestPredicates静态方法可能需要的次数，您的代码将会一团糟。在上述例子中，GET，POST，PUT，DELETE，accept和contentType都是静态RequestPredicates方法。下一个参数是a HandlerFunction，它是一个功能接口。这里有三件重要的信息，它有一个泛型类型&ltT extends ServerResponse>，它的handle方法返回一个Mono&ltT>并且需要一个ServerRequest。使用这些我们可以确定我们需要传递一个返回一个Mono&ltServerResponse>(或它的一个子类型)的函数。这显然对我们的处理函数返回的内容有严格的约束，因为它们必须满足这个要求，否则它们将不适合以这种格式使用。最后的结果是一个RouterFunction。这可以返回并用于路由到我们指定的任何函数。但通常情况下，我们希望一次将很多不同的请求发送给各种处理程序，这是WebFlux迎合的。由于route返回a RouterFunction以及RouterFunction也有其自己的路由方法的事实andRoute，我们可以将这些调用链接在一起并继续添加我们所需的所有额外路由。如果我们再回头看一下PersonRouter上面的例子，我们可以看到这些方法是以REST动词命名的，例如GET，POST它们定义了处理程序将要执行的请求的路径和类型。GET例如，如果我们以第一个请求为例，它将/people使用路径变量名称id(path表示的路径变量{id})和返回内容的类型(具体来说，使用该方法定义的APPLICATION_JSON静态字段from MediaType)进行路由accept。如果使用不同的路径，则不会被处理。如果路径正确但Accept头不是可接受的类型之一，则请求将失败。在我们继续之前，我想了解一下accept和contentType方法。这两个设置请求标头都accept与Accept标头和contentTypeContent-Type 匹配。Accept头定义了响应可接受的媒体类型，因为我们返回的Person对象的JSON表示设置为APPLICATION_JSON(application/json在实际头文件中)是有意义的。Content-Type具有相同的想法，但是却描述了发送请求正文内的媒体类型。这就是为什么只有动词POST和PUT动词才contentType包括在内，因为其他人在他们的身体中没有任何东西。DELETE不包括accept和contentType 所以我们可以得出这样的结论：它既没有期望返回任何东西，也没有在其请求中包含任何东西。现在我们知道如何设置路由，让我们看看如何编写处理传入请求的处理程序方法。以下是处理前面示例中定义的路由的所有请求的代码。@Component

public class PersonHandler {

private final PersonManager personManager;

public PersonHandler(PersonManager personManager) {

this.personManager = personManager;

}

public Mono&ltServerResponse> get(ServerRequest request) {

final UUID id = UUID.fromString(request.pathVariable("id"));

final Mono&ltPerson> person = personManager.findById(id);

return person

.flatMap(p -> ok().contentType(APPLICATION_JSON).body(fromPublisher(person, Person.class)))

.switchIfEmpty(notFound().build());

}

public Mono&ltServerResponse> all(ServerRequest request) {

return ok().contentType(APPLICATION_JSON)

.body(fromPublisher(personManager.findAll(), Person.class));

}

public Mono&ltServerResponse> put(ServerRequest request) {

final UUID id = UUID.fromString(request.pathVariable("id"));

final Mono&ltPerson> person = request.bodyToMono(Person.class);

return personManager

.findById(id)

.flatMap(

old ->

ok().contentType(APPLICATION_JSON)

.body(

fromPublisher(

person

.map(p -> new Person(p, id))

.flatMap(p -> personManager.update(old, p)),

Person.class)))

.switchIfEmpty(notFound().build());

}

public Mono&ltServerResponse> post(ServerRequest request) {

final Mono&ltPerson> person = request.bodyToMono(Person.class);

final UUID id = UUID.randomUUID();

return created(UriComponentsBuilder.fromPath("people/" + id).build().toUri())

.contentType(APPLICATION_JSON)

.body(

fromPublisher(

person.map(p -> new Person(p, id)).flatMap(personManager::save), Person.class));

}

public Mono&ltServerResponse> delete(ServerRequest request) {

final UUID id = UUID.fromString(request.pathVariable("id"));

return personManager

.findById(id)

.flatMap(p -> noContent().build(personManager.delete(p)))

.switchIfEmpty(notFound().build());

}

public Mono&ltServerResponse> getByCountry(ServerRequest serverRequest) {

final String country = serverRequest.pathVariable("country");

return ok().contentType(APPLICATION_JSON)

.body(fromPublisher(personManager.findAllByCountry(country), Person.class));

}

} 有一点非常明显，就是缺少注释。酒吧的@Component注释自动创建一个PersonHandler豆没有其他Spring注解。我试图将大部分存储库逻辑保留在这个类之外，并且通过经由它所包含的PersonManager代理来隐藏对实体对象的任何引用PersonRepository。如果你对代码感兴趣，PersonManager那么可以在我的GitHub上看到，关于它的进一步解释将被排除在这篇文章之外，所以我们可以专注于WebFlux本身。好的，回到手头的代码。让我们仔细看看get和post方法来弄清楚发生了什么。public Mono&ltServerResponse> get(ServerRequest request) {

final UUID id = UUID.fromString(request.pathVariable("id"));

final Mono&ltPerson> person = personManager.findById(id);

return person

.flatMap(p -> ok().contentType(APPLICATION_JSON).body(fromPublisher(person, Person.class)))

.switchIfEmpty(notFound().build());

}此方法用于从支持此示例应用程序的数据库中检索单个记录。由于Cassandra是选择的数据库，我决定使用UUID每个记录的主键，这使得测试示例更令人讨厌的不幸效果，但没有任何复制和粘贴无法解决的问题。请记住，此GET请求的路径中包含路径变量。使用pathVariable的方法ServerRequest传递到我们能够提取它的价值通过提供变量的名称，在这种情况下，方法id。然后将ID转换成一个UUID，如果字符串格式不正确，它会抛出一个异常，我决定忽略这个问题，所以示例代码不会变得混乱。一旦我们有了ID，我们就可以查询数据库中是否存在匹配的记录。Mono&ltPerson>返回的A 包含映射到a的现有记录，Person或者它保留为空Mono。使用返回的，Mono我们可以根据它的存在输出不同的响应。这意味着我们可以将有用的状态代码返回给客户端以跟随主体的内容。如果记录存在，则flatMap返回一个ServerResponse与OK状态。伴随着这种状态，我们希望输出记录，为此，我们在这种情况下指定正文的内容类型APPLICATION_JSON，并将记录添加到记录中。fromPublisher需要我们Mono&ltPerson>(这是一个Publisher)与Person课程一起，因此它知道它映射到身体中的是什么。fromPublisher是类的静态方法BodyInserters。如果记录不存在，那么流程将移动到switchIfEmpty块中并返回NOT FOUND状态。因为没有发现，身体可以留空，所以我们只是创建ServerResponse那里。现在到post处理程序。public Mono&ltServerResponse> post(ServerRequest request) {

final Mono&ltPerson> person = request.bodyToMono(Person.class);

final UUID id = UUID.randomUUID();

return created(UriComponentsBuilder.fromPath("people/" + id).build().toUri())

.contentType(APPLICATION_JSON)

.body(

fromPublisher(

person.map(p -> new Person(p, id)).flatMap(personManager::save), Person.class));

} 即使只是从第一行开始，我们就可以看到，这种get方法的工作方式已经不同了。由于这是一个POST请求，它需要接受我们希望从请求主体持续存在的对象。由于我们试图插入单个记录，因此我们将使用请求的bodyToMono方法Person从正文中检索。如果您正在处理多个记录，则可能需要使用它们bodyToFlux。我们将CREATED使用created接受a 的方法返回状态URI以确定插入记录的路径。然后，get通过使用该fromPublisher方法将新记录添加到响应主体，然后采用与该方法类似的设置。形成该代码的代码Publisher稍有不同，但输出仍然Mono&ltPerson>是一个重要的内容。为了进一步解释如何完成插入Person，从请求传入的内容将被映射到Person使用UUID我们生成的新内容，然后通过save调用传递给新内容flatMap。通过创建一个新的Person我们只将值插入我们允许的Cassandra中，在这种情况下，我们不希望UUID从请求体传入。所以说，这是关于处理程序。显然还有其他方法，我们没有经历。它们的工作方式都不相同，但都遵循相同的概念，ServerResponse如果需要，它返回一个包含适当状态代码和记录的体系。现在我们已经编写了所有我们需要的代码来获得基本的Spring WebFlux后端运行。剩下的就是将所有配置绑定在一起，这对Spring Boot来说很简单。@SpringBootApplication

public class Application {

public static void main(String args[]) {

SpringApplication.run(Application.class);

}

} 我们应该研究如何真正使用代码，而不是结束这篇文章。Spring提供了WebClient该类来处理请求而不会阻塞。我们现在可以利用这个来测试应用程序，尽管WebTestClient我们也可以在这里使用它。该WebClient是你可以使用，而不是阻止什么RestTemplate产生反应的应用程序时。下面是一些调用在PersonHandler。中定义的处理程序的代码。public class Client {

private WebClient client = WebClient.create("http://localhost:8080");

public void doStuff() {

// POST

final Person record = new Person(UUID.randomUUID(), "John", "Doe", "UK", 50);

final Mono&ltClientResponse> postResponse =

client

.post()

.uri("/people")

.body(Mono.just(record), Person.class)

.accept(APPLICATION_JSON)

.exchange();

postResponse

.map(ClientResponse::statusCode)

.subscribe(status -> System.out.println("POST: " + status.getReasonPhrase()));

// GET

client

.get()

.uri("/people/{id}", "a4f66fe5-7c1b-4bcf-89b4-93d8fcbc52a4")

.accept(APPLICATION_JSON)

.exchange()

.flatMap(response -> response.bodyToMono(Person.class))

.subscribe(person -> System.out.println("GET: " + person));

// ALL

client

.get()

.uri("/people")

.accept(APPLICATION_JSON)

.exchange()

.flatMapMany(response -> response.bodyToFlux(Person.class))

.subscribe(person -> System.out.println("ALL: " + person));

// PUT

final Person updated = new Person(UUID.randomUUID(), "Peter", "Parker", "US", 18);

client

.put()

.uri("/people/{id}", "ec2212fc-669e-42ff-9c51-69782679c9fc")

.body(Mono.just(updated), Person.class)

.accept(APPLICATION_JSON)

.exchange()

.map(ClientResponse::statusCode)

.subscribe(response -> System.out.println("PUT: " + response.getReasonPhrase()));

// DELETE

client

.delete()

.uri("/people/{id}", "ec2212fc-669e-42ff-9c51-69782679c9fc")

.exchange()

.map(ClientResponse::statusCode)

.subscribe(status -> System.out.println("DELETE: " + status));

}

} 不要忘了在Client某个地方实例化，下面是一个很好的偷懒方式来做到这一点！@SpringBootApplication

public class Application {

public static void main(String args[]) {

SpringApplication.run(Application.class);

Client client = new Client();

client.doStuff();

}

} 首先我们创建一个WebClient。private final WebClient client = WebClient.create("http://localhost:8080");

一旦创建，我们就可以开始做它的东西，因此doStuff方法。我们来分解一下POST发送给后端的请求。final Mono&ltClientResponse> postResponse =

client

.post()

.uri("/people")

.body(Mono.just(record), Person.class)

.accept(APPLICATION_JSON)

.exchange();

postResponse

.map(ClientResponse::statusCode)

.subscribe(status -> System.out.println("POST: " + status.getReasonPhrase()));我写下这个稍有不同，所以你可以看到a Mono&ltClientResponse>是从发送请求返回的。该exchange方法将HTTP请求发送到服务器。然后，只要响应到达，就会处理响应，如果有的话。使用WebClient我们指定我们想要POST使用post当然的方法发送请求。在URI随后与所添加的uri方法(重载的方法，这一个接受一个String但另一个接受URI)。我厌倦了说这个方法做了什么方法，所以，身体的内容随后与Accept头一起添加。最后我们通过电话发送请求exchange。请注意，媒体类型APPLICATION_JSON与POST路由器功能中定义的类型相匹配。如果我们要发送不同的类型，比如说TEXT_PLAIN我们会得到一个404错误，因为没有处理程序存在与请求期望返回的内容相匹配的地方。使用Mono&ltClientResponse>通过调用返回exchange，我们可以绘制它的内容给我们所需的输出。在上面的例子中，状态代码被打印到控制台。如果我们回想一下post方法PersonHandler，请记住它只能返回“创建”状态，但如果发送的请求没有正确匹配，则会打印出“未找到”。我们来看看其他请求之一。client

.get()

.uri("/people/{id}", "a4f66fe5-7c1b-4bcf-89b4-93d8fcbc52a4")

.accept(APPLICATION_JSON)

.exchange()

.flatMap(response -> response.bodyToMono(Person.class))

.subscribe(person -> System.out.println("GET: " + person));这是我们的典型GET要求。它看起来与POST我们刚刚经历的请求非常相似。主要区别在于uri，请求路径和UUID(作为String在这种情况下)作为参数来取代路径变量{id}并且主体留空。响应如何处理也是不同的。在这个例子中，它提取了响应的主体并将其映射到a Mono&ltPerson>并打印出来。这可以在前面的POST例子中完成，但是响应的状态代码对于它的情况更有用。对于略有不同的观点，我们可以使用cURL发出请求并查看响应的样子。CURL -H "Accept:application/json" -i localhost:8080/people

HTTP/1.1 200 OK

transfer-encoding: chunked

Content-Type: application/json

[

{

"id": "13c403a2-6770-4174-8b76-7ba7b75ef73d",

"firstName": "John",

"lastName": "Doe",

"country": "UK",

"age": 50

{

"id": "fbd53e55-7313-4759-ad74-6fc1c5df0986",

"firstName": "Peter",

"lastName": "Parker",

"country": "US",

"age": 50

}

]

响应看起来像这样，显然它会根据您存储的数据而有所不同。请注意响应标题。transfer-encoding: chunked

Content-Type: application/json 在transfer-encoding这里表示的是在可用于流式传输的数据块传输的数据。这就是我们需要的，因此客户可以对返回给它的数据采取反应态度。我认为这应该是一个停止的好地方。我们在这里已经涵盖了相当多的材料，希望能够帮助您更好地理解Spring WebFlux。还有一些其他的话题我想关注WebFlux，但是我会在单独的帖子中做这些，因为我认为这个主题足够长。总之，在这篇文章中，我们非常简要地讨论了为什么你想在典型的Spring MVC后端中使用Spring WebFlux。然后我们看看如何设置路由和处理程序来处理传入的请求。处理程序实现了可以处理大多数REST动词的方法，并在响应中返回了正确的数据和状态代码。最后，我们研究了向后端发送请求的两种方式，一种是使用a WebClient直接在客户端处理输出，另一种使用cURL查看返回的JSON的外观。注意源码中使用的数据为：cassandra提示：项目源码下载:demo-springboot-webflux-0401.zip