为响应式序列添加上下文
目录
从命令式编程思维转向响应式编程思维时，遇到的一大技术挑战在于如何处理线程。
与您过去习惯的方式不同，在响应式编程中，您可以用一个 Thread 来处理若干条大致同时运行的异步序列（实际上是以非阻塞的步调协同推进的）。执行过程也可以很轻松地、并且经常地从一个线程跳到另一个线程。
这种安排对于那些依赖线程模型更“稳定”的特性（例如 ThreadLocal）的开发者来说尤其难以适应。由于 ThreadLocal 让您把数据与线程关联起来，它在响应式上下文中就变得很难使用。因此，依赖 ThreadLocal 的类库与 Reactor 配合使用时，往轻里说会带来新的挑战，往坏里说会工作失常甚至彻底失败。用 Logback 的 MDC 来存储并打印关联 ID，就是这种情况的一个典型例子。
对于 ThreadLocal 的惯常变通办法是：把上下文数据 C 随着业务数据 T 一起在序列中传递，例如使用 Tuple2<T, C>。这样做既不好看，又把一个正交的关注点（上下文数据）泄漏进了您的方法签名和 Flux 签名。
从 3.1.0 版本起，Reactor 提供了一项高级特性，它在某种程度上类似于 ThreadLocal，但作用对象是 Flux 或 Mono，而不是 Thread。这项特性叫做 Context。
为了说明它长什么样，下面这个示例同时对 Context 进行了读取和写入：
String key = "message";
Mono<String> r = Mono.just("Hello")
.flatMap(s -> Mono.deferContextual(ctx ->
Mono.just(s + " " + ctx.get(key))))
.contextWrite(ctx -> ctx.put(key, "World"));
StepVerifier.create(r)
.expectNext("Hello World")
.verifyComplete();
接下来几节将介绍 Context 及其用法，读完之后您就能理解上面这个示例了。
这是一项更面向类库开发者的高级特性。它要求您充分理解Subscription 的生命周期，面向的是那些负责订阅的类库。
1. Context API
Context 是一个让人联想到 Map 的接口。它存储键值对，并让您按键取出自已存入的值。它还有一个简化版本，只暴露读取方法，即 ContextView。具体来说：
-
键和值都是
Object类型，因此一个Context（以及ContextView）实例可以容纳来自不同类库、不同来源、彼此差异极大的任意数量的值。 -
Context是不可变的。它虽然暴露了put、putAll等写入方法，但这些方法会产生一个新实例。 -
自 3.4.0 起，有了一个连这些写入方法都不暴露的只读 API 超接口
ContextView。 -
可以用
hasKey(Object key)检查某个键是否存在。 -
使用
getOrDefault(Object key, T defaultValue)取值（会转型为T）；如果Context实例中没有该键，则回退到默认值。 -
使用
getOrEmpty(Object key)得到一个Optional<T>（Context实例会尝试把存储的值转型为T）。 -
使用
put(Object key, Object value)存入一个键值对，并返回一个新的Context实例。也可以用putAll(ContextView)把两个上下文合并成一个新的。 -
使用
delete(Object key)移除与某个键关联的值，并返回一个新的Context。
创建 Context 时，可以使用静态方法 Context.of 创建预填值的 Context 实例，最多可包含五个键值对。它们分别接受 2、4、6、8 或 10 个 Object 实例，每两个 Object 构成一对要加入 Context 的键值对。
另外，也可以用 Context.empty() 创建一个空的 Context。
2. 把 Context 绑定到 Flux 并写入
要让 Context 发挥作用，必须把它绑定到某一条具体序列上，并让链路中的每个操作符都能访问它。请注意，操作符必须是 Reactor 原生操作符，因为 Context 是 Reactor 特有的。
实际上，一个 Context 是绑定到链路中每一个 Subscriber 上的。它借助 Subscription 的传播机制，从最终的 subscribe 开始、沿链路自下而上地让每个操作符都能访问到自己。
要填充 Context（这只能在订阅时进行），需要使用 contextWrite 操作符。
contextWrite(ContextView) 会把您提供的 ContextView 与下游传来的 Context 合并（请记住，Context 是从链路底部向顶部传播的）。这通过一次 putAll 调用来完成，结果是为上游生成一个新的 Context。
您也可以使用更高级的 contextWrite(Function<Context, Context>)。它会收到下游传来的 Context 的一个副本，让您按需放入或删除值，并返回要使用的新 Context。您甚至可以决定返回一个完全不同的实例，但这非常不推荐（这样做可能影响依赖该 Context 的第三方类库）。
3. 通过 ContextView 读取 Context
填充好 Context 之后，您可能想在运行时窥探其中的内容。大多数情况下，把信息放进 Context 的责任在最终用户一侧，而利用这些信息的责任在第三方类库一侧——因为这类类库通常位于客户端代码的上游。
面向读取的操作符会在一条操作符链中暴露 Context 的 ContextView，从而从中获取数据：
-
要从一个类似源的操作符中访问上下文，使用
deferContextual工厂方法 -
要从操作符链的中间访问上下文，使用
transformDeferredContextual(BiFunction) -
另外，当处理的是一条内部序列时（例如在
flatMap内部），可以用Mono.deferContextual(Mono::just)把ContextView物化出来。不过通常您可能更想直接在 defer 的 lambda 里做有意义的工作，例如Mono.deferContextual(ctx -> doSomethingAsyncWithContextData(v, ctx.get(key)))，其中v是正被 flatMap 的那个值。
为了在数据流经管道的过程中从 Context 读取内容、同时又不会误导用户以为此时可以写入，上面这些操作符只暴露 ContextView。万一需要使用某个仍要求 Context 的 API，可以用 Context.of(contextView) 进行转换。
4. Context 简单示例
本节的示例旨在帮助您更好地理解使用 Context 时的一些注意事项。
我们先回过头，稍微详细地看看引言中的那个简单示例，如下例所示：
String key = "message";
Mono<String> r = Mono.just("Hello")
.flatMap(s -> Mono.deferContextual(ctx ->
Mono.just(s + " " + ctx.get(key)))) 2
.contextWrite(ctx -> ctx.put(key, "World")); 1
StepVerifier.create(r)
.expectNext("Hello World") 3
.verifyComplete();
这条操作符链以一次
contextWrite(Function)调用结束，它把"World"以键"message"放入Context。我们对源元素做
flatMap，用Mono.deferContextual()物化出ContextView，直接取出与"message"关联的数据，并把它与原始单词拼接。得到的
Mono<String>发出"Hello World"。
上面的编号与实际的代码行顺序不一致，这不是笔误，它代表的是执行顺序。尽管 contextWrite 是链上的最后一段，但它却是最先被执行的（这是由它“发生在订阅时”的性质、以及订阅信号自下而上流动这一事实决定的）。
在您的操作符链中，“写入 Context 的位置”与“读取 Context 的位置”之间的相对先后关系很重要。Context 是不可变的，它的内容只能被位于它上方的操作符看到，如下例所示：
String key = "message";
Mono<String> r = Mono.just("Hello")
.contextWrite(ctx -> ctx.put(key, "World")) 1
.flatMap( s -> Mono.deferContextual(ctx ->
Mono.just(s + " " + ctx.getOrDefault(key, "Stranger")))); 2
StepVerifier.create(r)
.expectNext("Hello Stranger") 3
.verifyComplete();
Context写在了链中过于靠上的位置。结果，在
flatMap中找不到与我们的键关联的值，于是改用默认值。因此得到的
Mono<String>发出的是"Hello Stranger"。
类似地，如果多次向 Context 写入同一个键，这些写入之间的相对先后也很重要。读取 Context 的操作符看到的，是离它最近（在它下方）的那次写入所设置的值，如下例所示：
String key = "message";
Mono<String> r = Mono
.deferContextual(ctx -> Mono.just("Hello " + ctx.get(key)))
.contextWrite(ctx -> ctx.put(key, "Reactor")) 1
.contextWrite(ctx -> ctx.put(key, "World")); 2
StepVerifier.create(r)
.expectNext("Hello Reactor") 3
.verifyComplete();
对键
"message"的一次写入尝试。对键
"message"的又一次写入尝试。deferContextual只看到了离它最近（且在它下方）设置的值："Reactor"。
在上例中，订阅期间 Context 先被填入 "World"。随后订阅信号继续向上游移动，又发生一次写入。这产生了第二个不可变的 Context，其值为 "Reactor"。此后数据才开始流动。deferContextual 看到的是离它最近的那个 Context，也就是我们这个带有 "Reactor" 值的第二个 Context（以 ContextView 的形式暴露给用户）。
您也许会问：Context 是否会随着数据信号一起传播？如果是这样，那么在这两次写入之间再插入一个 flatMap，就应该用到来自上方那个 Context 的值。但事实并非如此，如下例所示：
String key = "message";
Mono<String> r = Mono
.deferContextual(ctx -> Mono.just("Hello " + ctx.get(key))) 3
.contextWrite(ctx -> ctx.put(key, "Reactor")) 2
.flatMap( s -> Mono.deferContextual(ctx ->
Mono.just(s + " " + ctx.get(key)))) 4
.contextWrite(ctx -> ctx.put(key, "World")); 1
StepVerifier.create(r)
.expectNext("Hello Reactor World") 5
.verifyComplete();
这是最先发生的写入。
这是第二发生的写入。
最上方的上下文读取看到的是第二次写入。
flatMap把最初读取得到的结果与第一次写入的值拼接起来。这个
Mono发出"Hello Reactor World"。
原因在于：Context 是与 Subscriber 关联在一起的，每个操作符都是通过向自己下游的 Subscriber 请求来访问 Context 的。
最后一个有意思的传播情形，是在 flatMap 内部也写入 Context，如下例所示：
String key = "message";
Mono<String> r = Mono.just("Hello")
.flatMap( s -> Mono
.deferContextual(ctxView -> Mono.just(s + " " + ctxView.get(key)))
)
.flatMap( s -> Mono
.deferContextual(ctxView -> Mono.just(s + " " + ctxView.get(key)))
.contextWrite(ctx -> ctx.put(key, "Reactor")) 1
)
.contextWrite(ctx -> ctx.put(key, "World")); 2
StepVerifier.create(r)
.expectNext("Hello World Reactor")
.verifyComplete();
这个
contextWrite不会影响其flatMap之外的任何地方。这个
contextWrite影响的是主序列的Context。
在上例中，最终发出的值是 "Hello World Reactor" 而不是 "Hello Reactor World"，因为写入 "Reactor" 的那个 contextWrite 是作为第二个 flatMap 的内部序列的一部分执行的。因此，它对主序列不可见、也不会传播到主序列，第一个 flatMap 也就看不到它。在 flatMap 这类会创建中间内部序列的操作符中，传播机制与不可变性共同把 Context 隔离开了。
5. 完整示例
现在我们来看一个更贴近现实的例子：一个类库从 Context 中读取信息。这是一个响应式 HTTP 客户端，它以一个 Mono<String> 作为 PUT 请求的数据源，同时还会查找某个特定的 Context 键，以便把一个关联 ID 添加到请求头中。
从用户的角度看，它是这样调用的：
doPut("www.example.com", Mono.just("Walter"))
要传播一个关联 ID，则像这样调用：
doPut("www.example.com", Mono.just("Walter"))
.contextWrite(Context.of(HTTP_CORRELATION_ID, "2-j3r9afaf92j-afkaf"))
如上两段代码所示，用户代码用 contextWrite 往 Context 中填入一个 HTTP_CORRELATION_ID 键值对。该操作符的上游是 HTTP 客户端类库返回的一个 Mono<Tuple2<Integer, String>>（对 HTTP 响应的一种简化表示）。因此，它实际上是把信息从用户代码传递给了类库代码。
下面这个示例从类库的角度展示了一段模拟代码：它读取上下文，如果能找到关联 ID 就“增强请求”：
static final String HTTP_CORRELATION_ID = "reactive.http.library.correlationId";
Mono<Tuple2<Integer, String>> doPut(String url, Mono<String> data) {
Mono<Tuple2<String, Optional<Object>>> dataAndContext =
data.zipWith(Mono.deferContextual(c -> 1
Mono.just(c.getOrEmpty(HTTP_CORRELATION_ID))) 2
);
return dataAndContext.<String>handle((dac, sink) -> {
if (dac.getT2().isPresent()) { 3
sink.next("PUT <" + dac.getT1() + "> sent to " + url +
" with header X-Correlation-ID = " + dac.getT2().get());
}
else {
sink.next("PUT <" + dac.getT1() + "> sent to " + url);
}
sink.complete();
})
.map(msg -> Tuples.of(200, msg));
}
通过
Mono.deferContextual物化出ContextView，然后…在 defer 内部，取出关联 ID 键对应的值，作为一个
Optional。如果该键存在于上下文中，就把这个关联 ID 用作请求头。
这段类库代码把数据 Mono 与 Mono.deferContextual(Mono::just) 做了 zip。这给了类库一个 Tuple2<String, ContextView>，而这个上下文中包含来自下游的 HTTP_CORRELATION_ID 条目（因为它位于通往订阅者的直接路径上）。
然后类库代码用 map 取出该键对应的 Optional<String>；如果该条目存在，就把传入的关联 ID 用作 X-Correlation-ID 请求头。最后这部分由 handle 来模拟。
用于验证类库代码确实使用了关联 ID 的完整测试，可以写成如下形式：
@Test
public void contextForLibraryReactivePut() {
Mono<String> put = doPut("www.example.com", Mono.just("Walter"))
.contextWrite(Context.of(HTTP_CORRELATION_ID, "2-j3r9afaf92j-afkaf"))
.filter(t -> t.getT1() < 300)
.map(Tuple2::getT2);
StepVerifier.create(put)
.expectNext("PUT <Walter> sent to www.example.com" +
" with header X-Correlation-ID = 2-j3r9afaf92j-afkaf")
.verifyComplete();
}