blocks|key|2208523|text|您可以简单地使用rxjs间隔|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|2208524|import+{+interval+}+from+'rxjs';

const+seconds+=+3600;
const+source+=+interval(seconds+*+1000);

source.subscribe(()=>+{
++//+make+api+request+here
});|code-block|syntax|javascript|2208525|entityMap|0|LINK|mutability|MUTABLE|url|https://www.learnrxjs.io/learn-rxjs/operators/creation/interval^0|8|6|0|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@$A|R|B|S|1|T]]|C|$]]|$1|D|3|E|5|F|7|U|8|@]|9|@]|C|$G|H]]|$1|I|3|-4|5|6|7|V|8|@]|9|@]|C|$]]]|J|$K|$5|L|M|N|C|$O|P]]]]

You can simply use <a href="https://www.learnrxjs.io/learn-rxjs/operators/creation/interval" rel="nofollow noreferrer">rxjs interval</a>
<pre><code>import { interval } from 'rxjs';

const seconds = 3600;
const source = interval(seconds * 1000);

source.subscribe(()=&gt; {
 // make api request here
});
</code></pre>

blocks|key|2633875|text|也许您可以这样做(我没有测试这段代码)：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2633876|import+{+merge,+concat,+of,+timer,+interval+}+from+'rxjs';
import+{+switchMap+}+from+'rxjs/operators';

...

isUpdated()+{
++return+concat(
++++merge(
++++++of(null),+//+make+call+inmediately
++++++timer((60+-+new+Date().getMinutes())+*+60+*+1000),+//+the+next+full+hour+eg.+2:00
++++),
++++interval(60+*+60+*+1000),+//+every+hour
++).pipe(
++++switchMap(()+=>+this.http.get(...)),
++)
})|code-block|syntax|javascript|2633877|concat只有在第一个merge完成后才会订阅interval，这是在下一个完整的小时(例如:+2:00+)，然后每小时发出一次。|offset|length|style|CODE|2633878|@MrkSef有一个很好的想法，它可以简化成这样的东西：|2633879|isUpdated()+{
++return+timer(
++++(60+-+new+Date().getMinutes())+*+60+*+1000,+//+the+next+full+hour+eg.+2:00
++++60+*+60+*+1000+//+every+hour
++).pipe(
++++startWith(null),+//+initial+request
++++switchMap(()+=>+this.http.get(...)),
++)
})|2633880|entityMap|0|LINK|mutability|MUTABLE|url|https://stackoverflow.com/users/13500986/mrk-sef^0|0|0|0|6|C|5|O|8|0|0|7|0|0|0^^$0|@$1|2|3|4|5|6|7|Y|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|Z|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|10|8|@$I|11|J|12|K|L]|$I|13|J|14|K|L]|$I|15|J|16|K|L]]|9|@]|A|$]]|$1|M|3|N|5|6|7|17|8|@]|9|@$I|18|J|19|1|1A]]|A|$]]|$1|O|3|P|5|D|7|1B|8|@]|9|@]|A|$E|F]]|$1|Q|3|-4|5|6|7|1C|8|@]|9|@]|A|$]]]|R|$S|$5|T|U|V|A|$W|X]]]]

Maybe you could do it like this (I didn't test this code):
<pre><code>import { merge, concat, of, timer, interval } from 'rxjs';
import { switchMap } from 'rxjs/operators';

...

isUpdated() {
 return concat(
 merge(
 of(null), // make call inmediately
 timer((60 - new Date().getMinutes()) * 60 * 1000), // the next full hour eg. 2:00
 ),
 interval(60 * 60 * 1000), // every hour
 ).pipe(
 switchMap(() =&gt; this.http.get(...)),
 )
})
</code></pre>
<code>concat</code> will subscribe to <code>interval</code> only after the first <code>merge</code> completes which is at the next full hour eg.: 2:00 and then emit every hour.
<a href="https://stackoverflow.com/users/13500986/mrk-sef">@MrkSef</a> had a good idea that this could be simplified to something like this:
<pre><code>isUpdated() {
 return timer(
 (60 - new Date().getMinutes()) * 60 * 1000, // the next full hour eg. 2:00
 60 * 60 * 1000 // every hour
 ).pipe(
 startWith(null), // initial request
 switchMap(() =&gt; this.http.get(...)),
 )
})
</code></pre>

