为什么 Observable.Merge() 在源 observable 上调用 Select()？

问题描述

目前正在试验 Observable.Interval，发现了一个我自己无法解释的行为。

export const addOrderItems = asyncHandler(async (req,res) => {
  const {
    orderItems,mobile,paidamount,discountAmount,totalPrice,} = req.body

  if (orderItems && orderItems.length === 0) {
    res.status(400)
    throw new Error('No order items')
  } else {
    const order = new OrderModel({
      orderItems,user: req.user._id,})

    // Make sure all products are available at first
    const products = []
    for (const item of order.orderItems) {
      const product = await ProductModel.findById(item.product)
      if (item.qty <= product.countInStock) {
        product.countInStock = product.countInStock - Number(item.qty)
        products.push(product);
        // Not save the product here,you can not make sure that the order is success
      } else {
        // One of them is out of stock,stop immediate
        res.status(400)
        throw new Error(`The product ${product.name} is out of stock`) // product.name ?
      }
    }

    // Save order at first,You need read about transaction for these actions
    const createdOrder = await order.save()

    // .map with async callback function will return an array of promise
    await Promise.all(products.map(async (pro) => await pro.save()))

    res.status(201).json(createdOrder)
  }
})

我在测试这个东西时的经验是，一旦我使用 _rate = Observable.Interval(TimeSpan.FromMilliseconds(250),scheduler).Select(_ => CalculaterateValue()); _thickness = Observable.Merge(_rate).Select(CalculateThicknessValue); ，CalculaterateValue() 就会被调用两次。 直到今天，我的理解是，只要 Observable.Merge _rate 上有新值，就会使用该新速率值调用。但不应再次从源 observable 调用 CalculateThicknessValue 部分。

使用 Select 或某种 Zip 时也会发生同样的情况。

测试代码如下：

Combine

每次调用 TestScheduler scheduler = new TestScheduler(); long ticks = TimeSpan.FromMilliseconds(250).Ticks; scheduler.AdvanceBy(ticks); ... 都会触发 AdvanceBy 两次。

那么为什么会出现这种行为，我该怎么做才能阻止它？

---

CalculaterateValue 和 CalculaterateValue 是返回 CalculateThicknessValue 的简单方法。 两个 observable 也是 double 类型。

解决方法

通过将 .Publish().RefCount() 添加到 observable 来解决。