如何从可观察到的源中仅缓冲选定的项目集，并立即发出其他项目？
从可观察到的源中仅缓冲选定的项目集，并立即发出其他项目的方法是通过使用消息队列（Message Queue）来实现。
消息队列是一种在应用程序之间传递消息的通信方式，它可以将消息存储在队列中，然后按照一定的规则进行处理和传递。在这种情况下，可观察到的源可以将所有的项目发送到消息队列中，然后消费者可以选择性地从队列中获取并处理特定的项目集，而不会影响其他项目的传递。
以下是消息队列的一些概念、分类、优势、应用场景以及腾讯云相关产品的介绍：
- 概念：消息队列是一种异步通信机制，通过将消息发送到队列中，实现不同组件之间的解耦和通信。
- 分类：消息队列可以分为点对点模型和发布/订阅模型。点对点模型中，消息发送者将消息发送到队列中，只有一个消费者可以接收并处理该消息。发布/订阅模型中，消息发送者将消息发送到主题（Topic）中，多个消费者可以订阅该主题并接收消息。
- 优势：
- 解耦：消息队列可以将消息发送者和接收者解耦，提高系统的可维护性和扩展性。
- 异步：消息队列支持异步通信，发送者无需等待接收者的响应即可继续处理其他任务。
- 削峰填谷：消息队列可以平滑处理系统的峰值流量，避免系统过载。
- 可靠性：消息队列提供消息持久化、消息重试等机制，确保消息的可靠传递。
- 应用场景：
- 异步任务处理：将耗时的任务放入消息队列中，由消费者异步处理，提高系统的响应速度。
- 流量削峰：将突发的请求放入消息队列中，按照系统的处理能力逐渐消化，避免系统崩溃。
- 日志处理：将日志消息发送到消息队列中，由消费者进行实时处理和存储。
- 分布式系统通信：不同服务之间通过消息队列进行通信，实现解耦和水平扩展。
腾讯云相关产品：腾讯云消息队列 CMQ（Cloud Message Queue）
- 产品介绍链接地址：https://cloud.tencent.com/product/cmq
相关·内容
2回答
从医生那里我可以看出：
冷观测值从一开始就发出整个值序列到每个观察者。
和
ReplaySubject向任何观察者发出可观察到的源发出的所有项目，而不管观察者何时订阅。
那么这两个概念之间有什么区别呢？
谢谢
浏览 4提问于2016-03-26得票数 11
回答已采纳
我有一个可观察的，当它被添加到db时，它是在听数据库和发射项目。当我订阅这个可观察到的内容时，它会一个一个地在db中快速释放已经存储的项。我的问题是，我是否可以创建可观察的方法，将以特定间隔(例如，100 millis)发出的项收集到列表中，并发出(或在某些函数中返回，如doOnNext)整个列表和单独的项(如果发出的间隔较大)？
提前感谢！
浏览 1提问于2017-09-04得票数 1
回答已采纳
我想在一个固定的时间间隔内重新发出可观察到的最后一个值，我尝试了。
obs.pipe(repeat({delay:1000})).subscribe(x => console.log('Emitted', x));
但这不起作用。在研究过这个之后，我的可观察到的实际上是一个BehaviorSubject。
所以我的问题是为什么第一颗卫星每秒钟发射一次
of('Observable').pipe(repeat({ delay: 1000 })).subscribe(x => console.log(x));
但不是这个？
var bs = new Be
浏览 12提问于2022-12-02得票数 0
与docs略有不同的stackblitz：
input-fieldcheck中的类型控制台输出
我想要的和我期望的：
我希望当我触发关键事件时，of(null)的发射会被debounceTime延迟。这里的情况并非如此。
throttleTime不工作eitherdelay似乎在工作
(我的假设)：也许of完成了，因此debounceTime不等待任何秒值？
有人能澄清吗？谢谢
代码副本：
keyup$
.pipe(
switchMap(() =>
of(null).pipe(
debounceTime(1000),
tap(()
浏览 2提问于2020-12-22得票数 0
回答已采纳
1回答
我有一个共享服务data.service.ts，它有
public pauseProjectTask$: Subject<any> = new Subject<any>();
pauseTaskProject(taskData,type){
this.pauseProjectTask$.next(taskData);
}
而且我有一个footer.component.ts，它显示当前每个页面中的活动任务，任务可以从页脚组件中暂停。我在脚注中有一个暂停函数。
pauseTask(taskData,type){
console.l
浏览 0提问于2020-08-24得票数 0
我使用的是Angular 5，ngrx/store。我有一个效果：
@Effect()
updateAfterChanges$ = this.actions$.pipe(
ofType(ActionTypes.ChangesInItems),
withLatestFrom(
this.store.select(fromReducers.getItemsList)
),
switchMap(([action, itemsList]: [ActionType, Items[]]) => {
return [new UpdateIt
浏览 0提问于2018-06-01得票数 4
已经尝试解决这个问题有一段时间了。目前我有一个对象数组(我称它们为tiles)，这是相当大的。我有一个API端点，我应该一个接一个地发送这些对象，这个API不返回任何东西，只返回状态。我需要以并行和并发的方式将这些对象发送到端点，当最后一个成功时，我应该发出一些字符串值，该值将进入redux存储。 const tilesEpic =(action$, _state$) => {
action$.pipe(
ofType('TILE_ACTION'),
map(tilesArray => postTilesConcurrent
浏览 29提问于2021-06-23得票数 0
3回答
编辑2:更有效的是shareReplay(1)，如更新中所述。
编辑1:最终效果最好的是：
@Injectable()
export class MyGlobalService {
private resource$;
private resource$Connected;
constructor(private http: Http) {
this.resource$Connected = false;
this.resource$ = this.http
.get('/api/resource')
.map((res
浏览 6提问于2017-04-28得票数 1
回答已采纳
1回答
我一直试图以顺序顺序获取对象流，但是concatMap并不适合我。使用mergeMap，我得到了流，但不是按顺序排列的。
下面是我正在尝试的代码：
this.result
.pipe(
map((result: Result) => {
return result.contents
.map((content: Content) => this.databaseService
.getResource<Resource>('Type', conten
浏览 2提问于2022-10-17得票数 1
回答已采纳
我有一个BehaviourSubject，它是通过Service.This行为主体的构造函数中的getSaveBehaviorSubject()订阅的，它可以接收来自其他组件/服务的多种类型的数据，其中一些类型需要存储在后端，而其他类型则不需要存储。当它接收到一个确实需要存储的值时，它会通过一个put/post请求将其发送到后端。作为请求速率限制器，我使用的是debounce(300)操作符，所以对服务器的实际请求是在300 ms的不活动通过之后进行的。这将防止服务器引发429 Too many requests错误。只有当需要存储的值类型被推送时，才应该应用debounce运算符。我尝试用i
浏览 3提问于2020-06-17得票数 0
回答已采纳
2回答
这个rxjs合并逻辑可以简化吗？
、、、、
我从Firebase客户端获得了一个可观察的(onAuthStateChanged)，即：
如果用户没有登录，会立即发出null，而会在几分钟后发出null，然后在用户登录时发出user object。
const observable = new Observable((obs) => {
return app.auth().onAuthStateChanged(
obs.next,
obs.error,
obs.complete
)
})
我想要的是：
accepted)always 忽略应用生命周期前1000 is发出
浏览 3提问于2021-08-09得票数 3
回答已采纳
尝试为BufferUntil实现Observable (即，每当满足某些条件时，缓冲区元素和发出缓冲列表)
为RxNET实现
界定：
public static IObservable<IList<TSource>> BufferUntil<TSource>(
this IObservable<TSource> source,
Func<TSource, bool> predicate)
{
var published = source.Publish(
浏览 5提问于2017-02-16得票数 2
回答已采纳
我用的是角4.4.2。
我尝试使用http.get发出请求，并将发出的响应缓冲到定义的时间。但是它不起作用，而当我试图拖延的时候，它就起作用了。
你能解释一下为什么吗？
this.http.get('https://jsonplaceholder.typicode.com/posts/2')
.bufferTime(4000)
.subscribe(
_ => console.log('BufferTime', _)
);
下面是一个柱塞：
谢谢
浏览 4提问于2017-09-20得票数 0
回答已采纳
在调用onNext之后，我希望发出一个随机值10秒。因此，12345和10秒后将调用onNext，随机数应该调用onNext。使用RxJava实现这一目标的最佳方法是什么？提前谢谢。
Observable.create(subscriber -> {
subscriber.onNext(12345);
}).subscribeOn(...)
.observeOn(...)
.subscribe(new Subscriber<Long>(){
@Override
public void onNext(Long aLong) {
//Do
浏览 2提问于2017-01-30得票数 3
回答已采纳
我有一个布尔值，它(通过一个观察者)启动和停止一个Android服务。传递true时，必须立即启动服务。当false被传递时，我想揭穿它，以防不久之后出现true，以避免不必要的(和破坏性的)服务的停止和启动。是否有一个标准操作符可以这样做，或者我必须自己写，如果是的话，我应该基于OperatorDebounceWithTime.java还是有一个更简单的方法？谢谢!
浏览 4提问于2015-10-06得票数 3
回答已采纳
我有两个带有默认值的BehaviorSubjects $A和$B。它们都在使用.next(value)在不同时间使用值进行更新。在后面的代码中，我试图提取如下所示的值
Observable.combineLatest($A, $B)
.take(1)
.map(([$a, $b]) => {
return {
// stuff
}
})
.subscribe(fn)
我看到的行为很奇怪。我期望可观察到的对象等待任何一个主体发出一个值，然后给我一个发出的值，我可以将其传递给订阅函数。但它只释放值与$B发出一个新的价值。当$A发布时，它将从$B获
浏览 1提问于2018-08-21得票数 1
2回答
我正在实现一个可观察的，在延迟5s后重试错误的方法。我在用改装来建立关系网。我面临的问题是，当API返回一个错误时，会有很多重试。我只想在5秒后再试一次，但是重试的速度非常快(一秒钟几乎有三次)。知道为什么吗？
userAPI.getUsers()
.filter { it.users.isNotEmpty() }
.subscribeOn(Schedulers.io())
.retryWhen { errors -> errors.flatMap { errors.delay(5, TimeUnit.SECONDS) } }
.o
浏览 1提问于2018-08-13得票数 1
回答已采纳
如果用户保持不活动状态，我正在尝试在5秒后触发一个模式，并为此编写了一个包含以下代码的服务：
public initScreenListen(): void {
this.mergedObservable$ = merge(
fromEvent(document, BanKeyboardEvent.KEY_DOWN),
fromEvent(document, BanMouseEvent.MOUSE_DOWN),
fromEvent(document, BanMouseEvent.MOUSE_MOVE),
fromEvent(documen
浏览 4提问于2021-09-02得票数 1
回答已采纳
我在学习角和RxJS。读了一本叫做“行动中的RxJS”的书，我看到两个运营商都提到了。第一个接收并可观察为一个参数，第二个接收一个选择器函数。基于这些参数，两种缓冲区内容似乎都是基于这些参数。
他们之间有什么区别？什么时候应该使用缓冲区而不是bufferWhen操作符？
浏览 3提问于2020-04-02得票数 4
回答已采纳
1回答
我们是新的角度，我们有很少的方法需要按顺序调用并返回一个值。在这里，返回是在完成执行之前调用的。我尝试过不同的东西，比如异步等待，承诺，setTimeout，但是没有运气。是否有任何解决方案可以按角度同步顺序工作？
transform(input:any)
{
this.getCachedUsers(); // this will set the this.CachedUsers array using a http get call(subscribe)
this.getUserFromCachedUsers(input); // this will set the thi
浏览 3提问于2022-04-06得票数 0
在运行我的示例代码之后，我期望得到如下结果。
atest
btest
ctest
但实际上什么都没发生。
请让我知道我的密码有什么问题。
Subject<String, String> subject = PublishSubject.create();
subject.onNext("test");
Observable<String> observable = Observable.from(new String[] {"a", "b", "c"}).repeat(2);
浏览 2提问于2015-11-24得票数 1
回答已采纳
是否有一个操作员缓存最后发出的项目并将其传递给每个新订户？换句话说，使可观察到的行为像BehaviorSubject的操作符
浏览 4提问于2017-05-09得票数 12
回答已采纳
我有这样的JSON：
{
"success":true,
"data":[
{
"id":"29",
"name":"\u0420\u0435\u0441\u0442\u043e\u0440\u0430\u0446\u0456\u044f \u0411\u0430\u0447\u0435\u0432\u0441\u044c\u043a\u0438\u0445 \/ Baczewski Restaurant",
"street":
浏览 4提问于2016-05-16得票数 4
回答已采纳
1回答
我从get调用API接收元素流，在第一次这样做时，我不需要过滤这个数据。我可以随时提供信息。
但我有一个表单a根据过滤器，有人可以选择，我需要过滤数据。
我遇到的主要问题是，我使用一个映射运算符迭代数组中的所有元素，当它和filter.continent拥有相同的大陆时，返回这个国家，但是当它不匹配时，它返回的是未定义的。
经过这个过滤器之后，我得到了这样的信息：
[undefined, {some country data}, undefined, undefined, {some country data}]
这是我为这个重新筛选函数编写的代码示例。
refilterValues(){
浏览 3提问于2018-01-13得票数 2
回答已采纳
我找到一个代码片段：
<?php
$sql = "SELECT * FROM users WHERE first='sof';";
$result = mysqli_query($conn, $sql);
$resultCheck = mysqli_num_rows($result);
var_dump($result);
if($resultCheck > 0){
while($row = mysqli_fetch_assoc($result)){
var_dump($row) ;
}
}
我有一个关于上面
浏览 0提问于2018-08-13得票数 0
1回答
代码：
const Rx = require('rxjs')
const data = [
{ name: 'Zachary', age: 21 },
{ name: 'John', age: 20 },
{ name: 'Louise', age: 14 },
{ name: 'Borg', age: 15 }
]
const dataSubj$ = new Rx.Subject()
function getDataStream() {
return dataSubj$
浏览 1提问于2018-02-21得票数 2
回答已采纳
管道(采取(1))在从authService调用方法时扮演什么角色？
this.authService
.signIn(email, password)
.pipe(take(1))
.subscribe(...)
浏览 5提问于2020-03-14得票数 1
回答已采纳
11回答
我想知道如何忽略异常并继续无限流(在我的例子中是位置流)？
我正在获取当前的用户位置(使用)，然后将它们发送到我的API (使用)。
在我的示例中，当异常发生在网络调用(例如超时)时，将调用onError方法并停止流本身。如何避免呢？
活动：
private RestService mRestService;
private Subscription mSubscription;
private LocationRequest mLocationRequest = LocationRequest.create()
.setPriority(LocationRequest
浏览 17提问于2015-03-10得票数 55
回答已采纳
有没有一种方法可以在AdvancedDataGrid中移动一行来响应外部事件？在这种情况下，我希望有一个工具栏按钮来向上或向下移动网格中选定的一个或多个项目。我所看到的问题是保持分层数据源与网格上的项目呈现器和选择数据的数组同步。
浏览 3提问于2009-07-28得票数 0
我第一次使用RxJS，我试图创建一个简单的可观察性，它将占用最大数量的条目，并等待一个时间间隔才能将数据传递给订阅者。
这就是我所拥有的：
import EventEmitter from "events";
import { fromEvent } from "rxjs";
import { bufferTime, filter, take } from "rxjs/operators";
const ev = new EventEmitter();
const observer = fromEvent(ev, "log")
浏览 1提问于2018-11-11得票数 0
1回答
我只是想弄清楚热和冷之间的概念，观察和尝试莫尼福图书馆。我的理解是，下面的代码应该只导致一个订阅者得到可观察到的事件，但事实并非如此！
scala> :paste
// Entering paste mode (ctrl-D to finish)
import monifu.reactive._
import scala.concurrent.duration._
import monifu.concurrent.Implicits.globalScheduler
val obs = Observable.interval(1.second).take(10)
val x = o
浏览 0提问于2015-07-09得票数 2
回答已采纳
我有一个流，它可能出错，也可能不出错，我想要catchError以返回错误之前的前一个/最后一个值，以便继续流而不完成。与此类似的东西：
inputStream$.pipe(catchError(err => inputStream$.previousValue)).subscribe(...)
浏览 0提问于2018-09-11得票数 0
回答已采纳
1回答
将多个可观测数据连接到一个源
、、、、
我有很多不同类型的网络调用，比如字符串、int等等。
我在试着让它们平行。
在官方的rxjava文档中，我们可以读到：
Flowable.range(1, 10)
.parallel()
.runOn(Schedulers.computation())
.map(v -> v * v)
.sequential()
.blockingSubscribe(System.out::println);
这个例子很简单，因为我们的所有类型都是Int。但是，如果我们有不同的类型，例如字符串、布尔、Int，该如何做呢？
这个5的每个调用都是independentgroup，这5个调
浏览 3提问于2020-12-28得票数 0
我有一个绑定到CollectionViewSource的ListView (它的源是一个ObservableCollection。下面示例中的myList )。
<CollectionViewSource x:Key="cvs" Source="{Binding myList}" IsLiveGroupingRequested=true />
<ListView Name="itemsList"
SelectionMode="Extended"
浏览 1提问于2018-03-23得票数 1
我有一个单元测试来测试BufferWithTime。当在缓冲区将发出新值的点发出值时，我似乎得到了不一致的结果。
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
new Recorded<Notification<int>>(50, new Notification<int>.OnNext(1)),
new Recorded<Notification<int>>(100, new Notificatio
浏览 0提问于2011-03-10得票数 1
回答已采纳
在React/Redux应用程序中，我使用可观察的Redux，我想要一个epic来输出两个不同的操作。
首先，我尝试了这个：
const epic1 = action$ =>
action$
.ofType("A")
.mapTo({ type: "B" })
.merge(Observable.of({ type: "C" }));
由于它不起作用，我不得不切换到：
const epic2 = action$ =>
action$
.ofType("D")
.merge
浏览 0提问于2018-05-07得票数 2
回答已采纳
RxJS版本： 5.5.6
代码复制：
var obs = Rx.Observable.interval(100)
.startWith(0)
.flatMap(() => {
return Rx.Observable.empty();
});
var sub = obs.subscribe(
(data) => {console.log(data, 1)},
(err) => {c
浏览 1提问于2018-02-02得票数 1
回答已采纳
我正在修改NestJ，我试图在onModuleInit()中调用一些数据，我的想法是在组件完全挂载之前获得数据。我的尝试是：
@Injectable()
export class CategoriesService implements OnModuleInit {
constructor(private httpService: HttpService) {}
onModuleInit() {
console.log(this.httpService.get('https://api.github.com/users/januaaA
浏览 11提问于2022-06-23得票数 0
回答已采纳
1回答
我正在寻找一种使流组合起来的方法--注意:这是我问题的最简单的形式，实际上我正在组合8种不同的流--有些是相互交织的，有些是异步的，等等:(
import { BehaviorSubject, map, combineLatest } from 'rxjs';
const $A = new BehaviorSubject(1)
const $B = $A.pipe(map(val => `$B : ${val}`))
const $C = $A.pipe(map(val => `$C : ${val}`))
// prints out:
// (1) [1, &
浏览 8提问于2022-10-29得票数 1
回答已采纳
问题的简介:当使用android添加的MediatorLiveData时，我发现自己经常从每个源调用相同的函数。例如，这可能是因为，每当一个源被更新时，我必须检查它是否有影响，或者另一个源是否更重要。代码中的一个示例(Kotlin，但不重要)：
val isHovered = MutableLiveData<Boolean>()
val isSelected = MutableLiveData<Boolean>()
val color = MediatorLiveData<Int>().apply {
addSource(isHovered) {
浏览 0提问于2019-07-29得票数 2
我有Android应用程序的方法，返回图像作为可观察的。我需要这个图像来使用它在button.setDrawable(可绘图)。所以我需要简单的可拉式。如何从返回语句中获得这样的图像？
浏览 2提问于2016-10-26得票数 1
回答已采纳
我使用rxJava从数据库中获取数据，并在回收视图中显示数据。相关代码如下所示
function updateUI(){
ContactsLab contactsLab = ContactsLab.get(getActivity());
Subscription sub = contactsLab.getContactList().subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.toList()
.subscribe(
浏览 4提问于2016-08-16得票数 1
回答已采纳
1回答
我正在过滤一个JSON对象数组，我想根据在id参数中传递的内容返回找到的对象。
getClient(id) {
return this.http.get<any>(' http://localhost:3000/clients')
.pipe(
filter(client => client.idNumber === id)
);
}
上面的观察者不会被触发，也不会接收任何数据。
getClient(id)
.subscribe(
(data) => {
// nothing here
浏览 1提问于2020-08-19得票数 1
回答已采纳
2回答
我试图使用存储API来存储我的文件，但是管道被执行了两次。有什么办法让它只触发一次吗？
// def
task: AngularFireUploadTask;
storage: AngularFireStorage;
// code
this.task = this.storage.upload(path, file, { customMetadata });
return this.task.snapshotChanges().pipe(
map(doc => {
console.log('me'); // THIS I
浏览 0提问于2019-12-27得票数 1
回答已采纳
我正在尝试在子组件之间进行通信，但是通过文档，我需要对每个服务执行此操作，当我检索服务信息时，当我试图分配订阅返回时，如果我给出一个console.log()，它就会工作，现在如果我给一个变量赋值，它就不会考虑事后访问它，给出一个未定义的答案。
将信息传递给另一个的子类
import { Component, OnInit } from '@angular/core';
import {Angular2TokenService, ResetPasswordData } from 'angular2-token';
import { ActivatedRoute,
浏览 2提问于2017-03-01得票数 0
回答已采纳
我正在学习如何使用RxJava (io.reactivex.rxjava2:rxandroid:2.0.1) + retrofit2 + kotlin。
一般情况下，如何观察到这一点，甚至学会了如何在Observable.zip的帮助下结合基于所有人的性能的2-3个并行查询的结果，但是如何执行类似的3次以上的并行查询，我无法理解。
可观察类的源代码有一个方法
static <T1, T2, T3, T4, R> Observable <R> zip (),
但它的工作方式不像
static <T1, T2, T3, R> Observable <R&g
浏览 0提问于2018-08-18得票数 2
回答已采纳
我试图在Qt中为一个学术项目开发一个用户界面，它向微控制器发送串行消息。不过，我正在努力关闭串口:在关闭串口之前，用户界面应该向微控制器发送串行消息。这是管理此操作的代码的一部分：
device->write(QString("ABORT___").toStdString().c_str());
displayMessage("Disconnected");
deviceIsConnected = false;
device->close();
ui->PortComboBox->setEnabled(true);
ui->tabW
浏览 0提问于2020-03-16得票数 0
回答已采纳
private static PublishSubject<UserBean> mPublishSubject;
private static AtomicBoolean mRefreshing = new AtomicBoolean(false);
private static Observable<UserBean> mTokenObservable;
public Observable<UserBean> getNetTokenLocked() {
if (mRefreshing.compareAndSet(false, true)) {
浏览 0提问于2020-11-04得票数 0
回答已采纳
我有一个问题，关于使用批处理，重复和洗牌与tf.Dataset。
我不清楚如何使用重复和洗牌。我知道.batch会决定有多少训练样本会经历随机梯度下降，.repeat和.shuffle的使用对我来说还不清楚。
第一个问题
即使在查看和之后，一旦抛出tf.errors.OutOfRangeError，也会使用.repeat在dataset上重申。因此，在我的代码中，这是否意味着我不再需要实现：
try:
while True:
_ = sess.run(self.optimizer)
except tf.errors.OutOfRangeError:
p
浏览 0提问于2019-07-09得票数 6
回答已采纳
我正在尝试使用RxJava创建一个顺序下载服务。用户可以在批(20，30等)或个别项目中添加项目。这些项目将被添加到队列中，然后按10批顺序下载。为此，我使用PublishSubject：
PublishSubject<Int> pubSubject = PublishSubject.create();
它发出用户添加的项(ids)，然后将缓冲区运算符应用于批处理项。使用这些ids，项目在flatMap中下载，在订阅的onNext中返回。
pubSubject.buffer(1, TimeUnit.SECONDS, 10)
.observeOn(Sche
浏览 2提问于2019-06-26得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
