可观察的角度不平行工作

Question

我有一个代码，我在其中使用 subscribe 调用一个可观察的函数。 我希望它并行运行，但它是按顺序运行的。

makedownloadData() 是将“showData”列表变量的所有变量存储在不同变量中的函数，比如说“downloadData”。函数需要大量时间将所有变量复制到 downloadData 和进一步处理。所以我想使用可观察/订阅技术调用这个函数，以便它应该并行运行而不会导致当前序列延迟，但它只是按顺序运行。没有增强。

第一种方法（没有可观察的）

调用函数

this.downloadData=this.makeDownloadData() //This step is taking lot's of time as function respond late. 
console.log("print after function call")

要调用的函数

public makeDownloadData() {
var main_data = this.showData;
var down_data:any = [];
for (var i=0; i<main_data.length; i++){
  var element: String = "";
  var newDate = new Date(main_data[i]["@timestamp"]);
  element = element.concat(this.convertDateToLocalFormat(newDate)+" ");
  element = element.concat(String(main_data[i]["cu_hostname"])+" ");
  element = element.concat(String(main_data[i]["log_agent"])+".");
  element = element.concat(String(main_data[i]["log_level"])+" ");
  element = element.concat(String(main_data[i]["app_name"])+": ");
  element = element.concat(String(main_data[i]["log_message"])+" ");
  down_data.push(element.concat("\n"));
}
return down_data;

}

输出：

//Execution of function
"print after function call"

第二种方法（使用可观察）

导入需求

import { Observable } from 'rxjs';
import 'rxjs/add/observable/of'

调用可观察函数。

this.makeDownloadData().subscribe(response => {
   console.log("Expected to print after") //This should be run in parallel and must printed after upcoming statement as this function is taking time to respond.
   console.log(response); //Expected to print after 
},
error => {
  console.log("Did not got response")
});

console.log("Expected to print before")

要调用的函数

public makeDownloadData(): Observable<any> {
var main_data = this.showData;
var down_data:any = [];
for (var i=0; i<main_data.length; i++){
  var element: String = "";
  var newDate = new Date(main_data[i]["@timestamp"]);
  element = element.concat(this.convertDateToLocalFormat(newDate)+" ");
  element = element.concat(String(main_data[i]["cu_hostname"])+" ");
  element = element.concat(String(main_data[i]["log_agent"])+".");
  element = element.concat(String(main_data[i]["log_level"])+" ");
  element = element.concat(String(main_data[i]["app_name"])+": ");
  element = element.concat(String(main_data[i]["log_message"])+" ");
  down_data.push(element.concat("\n"));
}
return Observable.of(down_data)

}

输出：

"Expected to print after"
Printing response
"Expected to print before"

预期输出：

"Expected to print before"
"Expected to print after"
Printing response

我想使用 observable 进行并行执行。请帮忙。如果有任何不清楚的地方，我会修改问题以使其更清楚。 谢谢

Answer 1

首先，您需要在订阅获取数据之后调用打印前的订阅方法，然后调用打印后的常规方法，这样现在您就可以获得正确的答案了，它会帮助您进行任何澄清，请依赖这个答案

Answer 2

有两个问题：

1. 您的函数makeDownloadData() 不是可观察的，它只是返回一个可观察的（需要时间的代码在返回语句之前）
2. 并非所有的 observables 默认都是异步的（事实上，只有少数是默认异步的）

以下代码具有完全异步的 observable（感谢 of 函数上的 asyncScheduler 参数 - 没有它，of 是一个返回 observable 的同步函数）：

const source = of('Hello, subscriber', asyncScheduler).pipe(
  tap(() => console.log('Inside observable')),
  map(x => {
    for(let i = 0; i < 1000000000; i++)
      for(let j = 0; j < 1; j++);

    return x;
  })
);
console.log('After observable')

source.subscribe(x => console.log(x));

说这是异步的，我的意思是After observable 将立即出现在控制台中。一段时间后，您会看到'Inside observable'，然后是'Hello, subscriber'。

如果你删除了 observable 的 asyncScheduler，你会等待一段时间并且会看到相同的序列（'After observable', 'Inside observable', 'Hello, subscriber'），但是所有的代码都会被阻塞，直到 observable 内部的循环结束，紧接着，您会看到三个字符串几乎同时打印在控制台上。看看这个演示：https://stackblitz.com/edit/rxjs-stackoverflow-556297976433166?file=index.ts

所以你可以试试这个：

makeDownloadData(): any { // <== doesn't need to return an observable
  var main_data = this.showData;
  var down_data:any = [];
  for (var i=0; i<main_data.length; i++){
    var element: String = "";
    var newDate = new Date(main_data[i]["@timestamp"]);
    element = element.concat(this.convertDateToLocalFormat(newDate)+" ");
    element = element.concat(String(main_data[i]["cu_hostname"])+" ");
    element = element.concat(String(main_data[i]["log_agent"])+".");
    element = element.concat(String(main_data[i]["log_level"])+" ");
    element = element.concat(String(main_data[i]["app_name"])+": ");
    element = element.concat(String(main_data[i]["log_message"])+" ");
    down_data.push(element.concat("\n"));
  }
  return down_data // <== this doesn't need to return an observable
}

更新：我将of(this.makeDownloadData()) 替换为of(1)，否则将花费相同的时间来创建运行原始问题中的函数所需的运算符。然后你可以将它映射到所需的功能。

import { of, asyncScheduler } from 'rxjs'; 
import { map } from 'rxjs/operators';

...

this.downloadData=of(1, asyncScheduler)
    .pipe(map(() => this.makeDownloadData());

在这里使用of 不是创建可观察对象的最佳方法，但它是迄今为止最清楚的，因为我需要asyncScheduler 使其异步。作为第一个参数的“1”是一个哑数据。你可以使用任何东西。 of 函数需要一个参数。

Answer 3

与 promise 不同，代码类似

Promise.resolve("one").then(console.log);
console.log("two");

导致在“one”之前打印“two”，observables 本质上不是异步的，就像回调不是。所以of("one").subscribe(console.log) 将立即打印，而不是在事件循环结束后打印。

这是一个更长的讲座： https://www.syntaxsuccess.com/viewarticle/rxjs-subjects-emit-synchronous-values