blocks|key|2208589|text|好吧，所以你想：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2208590|创建两个流:第一个总是每小时滴答一次(60000+ms)，第二个控制每小时发出这些滴答的余量。|ordered-list-item|2208591|const+every60Minutes$+=+interval(60+*+1000);
const+isAllowedToProceedSubject+=+new+Subject<boolean>();+//+subject+is+also+an+Observable+stream|code-block|syntax|javascript|2208592|组合这两个流，并且只允许在第二个流以"true“作为最新值的情况下发出值。|2208593|const+every60MinsIfAllowed$+=+combineLatest([
++isAllowedToProceedSubject,
++every60Minutes$
]).pipe(
++filter(([isAllowed,+intervalTick])+=>+{
++++return+isAllowed;
++}),
++flatMap(()+=>+{
++++return+of(new+Date());+//+replace+with+HTTP+call
++})
)|2208594|能够控制第二个流|2208595|setTimeout(()+=>+{
++isAllowedToProceedSubject.next(true);+//+this+allows+the+values+to+be+emitted
},+2500);

//+or+add+some+click+handler|2208596|工作实例+(但为了更好的清晰度，每2000毫秒发射一次)|offset|length|2208597|entityMap|0|LINK|mutability|MUTABLE|url|https://stackblitz.com/edit/two-streams-with-allowance?file=index.ts^0|0|0|0|0|0|0|0|0|4|0|0^^$0|@$1|2|3|4|5|6|7|13|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|14|8|@]|9|@]|A|$]]|$1|E|3|F|5|G|7|15|8|@]|9|@]|A|$H|I]]|$1|J|3|K|5|D|7|16|8|@]|9|@]|A|$]]|$1|L|3|M|5|G|7|17|8|@]|9|@]|A|$H|I]]|$1|N|3|O|5|D|7|18|8|@]|9|@]|A|$]]|$1|P|3|Q|5|G|7|19|8|@]|9|@]|A|$H|I]]|$1|R|3|S|5|6|7|1A|8|@]|9|@$T|1B|U|1C|1|1D]]|A|$]]|$1|V|3|-4|5|6|7|1E|8|@]|9|@]|A|$]]]|W|$X|$5|Y|Z|10|A|$11|12]]]]

Ok, so you want to:
<ol>
<li>create two streams: first one will always tick every hour (60 000 ms), second - will control the allowance of emitting these hourly ticks</li>
</ol>
<pre><code>const every60Minutes$ = interval(60 * 1000);
const isAllowedToProceedSubject = new Subject&lt;boolean&gt;(); // subject is also an Observable stream
</code></pre>
<ol start="2">
<li>combine those two streams and only allow to emit the values if the second stream has &quot;true&quot; as the latest value</li>
</ol>
<pre><code>const every60MinsIfAllowed$ = combineLatest([
 isAllowedToProceedSubject,
 every60Minutes$
]).pipe(
 filter(([isAllowed, intervalTick]) =&gt; {
 return isAllowed;
 }),
 flatMap(() =&gt; {
 return of(new Date()); // replace with HTTP call
 })
)
</code></pre>
<ol start="3">
<li>be able to control the second stream</li>
</ol>
<pre><code>setTimeout(() =&gt; {
 isAllowedToProceedSubject.next(true); // this allows the values to be emitted
}, 2500);

// or add some click handler
</code></pre>
<a href="https://stackblitz.com/edit/two-streams-with-allowance?file=index.ts" rel="nofollow noreferrer">Working example</a> (but with emitting every 2000ms for better clarity)

