java - RxJava 使用优化请求
问题描述
今天我试图解决一个小挑战:
您是一家拥有 500 个办事处的大公司,您想要计算全球收入(每个办事处的收入总和)。
每个办公室都公开一项服务以获取收入。调用需要一定的延迟(网络、数据库访问,...)。
显然,您希望尽快获得全球收入。
首先,我在 python 中尝试了非常好的结果:
import asyncio
import time
DELAYS = (475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175)
class Office:
def __init__(self, delay, name, revenue):
self.delay = delay
self.name = name
self.revenue = revenue
async def compute(self):
await asyncio.sleep(self.delay / 1000)
print(f'{self.name} finished in {self.delay}ms')
return self.revenue
async def main(offices, totest):
computed = sum(await asyncio.gather(*[o.compute() for o in offices]))
verdict = ['nok', 'ok'][computed == totest]
print(f'Sum of revenues = {computed} {verdict}')
if __name__ == "__main__":
offices = [Office(DELAYS[i % len(DELAYS)], f'Office-{i}', 3 * i + 10) for i in range(500)]
totest = sum(o.revenue for o in offices)
start = time.perf_counter()
asyncio.run(main(offices, totest))
end = time.perf_counter()
print(f'Ends in {(end-start)*1000:.3f}ms')
在我的电脑上大约需要 500 毫秒,理想情况下(因为 500 毫秒是最大延迟)
接下来,我用 RxJava 在 java 中进行了尝试:
import java.util.concurrent.TimeUnit;
public class Office {
private int sleepTime;
private String name;
private int revenue;
public Office(int sleepTime, String name, int revenue) {
this.sleepTime = sleepTime;
this.name = name;
this.revenue = revenue;
}
public int getRevenue() {
return revenue;
}
public int compute() {
try {
TimeUnit.MILLISECONDS.sleep(this.sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.printf("%s finished in %dms on thread %d%n", this.name, this.sleepTime, Thread.currentThread().getId());
return this.revenue;
}
}
import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
public class Tester {
private static int[] DELAYS = {475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175};
public static void main(String[] args) {
final ArrayList<Office> offices = new ArrayList<>();
for (int i = 0; i < 500; i++) {
offices.add(new Office(DELAYS[i % DELAYS.length], String.format("Office-%d", i), 3 * i + 10));
}
int totest = offices.stream().mapToInt(Office::getRevenue).sum();
final Instant start = Instant.now();
final Flowable<Office> officeObservable = Flowable.fromIterable(offices);
int computation = officeObservable.parallel(500).runOn(Schedulers.io()).map(Office::compute).reduce(Integer::sum).blockingSingle();
boolean verdict = computation == totest;
System.out.println("" + computation + " " + (verdict ? "ok" : "nok"));
final Instant end = Instant.now();
System.out.printf("Ends in %dms%n", Duration.between(start, end).toMillis());
}
}
在我的电脑上,大约需要 1000 毫秒(有 500 个线程池!!)。
当然,我尝试了不同数量的线程,但结果最差或相似。
我不想比较 Python 和 Java,我只想:
如果我做错了解释
更好的方法?
此外,python async 仅使用一个线程,但在 Java 中我没有找到如何不使用多线程来获得类似的结果。
也许有人可以帮助我?:-)
解决方案
这很简单。在 python 端,您在 java 端以异步模式(非阻塞)等待,您使用阻塞代码等待,因此有所不同。
java中正确的代码应该是:
package com.test;
import io.reactivex.Flowable;
import io.reactivex.Single;
import io.reactivex.schedulers.Schedulers;
import org.reactivestreams.Publisher;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
public class TestReactive {
public static class Office {
private int sleepTime;
private String name;
private int revenue;
public Office(int sleepTime, String name, int revenue) {
this.sleepTime = sleepTime;
this.name = name;
this.revenue = revenue;
}
public int getRevenue() {
return revenue;
}
public Publisher<Integer> compute() {
return Single.just("")
.delay(this.sleepTime, TimeUnit.MILLISECONDS)
.map(x-> {
System.out.printf("%s finished in %dms on thread %d%n", this.name, this.sleepTime, Thread.currentThread().getId());
return this.revenue;
}).toFlowable();
}
}
private static int[] DELAYS = {475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175};
public static void main(String[] args) {
final ArrayList<Office> offices = new ArrayList<>();
for (int i = 0; i < 500; i++) {
offices.add(new Office(DELAYS[i % DELAYS.length], String.format("Office-%d", i), 3 * i + 10));
}
int totest = offices.stream().mapToInt(Office::getRevenue).sum();
final Instant start = Instant.now();
final Flowable<Office> officeObservable = Flowable.fromIterable(offices);
int computation = officeObservable.parallel(2).runOn(Schedulers.io()).flatMap(Office::compute).reduce(Integer::sum).blockingSingle();
boolean verdict = computation == totest;
System.out.println("" + computation + " " + (verdict ? "ok" : "nok"));
final Instant end = Instant.now();
System.out.printf("Ends in %dms%n", Duration.between(start, end).toMillis());
}
}
编辑:我将并行设置为 2,但谁在乎,您可以放置一个线程,因为这不是 CPU 限制问题。
推荐阅读
- wordpress - 两个域名,单主机,htaccess
- python - 字典和嵌套字典之间的Python键匹配并将新值写入嵌套字典
- html - 同一个网站的未发布版本完美时,为什么我发布的网站会变形?
- corda - 每个节点的 Corda 帐户限制
- monit - 使用 monit 监控 bash 脚本执行
- python - Python 3.8 上的不和谐机器人
- websocket - 使用 AWS Lambda + API Gateway 的 Websocket 连接
- postgresql - 如何使用 row_number() 将行号分配给 postgresql 中的重复值
- python - 从不均匀的 pandas dict-like 系列中提取元素
- windows - Perl 无法在 Windows 10 中找到现有目录