如何做不同的油门在RxJS

Question

我已经花了几天时间，但找不到在RxJS中实现“独特节流”的方法。

假设每个事件以4个破折号完成，一个“不同的油门”将执行如下操作：

-①-②-①---------①-----|-

[明显的油门]

-①-②-------------①-----|-

如何使用现有的RxJS操作符构建“独特的节流阀”？

Answer 1

distinct和throttle在物品挑选方面有两个不同的特征distinct将选择第一项，而throttle将选择最后一项。

有时，您希望保持限制的行为。

假设流是：聊天消息编辑事件，其中包含更新的文本。用户可以在节流周期内多次编辑特定消息
您希望确保始终保留每条消息的最新版本（在不同当前消息的编辑流中）。

下面是一个可能的解决方案

const source$ = from([
{id:1,content:"1a"}, 
{id:1,content:"1b"}, 
{id:1,content:"1c"}, 
{id:2,content:"2a"}, 
{id:2,content:"2b"}, 
{id:3,content:"3a"},
{id:3,content:"3b"},
{id:1,content:"1d"},
{id:1,content:"1e"},
{id:4,content:"4a"},
{id:4,content:"4b"},
{id:4,content:"4c"},
{id:4,content:"4e"},
{id:4,content:"4f"},
{id:3,content:"3c"},
{id:3,content:"3d"},
{id:3,content:"3e"}
]).pipe(concatMap((el)=> of(el).pipe(delay(500)) ));



const distinctThrottle = (throttleTime, keySelector)=>
    pipe(bufferTime(throttleTime),
          concatMap((arr)=>from(arr.reverse()).pipe(distinct(keySelector))  
        )) ;


let throttledStream = source$.pipe(distinctThrottle(1550, ({id})=>id));

throttledStream.subscribe(console.log);

Answer 2

您可以使用groupBy按值分隔通知，然后可以应用throttleTime，然后可以使用mergeMap合并分组的观察值。这样地：

const { Subject } = rxjs;
const { groupBy, mergeMap, throttleTime } = rxjs.operators;

const source = new Subject();
const result = source.pipe(
  groupBy(value => value),
  mergeMap(grouped => grouped.pipe(
    throttleTime(400)
  ))
);

result.subscribe(value => console.log(value));

setTimeout(() => source.next(1), 100);
setTimeout(() => source.next(2), 300);
setTimeout(() => source.next(1), 400);
setTimeout(() => source.next(1), 900);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@6/bundles/rxjs.umd.min.js"></script>