blocks|key|2208703|text|我认为要解决这个问题，你需要把它分解成更小的部分。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2208704|首先，我们知道，在某个时候，根据当前时间，我们想知道何时触发下一个调用。如果我们得到一个时间戳，它给出了ms中的当前时间，并且我们希望在下一个小时之前得到ms的数量，那么我们可以这样做：|2208705|const+timeToNextHourInMs+=+(currentTimestampMs)+=>+{
++const+timestampSeconds+=+currentTimestampMs+/+1000;

++const+numberOfSecondsIntoTheCurrentHour+=+timestampSeconds+%25+3600;

++const+numberOfSecondsToTheNextHour+=+3600+-+numberOfSecondsIntoTheCurrentHour;

++return+numberOfSecondsToTheNextHour+*+1000;
};|code-block|syntax|javascript|2208706|我希望变量名足够明确，我不需要评论，但让我知道否则。|2208707|接下来，我们要处理流问题：|2208708|我们想立即触发一个HTTP调用|unordered-list-item|2208709|立即获取发出的值。|2208710|每次新的一小时开始(1:00，2:00，3:00等)，都要做上述所有的事情。|2208711|以下是你如何做到的：|2208712|this.http.get(`/update`).pipe(
++timestamp(),
++switchMap(({+timestamp,+value+})+=>
++++concat(
++++++of(value),
++++++EMPTY.pipe(delay(timeToNextHourInMs(timestamp)))
++++)
++),
++repeat()
);|2208713|让我们来看看上面的逻辑：|2208714|首先，我们立即进行HTTP调用|2208715|完成HTTP调用后，我们将得到当前的时间戳(稍后根据该时间戳确定我们何时要执行下一次调用)|2208716|我们执行一个switchMap，但是由于HTTP调用只返回1值，所以在这个非常特殊的情况下它并不重要。我们也可以用flatMap或concatMap|offset|length|style|CODE|2208717|在switchMap内部，我们首先使用concat发送我们刚刚从HTTP调用中得到的值，但也让这个可观察到的值一直保持到当前our的末尾(通过使用我们先前创建的函数)。|2208718|因此，在当前小时结束时，流将完成。但是，由于我们有了一个retry，一旦流完成，我们将再次订阅它(并且提醒您，流只在新的一个小时开始时才会完成！)|2208719|我建议在这里添加一件事情，但这并不是最初问题的一个要求，那就是要进行一些错误处理，以便当您进行调用时出现问题时，它会在几秒钟后自动重试。否则，想象一下，当轮询开始时，如果您的网络在那个时候不工作5s，那么您的流就会立即出错。|2208720|为此，您可以引用此精彩的回答，并在可重用的自定义操作符中这样做：|2208721|const+RETRY_DELAY+=+2000;
const+MAX_RETRY_FOR_ONE_HTTP_CALL+=+3;

const+automaticRetry+=+()+=>+(obs$)+=>
++obs$.pipe(
++++retryWhen((error$)+=>
++++++error$.pipe(
++++++++concatMap((error,+index)+=>
++++++++++iif(
++++++++++++()+=>+index+>=+MAX_RETRY_FOR_ONE_HTTP_CALL,
++++++++++++throwError(error),
++++++++++++of(error).pipe(delay(RETRY_DELAY))
++++++++++)
++++++++)
++++++)
++++)
++);|2208722|这将在每次重试之间延迟三次可观察到的重试。3次之后，流将抛出最后发出的错误，从而出错。|2208723|现在，我们只需将这个自定义操作符添加到流中：|2208724|this.http.get(`/update`).pipe(
++automaticRetry(),
++timestamp(),
++switchMap(({+timestamp,+value+})+=>
++++concat(
++++++of(value),
++++++EMPTY.pipe(delay(timeToNextHourInMs(timestamp)))
++++)
++),
++repeat()
);|2208725|我还没有对上面的代码进行实际测试，所以请在您这边这样做，并让我知道它是如何进行的。但如果我的逻辑是正确的，下面是事情的发展方向：|2208726|假设你在2:40启动你的应用程序|2208727|立即发出HTTP调用|2208728|你很快就会得到回应|2208729|这条溪流将保持20+be的开放状态。|2208730|3:00，流完成，重试启动:我们执行另一个HTTP调用|2208731|这一次，服务器重新部署，几秒钟内无法使用。|2208732|在内部，流错误，但是由于我们的自定义操作符automaticRetry，它等待了3秒，然后重新尝试了一次，仍然什么也没有。它再等待3秒，这一次很好，结果会被传递到下游|2208733|无限期重复:)|2208734|让我知道是怎么回事|2208735|entityMap|0|LINK|mutability|MUTABLE|url|https://stackoverflow.com/a/44911567/2398593^0|0|0|0|0|0|0|0|0|0|0|0|0|0|6|9|1L|7|1T|9|0|1|9|J|6|0|S|5|0|0|9|5|0|0|0|0|0|0|0|0|0|0|0|0|0|L|E|0|0|0^^$0|@$1|2|3|4|5|6|7|2H|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|2I|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|2J|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|2K|8|@]|9|@]|A|$]]|$1|K|3|L|5|6|7|2L|8|@]|9|@]|A|$]]|$1|M|3|N|5|O|7|2M|8|@]|9|@]|A|$]]|$1|P|3|Q|5|O|7|2N|8|@]|9|@]|A|$]]|$1|R|3|S|5|O|7|2O|8|@]|9|@]|A|$]]|$1|T|3|U|5|6|7|2P|8|@]|9|@]|A|$]]|$1|V|3|W|5|F|7|2Q|8|@]|9|@]|A|$G|H]]|$1|X|3|Y|5|6|7|2R|8|@]|9|@]|A|$]]|$1|Z|3|10|5|O|7|2S|8|@]|9|@]|A|$]]|$1|11|3|12|5|O|7|2T|8|@]|9|@]|A|$]]|$1|13|3|14|5|O|7|2U|8|@$15|2V|16|2W|17|18]|$15|2X|16|2Y|17|18]|$15|2Z|16|30|17|18]]|9|@]|A|$]]|$1|19|3|1A|5|O|7|31|8|@$15|32|16|33|17|18]|$15|34|16|35|17|18]]|9|@]|A|$]]|$1|1B|3|1C|5|O|7|36|8|@$15|37|16|38|17|18]]|9|@]|A|$]]|$1|1D|3|1E|5|6|7|39|8|@]|9|@]|A|$]]|$1|1F|3|1G|5|6|7|3A|8|@]|9|@$15|3B|16|3C|1|3D]]|A|$]]|$1|1H|3|1I|5|F|7|3E|8|@]|9|@]|A|$G|H]]|$1|1J|3|1K|5|6|7|3F|8|@]|9|@]|A|$]]|$1|1L|3|1M|5|6|7|3G|8|@]|9|@]|A|$]]|$1|1N|3|1O|5|F|7|3H|8|@]|9|@]|A|$G|H]]|$1|1P|3|1Q|5|6|7|3I|8|@]|9|@]|A|$]]|$1|1R|3|1S|5|O|7|3J|8|@]|9|@]|A|$]]|$1|1T|3|1U|5|O|7|3K|8|@]|9|@]|A|$]]|$1|1V|3|1W|5|O|7|3L|8|@]|9|@]|A|$]]|$1|1X|3|1Y|5|O|7|3M|8|@]|9|@]|A|$]]|$1|1Z|3|20|5|O|7|3N|8|@]|9|@]|A|$]]|$1|21|3|22|5|O|7|3O|8|@]|9|@]|A|$]]|$1|23|3|24|5|O|7|3P|8|@$15|3Q|16|3R|17|18]]|9|@]|A|$]]|$1|25|3|26|5|O|7|3S|8|@]|9|@]|A|$]]|$1|27|3|28|5|6|7|3T|8|@]|9|@]|A|$]]|$1|29|3|-4|5|6|7|3U|8|@]|9|@]|A|$]]]|2A|$2B|$5|2C|2D|2E|A|$2F|2G]]]]

