java - 与 LinkedBlockingQueue 相比,Apache Ignite Queue 慢得多
问题描述
我正在尝试在 Ignite 中复制一个简单的生产者-消费者场景:
public class QueueExample {
public static void main(String[] args) {
new QueueExample().start();
}
private void start() {
final AtomicBoolean finishedTest1 = new AtomicBoolean(false);
final BlockingQueue<Double> queue = new LinkedBlockingQueue<>(5);
final CountDownLatch latch = new CountDownLatch(2);
final int MAX = 1000;
new Thread(() -> {
System.out.println("test1 before latch");
latch.countDown();
try {
// wait until other runnable is able to poll
latch.await(20, TimeUnit.SECONDS);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
System.out.println(new Date().getTime() + " start test1");
double test = 2;
Random r = new Random();
StopWatch sw = new StopWatch();
sw.start();
for (int i = 0; i < MAX; i++) {
try {
queue.put(r.nextDouble());
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
sw.stop();
finishedTest1.set(true);
//LoggerFactory.getLogger(getClass()).info
System.out.println(new Date().getTime() + " end test1. " + test + ", took:" + sw.getTime() / 1000f);
}).start();
new Thread(() -> {
System.out.println("test2 before latch");
latch.countDown();
try {
// wait until other runnable is able to poll
latch.await(10, TimeUnit.SECONDS);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
System.out.println(new Date().getTime() + " start test2");
StopWatch sw = new StopWatch();
sw.start();
int counter = 0;
try {
for (int i = 0; i < MAX ; i++) {
Double res = queue.poll(1, TimeUnit.SECONDS);
counter++;
}
} catch (InterruptedException e) {
// expected
}
sw.stop();
//LoggerFactory.getLogger(getClass()).info
System.out.println(new Date().getTime() + " end test2. counter " + counter + ", finished:" + finishedTest1.get() + ", took:" + sw.getTime() / 1000f);
}).start();
}
}
为什么这比下面的 Ignite 代码快 100 倍(0.02 秒对 <2 秒)?
public class MyIgnite {
public static void main(String[] args) {
new MyIgnite().start();
}
private void start() {
IgniteConfiguration icfg = new IgniteConfiguration();
icfg.setIgniteInstanceName("test1");
Ignite ignite1 = Ignition.start(icfg);
final CountDownLatch latch = new CountDownLatch(2);
final int queueSize = 5;
CollectionConfiguration queueCfg = new CollectionConfiguration();
ignite1.compute().runAsync(new IgniteRunnable() {
@IgniteInstanceResource
Ignite ignite;
@Override
public void run() {
IgniteQueue<Double> queue = ignite.queue("test", queueSize, queueCfg);
System.out.println("test1 fetched queue");
latch.countDown();
try {
// wait until other runnable is able to poll
latch.await(20, TimeUnit.SECONDS);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
System.out.println("start test1");
double test = 2;
Random r = new Random();
StopWatch sw = new StopWatch();
sw.start();
for (int i = 0; i < 1000; i++) {
queue.put(r.nextDouble());
}
sw.stop();
//LoggerFactory.getLogger(getClass()).info
System.out.println("end test1. " + test + " at ignite " + ignite.name() + ", took:" + sw.getTime() / 1000f);
}
});
System.out.println("starting test2");
icfg = new IgniteConfiguration();
icfg.setIgniteInstanceName("test2");
Ignite ignite2 = Ignition.start(icfg);
ignite2.compute().runAsync(new IgniteRunnable() {
@IgniteInstanceResource
Ignite ignite;
@Override
public void run() {
IgniteQueue<Double> queue = ignite.queue("test", queueSize, queueCfg);
System.out.println("test2 fetched queue");
latch.countDown();
try {
// wait until other runnable is able to poll
latch.await(10, TimeUnit.SECONDS);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
System.out.println("start test2");
StopWatch sw = new StopWatch();
sw.start();
int counter = 0;
try {
for (int i = 0; i < 1000; i++) {
Double res = queue.poll(5, TimeUnit.SECONDS);
counter++;
}
} catch (IgniteException exc) {
System.out.println("Somehow cannot poll. " + exc);
}
sw.stop();
//LoggerFactory.getLogger(getClass()).info
System.out.println("end test2. counter " + counter + " at ignite " + ignite.name() + ", took:" + sw.getTime() / 1000f);
}
});
System.out.println("oldest node: " + ignite1.cluster().forOldest().hostNames());
System.out.println("nodes: " + ignite1.cluster().nodes().size());
// does it really gracefully shut the nodes down?
// Ignition.stop(ignite1.name(), false);
// Ignition.stop(ignite2.name(), false);
}
}
我尝试修改 ignite 存储,使其表现得更像内存,但没有成功更改数字。
解决方案
您在这里将航空母舰与玩具船进行比较。
LinkedBlockingQueue
是在单个 JVM 的内存中工作的数据结构。
IgniteQueue
是基于Ignite的key-value存储的分布式结构。它可以在数百台机器上工作,具有不同的一致性级别、备份副本和持久性。当然,它在底层得到了很多机器的支持,而且它比简单的本地队列慢。
推荐阅读
- google-data-studio - 尝试加载 repo 信息时,Github 连接器 Google Data Studio 抛出错误
- python - 在列表列表中转换 pd.DataFrame,同时为空文本“”替换“np.nan”值
- css - 在反应js中设置背景并模糊图像背景
- html - TailwindCSS - 修复了 flex 父级的 div 全宽
- excel - excel无法下载图片
- python - 简单的熊猫数据框转换
- ios - Ionic 5 Capacitor - 如何在 IOS 上动态更改状态栏背景颜色
- php - 如何将一个项目与多个域链接?
- r - R中的Smote函数
- zsh - 带有空格的转义字符串传递给 ZSH 函数