Как выполнить последовательность операций и обеспечить выполнение одной операции перед следующей в веб-приложении Spring Reactor?
У меня есть веб-приложение Spring Boot 2, в котором мне нужно идентифицировать посетителя сайта по cookie и собирать статистику просмотров страницы. Поэтому мне нужно перехватывать каждый веб-запрос. Код, который мне пришлось написать, сложнее, чем ад обратного вызова (та самая проблема, которую должен был решить Spring реактор).
Вот код:
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:
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); });
что вешает браузер (продолжает ждать ответа).
В основном я хочу вот что: Нельзя использовать block (), который даже не работает в коде веб-фильтра, также как блок зависает браузер. Сохранить просмотр страницы в mongo db. После сохранения pageview имеет действительный идентификатор mongodb, который необходимо сохранить в качестве ссылки в списке просмотров страниц объекта пользователя. Поэтому только после того, как он будет сохранен в базе данных, следующим шагом будет обновление списка просмотров страниц пользователя. Следующим шагом является сохранение обновленного пользователя без воздействия на нижестоящие методы контроллера, которые также могут обновлять пользователя и, возможно, также потребуется сохранить пользователя. Все это должно работать в данном контексте WebFilter.
Как решить эту проблему?
Предоставляемое решение должно гарантировать, что пользователь сохранен в веб-фильтре, прежде чем переходить к действиям контроллера, некоторые из которых также сохраняют пользователя с разными значениями из параметров строки запроса.
Привет @ace, 2 существующих сообщения ответили на ваш вопрос? В противном случае займусь вашим вопросом с завтрашнего дня;)
Предоставленные решения Bsquare не решают мою проблему, см. Мои комментарии к ответу Cepr0.
@ace что именно ты хочешь сделать?
Ответы 3
Для сбора статистики просмотров страниц я предлагаю изменить стратегию и вместо этого использовать Актуатор и Микрометр:
- Добавьте в проект зависимость актуатора
- Выставляем соответствующие конечные точки (здесь
metrics) - Перейдите к
/actuator/metricsи выберите метрику для HTTP-запросов сервера (см. справочная документация).
Micrometer предлагает гораздо больше и помогает вам получать правильные показатели, например: учет пауз GC при измерении времени, предоставление гистограмм / процентилей / ... и т. д.
Если я вас правильно понял, вам нужно выполнять длительные операции с базой данных асинхронно, чтобы фильтр (и сам запрос) не блокировался?
В этом случае я бы порекомендовал следующее решение, которое мне подходит:
@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), чтобы упростить свою реализацию.
Итак, если токен присутствует, мы пытаемся найти его пользователя. Если токена нет, мы создаем нового пользователя. Если пользователь найден или создан успешно, то вызывается метод process. После этого независимо от результата возвращаем chain.filter(exchange).
Метод process помещает значение токена в соответствующий атрибут запроса и асинхронно вызывает метод updateUserStatuserService:
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.");
}
}
Я добавил небольшую задержку в целях тестирования, чтобы убедиться, что запросы работают без задержек, независимо от продолжительности этого метода.
Случай, когда пользователя находят по токену:
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 после того, как у пользователя уже есть результат запроса «получить всех пользователей».
Случай, когда токена нет, а пользователь создан:
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.
Случай, когда токен присутствует, но пользователь не найден:
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-реактивный-фильтр-демонстрация
Cepr0 updateUserStat в userService использует block (). изменится ли это на блокировку и, таким образом, уничтожит ли цель использования реактивного API? В моем вопросе говорится, что нельзя использовать block (). Я, может быть, что-то упускаю, не могли бы вы рассказать об этом подробнее?
@ace Код updateUserStat вообще не блокирует выполнение методов фильтра и контроллера. Он работает в независимом потоке и «блокирует» только собственное выполнение, чтобы реактивный репозиторий сохранял данные. Вы можете запустить мою демонстрацию и проверить ее. Вы можете увеличить задержку в updateUserStat и убедиться, что она не влияет ни на какие запросы.
@ace Хитрость в том, что метод process не ждет завершения работы updateUserStat и немедленно возвращает управление основному процессу.
Cepr0 Я использовал ваш подход в своем проекте после большого количества перекодирования, к сожалению, он не работает. Он работает только для корневого пути, который показывает статическую страницу, и пользователь обновляется, но когда я перехожу на другие страницы, которые показывают список элементов, например, из mongodb, он зависает, браузер постоянно ждет ответа. Я думаю о том, чтобы изменить весь свой проект, чтобы удалить реактивную структуру Spring и преобразовать ее в синхронизирующую нереактивную обычную структуру Spring, что займет недели. В целом у меня был плохой опыт работы с фреймворком реактивной пружины в крупных проектах.
Cepr0 Я исправил проблему, вернувшись к версии Spring boot 2.0 вместо версии 2.1. Теперь ваш код работает. Я читал в блогах, что использование блока в приложении реактивной пружины является антипаттерном. Есть ли решение без использования BlockOptional или block ()?
Cepr0, к сожалению, еще раз ваше решение не работает в моем проекте, потому что некоторые действия контроллера также спасают пользователя от значений в параметрах запроса в запросе. Поскольку ваш updateUserStat запускается в собственном фоновом потоке, сохраняющем пользователя, если ваш фоновый поток позже сохраняет пользователя, он перезаписывает значение, сохраненное действием контроллера. Поэтому я все еще ищу решение, которое гарантирует, что пользователь сохранится в веб-фильтре, прежде чем переходить к действию контроллера.
@ace Не могли бы вы уточнить, что именно вам нужно? Вам нужно выполнить две неблокирующие операции: (1) найти существующего пользователя / или сохранить нового пользователя, (2) создать представление страницы и обновить пользователя в веб-фильтре перед контроллером? Или вам нужно что-то сделать с «созданием просмотра страницы и обновлением пользователя», что является довольно медленной операцией и приводит к зависанию браузера?
Cepr0 Я хочу создать просмотр страницы, обновить сохранение пользователя в веб-фильтре ПЕРЕД передачей управления действиям контроллера, которые также сохраняют пользователя с его собственным другим обновлением. И все это без зависания браузера или запуска пользователя wbfilter с сохранением в фоновом потоке.
Другой вариант, который создает просмотр страницы и обновляет пользователя в веб-фильтре неблокирующим образом перед передачей запроса контроллеру:
@Bean
public WebFilter filter() {
return (exchange, chain) -> {
ServerHttpRequest req = exchange.getRequest();
String uri = req.getURI().toString();
log.info("[i] Web Filter: received the request: {}", uri);
var headers = req.getHeaders();
List<String> tokenList = headers.get("token");
if (tokenList != null && tokenList.get(0) != null) {
String token = tokenList.get(0);
Mono<User> foundUser = userRepo
.findByToken(token)
.doOnNext(user -> log.info("[i] Web Filter: {} has been found", user));
return updateUserStat(foundUser, exchange, chain, uri);
} else {
String token = UUID.randomUUID().toString();
Mono<User> createdUser = userRepo
.save(new User(token))
.doOnNext(user -> log.info("[i] Web Filter: a new {} has been created", user));
return updateUserStat(createdUser, exchange, chain, uri);
}
};
}
private Mono<Void> updateUserStat(Mono<User> userMono, ServerWebExchange exchange, WebFilterChain chain, String uri) {
return userMono
.doOnNext(user -> exchange.getAttributes().put("_token", user.getToken()))
.doOnNext(u -> {
String token = exchange.getAttribute("_token");
log.info("[i] Web Filter: token attribute has been set to '{}'", token);
})
.flatMap(user -> pageViewRepo.save(new PageView(uri)).flatMap(user::addPageView).flatMap(userRepo::save))
.doOnNext(user -> {
int numberOfPages = 0;
List<PageView> pageViews = user.getPageViews();
if (pageViews != null) {
numberOfPages = pageViews.size();
}
log.info("[i] Web Filter: {} has been updated. Number of pages: {}", user, numberOfPages);
})
.then(chain.filter(exchange));
}
Этот код дает следующие результаты:
1) Токен отсутствует: создайте нового пользователя, создайте просмотр страницы, обновите нового пользователя, передайте запрос контроллеру
2019-01-20 14:39:10.033 [ctor-http-nio-3] : [i] Web Filter: received the request: http://localhost:8080/users?test=784
2019-01-20 14:39:10.110 [ctor-http-nio-3] : Inserting Document containing fields: [token, _class] in collection: user
2019-01-20 14:39:10.206 [ntLoopGroup-2-2] : [i] Web Filter: a new User(id=5c446bee24c86426ac6c0ae5, token=fba944cd-decb-4923-9757-724da5a60061) has been created
2019-01-20 14:39:10.212 [ntLoopGroup-2-2] : [i] Web Filter: token attribute has been set to 'fba944cd-decb-4923-9757-724da5a60061'
2019-01-20 14:39:11.227 [ parallel-1] : Inserting Document containing fields: [URL, createdDate, _class] in collection: pageView
2019-01-20 14:39:11.242 [ntLoopGroup-2-2] : Saving Document containing fields: [_id, token, pageViews, _class]
2019-01-20 14:39:11.256 [ntLoopGroup-2-2] : [i] Web Filter: User(id=5c446bee24c86426ac6c0ae5, token=fba944cd-decb-4923-9757-724da5a60061) has been updated. Number of pages: 1
2019-01-20 14:39:11.289 [ntLoopGroup-2-2] : [i] Controller: handling 'get all users' request. Token attribute is 'fba944cd-decb-4923-9757-724da5a60061'
2019-01-20 14:39:11.369 [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class io.github.cepr0.demo.User in collection: user
2) Токен присутствует: найдите существующего пользователя, создайте просмотр страницы, обновите пользователя, передайте запрос контроллеру
2019-01-20 14:51:21.983 [ctor-http-nio-3] : [i] Web Filter: received the request: http://localhost:8080/users?test=538
2019-01-20 14:51:22.074 [ctor-http-nio-3] : Created query Query: { "token" : "b613b810-cc36-4961-ad2e-db44f52cd2dd" }, Fields: { }, Sort: { }
2019-01-20 14:51:22.092 [ctor-http-nio-3] : find using query: { "token" : "b613b810-cc36-4961-ad2e-db44f52cd2dd" } fields: Document{{}} for class: class User in collection: user
2019-01-20 14:51:22.102 [ntLoopGroup-2-2] : [i] Web Filter: User(id=5c434c2eb338ac3530cbd56d, token=b613b810-cc36-4961-ad2e-db44f52cd2dd) has been found
2019-01-20 14:51:22.102 [ntLoopGroup-2-2] : [i] Web Filter: token attribute has been set to 'b613b810-cc36-4961-ad2e-db44f52cd2dd'
2019-01-20 14:51:23.103 [ parallel-2] : Inserting Document containing fields: [URL, createdDate, _class] in collection: pageView
2019-01-20 14:51:23.115 [ntLoopGroup-2-2] : Saving Document containing fields: [_id, token, pageViews, _class]
2019-01-20 14:51:23.117 [ntLoopGroup-2-2] : [i] Web Filter: User(id=5c434c2eb338ac3530cbd56d, token=b613b810-cc36-4961-ad2e-db44f52cd2dd) has been updated. Number of pages: 13
2019-01-20 14:51:23.118 [ntLoopGroup-2-2] : [i] Controller: handling 'get all users' request. Token attribute is 'b613b810-cc36-4961-ad2e-db44f52cd2dd'
2019-01-20 14:51:23.119 [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user
3) Токен присутствует, но пользователь не найден: передать запрос контроллеру
2019-01-20 14:52:41.842 [ctor-http-nio-3] : [i] Web Filter: received the request: http://localhost:8080/users?test=513
2019-01-20 14:52:41.844 [ctor-http-nio-3] : Created query Query: { "token" : "-b613b810-cc36-4961-ad2e-db44f52cd2dd" }, Fields: { }, Sort: { }
2019-01-20 14:52:41.845 [ctor-http-nio-3] : find using query: { "token" : "-b613b810-cc36-4961-ad2e-db44f52cd2dd" } fields: Document{{}} for class: class User in collection: user
2019-01-20 14:52:41.850 [ntLoopGroup-2-2] : [i] Controller: handling 'get all users' request. Token attribute is 'null'
2019-01-20 14:52:41.850 [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user
Демо: sb-reactive-filter-demo (ветвь: update-user-in-web-filter)
TL; DR: вместо использования subscribe() для вызова сохранения в mongoDB (который отправляет его в параллельный поток, в котором у вас нет контроля), позвольте оператору flatMap подписаться на save, что заставляет внешнюю цепочку ждать завершения сохранения.
@RajeshJAdvani Я использую flatMap, а не subscribe: .flatMap(user -> pageViewRepo.save(new PageView(uri)).flatMap(user::addPageView).flatMap(userRepo::save)) Что вы имеете в виду? ..
Извините, я не поправлял вас. Я пытался объяснить основную проблему, которую вы исправили, для людей, которые искали резюме.
Правильно ли я вас понял и вам просто нужно запустить эти длинные операции с базой данных параллельно, чтобы не блокировать фильтр и, собственно, сам запрос?