I think that to solve this problem, you need to break it down into smaller pieces.
First of all, we know that at some point, based on the current time, we'll want to know when to trigger the next call. If we get a timestamp which gives us the current time in ms and we want to get the number of ms before the next hour, here's how we can do it:
<pre><code>const timeToNextHourInMs = (currentTimestampMs) =&gt; {
 const timestampSeconds = currentTimestampMs / 1000;

 const numberOfSecondsIntoTheCurrentHour = timestampSeconds % 3600;

 const numberOfSecondsToTheNextHour = 3600 - numberOfSecondsIntoTheCurrentHour;

 return numberOfSecondsToTheNextHour * 1000;
};
</code></pre>
I hope that the variable names are explicit enough that I do not need to comment but let me know otherwise.
Next, we want to tackle the stream issue:
<ul>
<li>We want to trigger an HTTP call straight away</li>
<li>Get the emitted value straight away</li>
<li>Do all the above again every time a new hour start (1:00, 2:00, 3:00, etc..)</li>
</ul>
Here's how you can do this:
<pre><code>this.http.get(`/update`).pipe(
 timestamp(),
 switchMap(({ timestamp, value }) =&gt;
 concat(
 of(value),
 EMPTY.pipe(delay(timeToNextHourInMs(timestamp)))
 )
 ),
 repeat()
);
</code></pre>
Let's go through the above logic:
<ul>
<li>First, we make the HTTP call straight away</li>
<li>Once the HTTP call is done, we get the current timestamp (to later on based on that find out when we want to do the next call)</li>
<li>We do a <code>switchMap</code> but as our HTTP call is only ever going to return 1 value it doesn't really matter in this very specific case. We could use <code>flatMap</code> or <code>concatMap</code> too</li>
<li>Inside the <code>switchMap</code>, we use <code>concat</code> to first of all send the value that we just got from the HTTP call but also keep that observable alive until the end of the current our (by using the function we created earlier)</li>
<li>At the end of the current hour, the stream will therefore complete. BUT, as we've got a <code>retry</code>, as soon as the stream completes, we'll subscribe to it again (and as a reminder, the stream will only complete at the very beginning of a new hour!)</li>
</ul>
One thing I'd suggest to add here but which isn't a requirement of the initial issue would be to have some error handling so that if something goes wrong when you make that call, it automatically retries it a few seconds after. Otherwise imagine when the polling kicks in if your network doesn't work for 5s at that exact time your stream will error straight away.
For this, you can refer to this <a href="https://stackoverflow.com/a/44911567/2398593">brilliant answer</a> and do that in a reusable custom operator:
<pre><code>const RETRY_DELAY = 2000;
const MAX_RETRY_FOR_ONE_HTTP_CALL = 3;

