我目前正在使用 SpringBoot 2、spring-boot-starter-webflux 在 netty 和 jOOQ 上开发应用程序。
下面是我经过数小时的研究和 stackoverflow 搜索得出的代码。我内置了很多
记录以查看哪个线程上发生了什么。
用户 Controller :
@RequestMapping(value = "/user", method = RequestMethod.POST)
public Mono<ResponseEntity<Integer>> createUser(@RequestBody ImUser user) {
return Mono.just(user)
.map(it -> {
logger.debug("Receiving request on thread: " + Thread.currentThread().getName());
return it;
})
.map(userService::create)
.map(it -> {
logger.debug("Sending response on thread: " + Thread.currentThread().getName());
return ResponseEntity.status(HttpStatus.CREATED).body(it);
})
.mapError(DuplicateKeyException.class, e -> new SomeSpecialException(e.getMessage(), e));
}
用户服务:
public int create(ImUser user) {
return Mono.just(user)
.subscribeOn(Schedulers.elastic())
.map(u -> {
logger.debug("UserService thread: " + Thread.currentThread().getName());
return imUserDao.insertUser(u);
})
.block();
}
用户道:
@Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.READ_COMMITTED, rollbackFor = Exception.class)
public int insertUser(ImUser user) {
logger.debug("Insert DB on thread: " + Thread.currentThread().getName());
return dsl.insertInto(IM_USER,IM_USER.VERSION, IM_USER.FIRST_NAME, IM_USER.LAST_NAME, IM_USER.BIRTHDATE, IM_USER.GENDER)
.values(1, user.getFirstName(), user.getLastName(), user.getBirthdate(), user.getGender())
.returning(IM_USER.ID)
.fetchOne()
.getId();
}
代码按预期工作,“接收请求”和“发送响应”都在同一个线程上运行 (reactor-http-server-epoll-x) 而阻塞代码(对 imUserDao.insertUser(u) 的调用)在弹性调度程序线程 (elastic-x) 上运行。 事务绑定(bind)到调用注释方法的线程(即 elastic-x),因此按预期工作(我已经测试过 为简单起见,这里未发布其他方法)。
这是一个日志示例:
20:57:21,384 DEBUG admin.UserController| Receiving request on thread: reactor-http-server-epoll-7
20:57:21,387 DEBUG admin.UserService| UserService thread: elastic-2
20:57:21,391 DEBUG admin.ExtendedUserDao| Insert DB on thread: elastic-2
20:57:21,393 DEBUG tools.LoggerListener| Executing query
...
20:57:21,401 DEBUG tools.StopWatch| Finishing : Total: 9.355ms, +3.355ms
20:57:21,409 DEBUG admin.UserController| Sending response on thread: reactor-http-server-epoll-7
我已经研究响应式编程很长时间了,但从来没有完全接触过任何响应式编程。既然我是,我想知道我是否做对了。 所以这是我的问题:
1. 上面的代码是处理传入 HTTP 请求、查询数据库然后响应的好方法吗? 请忽略我为了理智而内置的 logger.debug(...) 调用 :) 我有点希望有一个 Flux< ImUser> 作为 Controller 方法的参数,从某种意义上说,我有一连串的多个潜在请求 这将在某个时候出现,并且将以相同的方式处理。相反,我发现的示例会在每次收到请求时创建一个 Mono.from(...);。
2. 在 UserService ( Mono.just(user) ) 中创建的第二个 Mono 感觉有些尴尬。我知道我需要开始一个新的流才能 在弹性调度程序上运行代码,但没有执行此操作的运算符(operator)吗?
3. 从代码的编写方式来看,我了解到 UserService 中的 Mono 将被阻塞,直到数据库操作完成, 但是服务于请求的原始流没有被阻塞。这个对吗?
4. 我计划将 Schedulers.elastic() 替换为并行调度程序,我可以在其中配置工作线程的数量。这个想法是最大工作线程数应该与最大数据库连接数相同。 当 Scheduler 中的所有工作线程都忙时会发生什么?那是背压进入的时候吗?
5. 我最初希望在我的 Controller 中包含以下代码:
return userService.create(user)
.map(it -> ResponseEntity.status(HttpStatus.CREATED).body(it))
.mapError(DuplicateKeyException.class, e -> new SomeSpecialException(e.getMessage(), e));
但我没能做到这一点并让事情在正确的线程中运行。有什么方法可以在我的代码中实现这一点吗?
如有任何帮助,我们将不胜感激。谢谢!
最佳答案
服务和 Controller
您的服务阻塞的事实是有问题的,因为在 Controller 中您从 map
内部调用一个阻塞方法,该方法不会在单独的线程上移动。这有可能阻止所有 Controller 。
您可以做的是从 UserService#create
返回一个 Mono
(删除末尾的 block()
)。由于该服务确保 Dao 方法调用是隔离的,因此问题较少。从那里开始,无需在 Controller 中执行 Mono.just(user)
:只需直接在生成的 Mono 上调用创建并开始链接运算符:
@RequestMapping(value = "/user", method = RequestMethod.POST)
public Mono<ResponseEntity<Integer>> createUser(@RequestBody ImUser user) {
//this log as you saw was executed in the same thread as the controller method
logger.debug("Receiving request on thread: " + Thread.currentThread().getName());
return userService.create(user)
.map(it -> {
logger.debug("Sending response on thread: " + Thread.currentThread().getName());
return ResponseEntity.status(HttpStatus.CREATED).body(it);
})
.mapError(DuplicateKeyException.class, e -> new SomeSpecialException(e.getMessage(), e));
}
记录
请注意,如果您想记录某些内容,有几个比执行 map
并返回 it
更好的选择:
doOnNext
方法专为此量身定制:对 react 信号之一使用react(在本例中,onNext
:发出一个值)并执行一些非-变异 Action ,使输出序列与源序列完全相同。 doOn 的“副作用”可以写入控制台或递增统计计数器,例如……还有 doOnComplete、doOnError、doOnSubscribe、doOnCancel 等……log
只是记录其上方序列中的所有事件。它将检测您是否使用 SLF4J,如果是,则在 DEBUG 级别使用配置的记录器。否则它将使用 JDK 日志记录功能(因此您还需要将其配置为显示 DEBUG 级别日志)。
关于交易的一句话或者更确切地说任何依赖于ThreadLocal
的事情
ThreadLocal 和线程粘性在响应式(Reactive)编程中可能会出现问题,因为底层执行模型在整个序列中保持不变的保证较少。 Flux
可以在多个步骤中执行,每个步骤在不同的 Scheduler
(以及线程或线程池)中。即使在特定步骤,一个值也可以由底层线程池的线程 A 处理,而下一个值稍后到达,将在线程 B 上处理。
在这种情况下,依赖 Thread Local 就不那么简单了,我们目前正在积极致力于提供更适合 react 世界的替代方案。
您创建连接池大小的池的想法很好,但不一定足够,事务流量可能会使用多个线程,因此可能会污染某些线程。
当线程池用完线程时会发生什么
如果您使用特定的 Scheduler
来隔离阻塞行为,就像这里一样,一旦它用完线程,它就会抛出一个 RejectedExecutionException
。
关于java - 了解 Spring 的 Web 响应式框架,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43130036/