如何在Spring Reactor Web应用程序中执行一系列操作并确保在完成下一个操作之前完成一个操作? [英] How to do sequence of operations and ensure one operation is complete before next one in Spring Reactor web app?
问题描述
我有Spring Boot 2 Web应用程序，我需要在其中通过Cookie识别网站访问者并收集页面浏览量统计信息.因此，我需要拦截每个Web请求.我必须编写的代码比回调地狱要复杂得多(Spring Reactor应该解决的问题).
I have Spring Boot 2 web app in which I need to identify site visitor by cookie and gather page view stats. So I need to intercept every web request. The code I had to write is more complex than call back hell (the very problem Spring reactor was supposed to solve).
这是代码:
package mypack.conf;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
import org.springframework.http.HttpCookie;
import org.springframework.http.ResponseCookie;
import org.springframework.web.reactive.config.ResourceHandlerRegistry;
import org.springframework.web.reactive.config.WebFluxConfigurer;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import mypack.dao.PageViewRepository;
import mypack.dao.UserRepository;
import mypack.domain.PageView;
import mypack.domain.User;
import mypack.security.JwtProvider;
import reactor.core.publisher.Mono;
@Configuration
@ComponentScan(basePackages = "mypack")
@EnableReactiveMongoRepositories(basePackages = "mypack")
public class WebConfig implements WebFluxConfigurer {
@Autowired
@Lazy
private UserRepository userRepository;
@Autowired
@Lazy
private PageViewRepository pageViewRepository;
@Autowired
@Lazy
JwtProvider jwtProvider;
@Bean
public WebFilter sampleWebFilter() {
return new WebFilter() {
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
String uri = exchange.getRequest().getURI().toString();
String path = exchange.getRequest().getPath().pathWithinApplication().value();
HttpCookie cookie = null;
String token = "";
Map<String, List<HttpCookie>> cookies = exchange.getRequest().getCookies();
try {
if((exchange.getRequest().getCookies().containsKey("_token") )
&& (exchange.getRequest().getCookies().getFirst("_token"))!=null ) {
cookie = exchange.getRequest().getCookies().getFirst("_token");
token = cookie.getValue();
return userRepository.findByToken(token).map(user -> {
exchange.getAttributes().put("_token", user.getToken());
PageView pg = PageView.builder().createdDate(LocalDateTime.now()).URL(uri).build();
pageViewRepository.save(pg).subscribe(pg1 -> {user.getPageviews().add(pg1); });
userRepository.save(user).subscribe();
return user;
})
.flatMap(user-> chain.filter(exchange)); // ultimately this step executes regardless user exist or not
// handle case when brand new user first time visits website
} else {
token = jwtProvider.genToken("guest", UUID.randomUUID().toString());
User user = User.builder().createdDate(LocalDateTime.now()).token(token).emailId("guest").build();
userRepository.save(user).subscribe();
exchange.getResponse().getCookies().remove("_token");
ResponseCookie rcookie = ResponseCookie.from("_token", token).httpOnly(true).build();
exchange.getResponse().addCookie(rcookie);
exchange.getAttributes().put("_token", token);
}
} catch (Exception e) {
e.printStackTrace();
}
return chain.filter(exchange);
} // end of Mono<Void> filter method
}; // end of New WebFilter (anonymous class)
}
}
其他相关类别:
@Repository
public interface PageViewRepository extends ReactiveMongoRepository<PageView, String>{
Mono<PageView> findById(String id);
}
@Repository
public interface UserRepository extends ReactiveMongoRepository<User, String>{
Mono<User> findByToken(String token);
}
@Data
@AllArgsConstructor
@Builder
@NoArgsConstructor
public class User {
@Id
private String id;
private String token;
@Default
private LocalDateTime createdDate = LocalDateTime.now();
@DBRef
private List<PageView> pageviews;
}
Data
@Document
@Builder
public class PageView {
@Id
private String id;
private String URL;
@Default
private LocalDateTime createdDate = LocalDateTime.now();
}
gradle文件的相关部分:
Relevant part of gradle file:
buildscript {
ext {
springBootVersion = '2.0.1.RELEASE'
}
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
}
}
dependencies {
compile('org.springframework.boot:spring-boot-starter-data-mongodb-reactive')
compile('org.springframework.boot:spring-boot-starter-security')
compile('org.springframework.boot:spring-boot-starter-thymeleaf')
compile('org.springframework.boot:spring-boot-starter-webflux')
compile('org.springframework.security:spring-security-oauth2-client')
compile('org.springframework.security.oauth:spring-security-oauth2:2.3.4.RELEASE')
runtime('org.springframework.boot:spring-boot-devtools')
compileOnly('org.projectlombok:lombok')
compile "org.springframework.security:spring-security-jwt:1.0.9.RELEASE"
compile "io.jsonwebtoken:jjwt:0.9.0"
testCompile('org.springframework.boot:spring-boot-starter-test')
testCompile('io.projectreactor:reactor-test')
compile('com.fasterxml.jackson.core:jackson-databind')
}
问题出在以下几行:
PageView pg = PageView.builder().createdDate(LocalDateTime.now()).URL(uri).build(); pageViewRepository.save(pg).subscribe(pg1-> {user.getPageviews().add(pg1);});
PageView pg = PageView.builder().createdDate(LocalDateTime.now()).URL(uri).build(); pageViewRepository.save(pg).subscribe(pg1 -> {user.getPageviews().add(pg1); });
这会挂起浏览器(保持等待状态).
which hangs the browser (keeps waiting for response).
基本上我想要的是: 一定不能使用甚至在webfilter代码中也不起作用的block()来阻止 挂起浏览器. 将综合浏览量保存在mongo db中.保存后，综合浏览量具有有效的mongodb id，需要将其作为引用存储在综合浏览量用户实体列表中.因此，只有将其保存在数据库中之后，下一步才是更新用户的综合浏览量列表. 下一步是在不影响下游控制器方法的情况下保存更新的用户，下游控制器方法也可能会更新用户，并且可能也需要保存用户. 所有这些都应该在给定的WebFilter上下文中起作用.
Basically what I want is this: Must not use block() which does not even work in webfilter code as block also hangs the browser. Save pageview in mongo db. After it is saved, pageview has valid mongodb id which is needed to be stored as reference in pageviews List of user entity. Therefore only after it is saved in db, the next step is update user's pageviews List. Next step is save the updated user without effecting downstream controller methods which may also update user and may need to save the user too. All this should work in the given WebFilter context.
如何解决这个问题?
提供的解决方案必须确保将用户保存在webfilter中，然后再执行控制器操作，其中某些操作还可以为用户保存与查询字符串参数不同的值.
The solution provided must make sure that user is saved in webfilter before passing on to controller actions some of which also saves user with different values from query string params.
推荐答案
如果我对您的理解正确，那么您需要对数据库进行长时间的异步操作，以防止过滤器(和请求本身)阻塞?
If I understand you correctly, you need to perform long operations with database asynchronously to prevent the filter (and the request itself) from blocking?
在这种情况下，我建议适用于我的以下解决方案:
In this case, I would recommend the following solution that works for me:
@Bean
public WebFilter filter() {
return (exchange, chain) -> {
ServerHttpRequest req = exchange.getRequest();
String uri = req.getURI().toString();
log.info("[i] Got request: {}", uri);
var headers = req.getHeaders();
List<String> tokenList = headers.get("token");
if (tokenList != null && tokenList.get(0) != null) {
String token = tokenList.get(0);
log.info("[i] Find a user by token {}", token);
return userRepo.findByToken(token)
.map(user -> process(exchange, uri, token, user))
.then(chain.filter(exchange));
} else {
String token = UUID.randomUUID().toString();
log.info("[i] Create a new user with token {}", token);
return userRepo.save(new User(token))
.map(user -> process(exchange, uri, token, user))
.then(chain.filter(exchange));
}
};
}
在这里，我略微更改了您的逻辑，并从适当的标头(而不是从Cookie)获取令牌值，以简化实现.
Here I slightly change your logic and take the token value from the appropriate header (not from cookies) to simplify my implementation.
因此，如果令牌存在，则我们尝试找到其用户.如果令牌不存在，那么我们将创建一个新用户.如果成功找到或创建了用户，则说明正在调用process方法.之后，无论结果如何，我们都返回chain.filter(exchange).
So if the token is present then we try to find its user. If the token isn't present then we create a new user. If the user is found or created successfully, then the process method is calling. After that, regardless of the result, we return chain.filter(exchange).
方法process将令牌值放入请求的适当属性，并异步调用userService的方法updateUserStat:
The method process puts a token value to the appropriate attribute of the request and call asynchronously the method updateUserStat of the userService:
private User process(ServerWebExchange exchange, String uri, String token, User user) {
exchange.getAttributes().put("_token", token);
userService.updateUserStat(uri, user); // async call
return user;
}
用户服务:
@Slf4j
@Service
public class UserService {
private final UserRepo userRepo;
private final PageViewRepo pageViewRepo;
public UserService(UserRepo userRepo, PageViewRepo pageViewRepo) {
this.userRepo = userRepo;
this.pageViewRepo = pageViewRepo;
}
@SneakyThrows
@Async
public void updateUserStat(String uri, User user) {
log.info("[i] Start updating...");
Thread.sleep(1000);
pageViewRepo.save(new PageView(uri))
.flatMap(user::addPageView)
.blockOptional()
.ifPresent(u -> userRepo.save(u).block());
log.info("[i] User updated.");
}
}
为了测试目的，我在此处添加了一个小的延迟，以确保无论此方法的持续时间如何，请求都不会出现任何延迟.
I've added here a small delay for test purposes to make sure that requests work without any delay, regardless of the duration of this method.
通过令牌找到用户的情况:
A case when the user is found by the token:
2019-01-06 18:25:15.442 INFO 4992 --- [ctor-http-nio-3] : [i] Got request: http://localhost:8080/users?test=1000
2019-01-06 18:25:15.443 INFO 4992 --- [ctor-http-nio-3] : [i] Find a user by token 84b0f7ec-670c-4c04-8a7c-b692752d7cfa
2019-01-06 18:25:15.444 DEBUG 4992 --- [ctor-http-nio-3] : Created query Query: { "token" : "84b0f7ec-670c-4c04-8a7c-b692752d7cfa" }, Fields: { }, Sort: { }
2019-01-06 18:25:15.445 DEBUG 4992 --- [ctor-http-nio-3] : find using query: { "token" : "84b0f7ec-670c-4c04-8a7c-b692752d7cfa" } fields: Document{{}} for class: class User in collection: user
2019-01-06 18:25:15.457 INFO 4992 --- [ntLoopGroup-2-2] : [i] Get all users...
2019-01-06 18:25:15.457 INFO 4992 --- [ task-3] : [i] Start updating...
2019-01-06 18:25:15.458 DEBUG 4992 --- [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user
2019-01-06 18:25:16.459 DEBUG 4992 --- [ task-3] : Inserting Document containing fields: [URL, createdDate, _class] in collection: pageView
2019-01-06 18:25:16.476 DEBUG 4992 --- [ task-3] : Saving Document containing fields: [_id, token, pageViews, _class]
2019-01-06 18:25:16.479 INFO 4992 --- [ task-3] : [i] User updated.
在这里我们可以看到，在用户已经具有获取所有用户"请求的结果之后，在独立的task-3线程中执行了更新用户的操作.
Here we can see that updating the user is performed in the independent task-3 thread after the user already has a result of 'get all users' request.
不存在令牌并创建用户的情况:
A case when the token is not present and the user is created:
2019-01-06 18:33:54.764 INFO 4992 --- [ctor-http-nio-3] : [i] Got request: http://localhost:8080/users?test=763
2019-01-06 18:33:54.764 INFO 4992 --- [ctor-http-nio-3] : [i] Create a new user with token d9bd40ea-b869-49c2-940e-83f1bf79e922
2019-01-06 18:33:54.765 DEBUG 4992 --- [ctor-http-nio-3] : Inserting Document containing fields: [token, _class] in collection: user
2019-01-06 18:33:54.776 INFO 4992 --- [ntLoopGroup-2-2] : [i] Get all users...
2019-01-06 18:33:54.777 INFO 4992 --- [ task-4] : [i] Start updating...
2019-01-06 18:33:54.777 DEBUG 4992 --- [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user
2019-01-06 18:33:55.778 DEBUG 4992 --- [ task-4] : Inserting Document containing fields: [URL, createdDate, _class] in collection: pageView
2019-01-06 18:33:55.792 DEBUG 4992 --- [ task-4] : Saving Document containing fields: [_id, token, pageViews, _class]
2019-01-06 18:33:55.795 INFO 4992 --- [ task-4] : [i] User updated.
存在令牌但找不到用户的情况:
A case when the token is present but user is not found:
2019-01-06 18:35:40.970 INFO 4992 --- [ctor-http-nio-3] : [i] Got request: http://localhost:8080/users?test=150
2019-01-06 18:35:40.970 INFO 4992 --- [ctor-http-nio-3] : [i] Find a user by token 184b0f7ec-670c-4c04-8a7c-b692752d7cfa
2019-01-06 18:35:40.972 DEBUG 4992 --- [ctor-http-nio-3] : Created query Query: { "token" : "184b0f7ec-670c-4c04-8a7c-b692752d7cfa" }, Fields: { }, Sort: { }
2019-01-06 18:35:40.972 DEBUG 4992 --- [ctor-http-nio-3] : find using query: { "token" : "184b0f7ec-670c-4c04-8a7c-b692752d7cfa" } fields: Document{{}} for class: class User in collection: user
2019-01-06 18:35:40.977 INFO 4992 --- [ntLoopGroup-2-2] : [i] Get all users...
2019-01-06 18:35:40.978 DEBUG 4992 --- [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user
我的演示项目: sb-reactive-filter-demo
这篇关于如何在Spring Reactor Web应用程序中执行一系列操作并确保在完成下一个操作之前完成一个操作?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
