【发布时间】：2018-10-21 19:03:03
【问题描述】：
我有一系列股票代码和一系列股票价格。每次我得到一个股票代码（它保证是唯一的）时，我都需要跨越一个 100 毫秒的窗口并从股票价格序列中处理股票的价格。如果这 100 毫秒内的序列中缺少股票价格，我需要处理没有价格的股票。
大理石图更好地可视化需求：
Stock : -S1--S2--
Price : ---P1-P2-
Result: -S1---S2P2-
所以股票1 进来了，100 毫秒的窗口被跨越，但是没有价格的符号，因此结果应该只是股票1 (S1)。
然后股票2 进来，再次跨越100 毫秒的窗口，还没有股票2 的价格。然而，在 100 毫秒窗口关闭之前，我们得到股票 2 (P2) 的价格，因此结果是股票 2 及其价格 (S2P2)。
价格有可能是随机出现的，所以不能对价格顺序做出假设。
我已经看到了这个related SO question，但不能让它工作。
我正在尝试使用GroupJoin
stockSubject
.GroupJoin(
stockPriceSubject,
stock => Observable.Timer(TimeSpan.FromMilliseconds(100)),
price => Observable.Never<Unit>(),
(stock, stockPrice) =>
{
var stockPrices = stockPrice.Where(sprice => sprice.Stock.Equals(stock))
.FirstOrDefaultAsync()
.DefaultIfEmpty();
return (Stock: stock, StockPrices: stockPrices);
})
.Subscribe(async tuple => WriteLine($"{tuple.Stock} | {(await tuple.StockPrices)?.Price ?? 'N'}"));
这不起作用，因为它会降低一些价格（发生不确定性，因此无法真正找出问题所在）。
我在工作时尝试过的另一种方法看起来并不理想
stockSubject
.Subscribe(stock =>
{
stockPriceSubject
.Buffer(TimeSpan.FromMilliseconds(100))
.Take(1)
.Subscribe(bufferedPrices =>
{
var stockPrice = bufferedPrices.FirstOrDefault(price => price.Stock.Equals(stock));
if (stockPrice == null)
{
Console.WriteLine($"{stock} is w/o price");
return;
}
Console.WriteLine($"{stockPrice}");
});
});
我真正不喜欢这个的一件事是，每次有新股票订阅缓冲价格时，我都会放弃订阅。
知道使用 Rx 实现此场景的最佳方法是什么吗？
股票和股价的相关类
sealed class Stock : IEquatable<Stock>
{
public Stock(string symbol)
{
Symbol = symbol;
}
public string Symbol { get; }
public override string ToString() =>
$"Stock[{Symbol}]";
// IEquatable implementation is emitted for the sake of brevity
}
sealed class StockPrice
{
public StockPrice(Stock stock, decimal price)
{
Stock = stock;
Price = price;
}
public Stock Stock { get; }
public decimal Price { get; }
public override string ToString() =>
$"{Stock} is traded @ {Price}";
}
按要求编辑添加测试数据代码生成器
每 10 毫秒将新股票推送到股票序列（MSFT -> GOOG -> APPL）。
每 20 毫秒将新价格推送到价格序列 (APPL -> GOOG)。
1 秒后，股票 MSFT 的价格被推送到价格序列。
预期输出：
一旦 MSFT 被推送到股票序列，100 毫秒的价格窗口打开......在 100 毫秒内，MSFT 没有价格被推送到价格序列，因此MSFT 股票应不带价格处理（在结果集中价格为空/null）
GOOG被推送到股票序列后，再次打开100毫秒窗口，这一次GOOG股票在100毫秒内有一个价格，因此 GOOG库存应按价格处理（1500万）。
最后是 APPL - 这里的预期输出与 MSFT 相同...因为在 100 毫秒内没有为 APPL 推送价格窗口，因为它被推送到股票序列，它应该无价处理。在此，之前发布了 APPL 股票价格这一事实不应影响任何事情。
var stockSubject = new Subject<Stock>();
var stockPriceSubject = new Subject<StockPrice>();
Observable
.Interval(TimeSpan.FromMilliseconds(10))
.Take(3)
.Subscribe(_ =>
{
switch (_)
{
case 0:
{
var stock = new Stock("MSFT");
stockSubject.OnNext(stock);
break;
}
case 1:
{
var stock = new Stock("GOOG");
stockSubject.OnNext(stock);
break;
}
case 2:
{
var stock = new Stock("APPL");
stockSubject.OnNext(stock);
break;
}
}
});
Observable
.Interval(TimeSpan.FromMilliseconds(20))
.Take(3)
.Subscribe(_ =>
{
switch (_)
{
case 0:
{
var stockPrice = new StockPrice(new Stock("APPL"), 10m);
stockPriceSubject.OnNext(stockPrice);
break;
}
case 1:
{
var stockPrice = new StockPrice(new Stock("GOOG"), 15m);
stockPriceSubject.OnNext(stockPrice);
break;
}
}
});
Observable
.Timer(TimeSpan.FromSeconds(1))
.Subscribe(_ =>
{
var stockPrice = new StockPrice(new Stock("MSFT"), 20m);
stockPriceSubject.OnNext(stockPrice);
});
【问题讨论】：
标签： c# system.reactive