const automaticRetry = () =&gt; (obs$) =&gt;
 obs$.pipe(
 retryWhen((error$) =&gt;
 error$.pipe(
 concatMap((error, index) =&gt;
 iif(
 () =&gt; index &gt;= MAX_RETRY_FOR_ONE_HTTP_CALL,
 throwError(error),
 of(error).pipe(delay(RETRY_DELAY))
 )
 )
 )
 )
 );
</code></pre>
This will retry the observable 3 times with a delay between each retry. After 3 times, the stream will error by throwing the last emitted error.
Now, we can just add this custom operator to our stream:
<pre><code>this.http.get(`/update`).pipe(
 automaticRetry(),
 timestamp(),
 switchMap(({ timestamp, value }) =&gt;
 concat(
 of(value),
 EMPTY.pipe(delay(timeToNextHourInMs(timestamp)))
 )
 ),
 repeat()
);
</code></pre>
I haven't actually tested the code above so please do that on your side and let me know how it goes. But if my logic is correct here's how things should go:
<ul>
<li>Imagine you start your app at 2:40</li>
<li>An HTTP call is made straight away</li>
<li>You get the response pretty much instantly</li>
<li>The stream is set to be kept open for 20mn</li>
<li>At 3:00, the stream is completed and the retry kicks in: We do another HTTP call</li>
<li>This time, the server got re-deployed and was not available for a few seconds</li>
<li>Internally, the stream errors but thanks to our custom operator <code>automaticRetry</code> it waits for 3 seconds then retries the HTTP call once, still nothing. It waits another 3 seconds and this time it's fine, the result is passed downstream</li>
<li>Repeat this indefinitely :)</li>
</ul>
Let me know how it goes

blocks|key|2208751|text|我觉得很简单。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2208752|计时器接受两个参数，第一个参数是第一个呼叫的延迟，然后是调用的间隔。你的下一个小时的偏移是正确的，我用你的代码。|2208753|使用startWith(null)立即启动。|offset|length|style|CODE|2208754|const+startDelayMs+=+(60+-+new+Date().getMinutes())+*+60+*+1000;
const+hourMs+=+60+*+60+*+1000;

timer(startDelayMs,+hourMs).pipe(
+++startWith(null),
+++switchMap(())+//+do+your+call+here
)|code-block|syntax|javascript|2208755|entityMap^0|0|0|2|F|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|R|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|S|8|@$F|T|G|U|H|I]]|9|@]|A|$]]|$1|J|3|K|5|L|7|V|8|@]|9|@]|A|$M|N]]|$1|O|3|-4|5|6|7|W|8|@]|9|@]|A|$]]]|P|$]]

I think it is very simple.
Timer accepts two parameters, the first is the delay for the first call and then the interval in which the call should be made. You got the next full hour offset right, I'm using your code for that.
Use <code>startWith(null)</code> to start immediately.
<pre><code>const startDelayMs = (60 - new Date().getMinutes()) * 60 * 1000;
const hourMs = 60 * 60 * 1000;

timer(startDelayMs, hourMs).pipe(
 startWith(null),
 switchMap(()) // do your call here
)
</code></pre>

I need to build a stream that makes api request immediately after the call and that each hour (1:00 PM, 2:00 PM) if page wasn't refreshed. I build this with setTimeout() but I want to implement it with RxJs. Can you help me pls.
<pre><code>isUpdated() {
 return new Observable&lt;Object&gt;(function update(obs) {
 this.http.get(`/update`).subscribe(data =&gt; {
 obs.next(data);
 });
 const timer = (60 - new Date().getMinutes()) * 60 * 1000;
 setTimeout(() =&gt; update.call(this, obs), timer);
 }.bind(this));
}

//call
isUpdated().subscribe(data =&gt; console.log(data));
</code></pre>

RxJS infinity stream that emits every hour

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我需要构建一个流，它在调用后立即发出api请求，如果页面没有刷新，则每小时(下午1:00，下午2:00 )。我用setTimeout()构建它，但是我想用RxJ实现它。你能帮帮我吗？isUpdated() { return new Observable<Object>(function update(obs) {...

问每小时发射一次的RxJS无穷大流
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问每小时发射一次的RxJS无穷大流EN