Programmiererforum

fun main(args: Array) {
val start = System.currentTimeMillis()
Internal().doWork()
println("Duration is ${(System.currentTimeMillis() - start)/1000} sec")
}

class Internal {

fun doWork() {

val pool = ThreadPoolExecutor(
3, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
ArrayBlockingQueue(1000),
)

val future = CompletableFuture.supplyAsync(
{
// 1 subtask
val future1 = CompletableFuture.supplyAsync(
{
(1..10).map {
CompletableFuture.supplyAsync(SingleExternalCall(), pool)
}.sumOf { it.join() }
},
pool,
)
// 2 subtask
val future2 = CompletableFuture.supplyAsync(
{
(1..5).map {
CompletableFuture.supplyAsync(SingleExternalCall(), pool)
}.sumOf { it.join() }
},
pool,
)
// aggregate
future1.join() + future2.join()
},
pool,
)
println(future.join())
}

class SingleExternalCall : Supplier {

override fun get(): Int {
Thread.sleep(5000)
return counter.incrementAndGet().toInt()
}
}

companion object {

private val counter = AtomicLong()
}
}
< /code>
Wenn Sie versuchen, sie auszuführen - hängt die Anwendung.
Die Ursache ist für mich jetzt klar. Zunächst werden 3 Threads erstellt.sumOf { it.join() }

join

Executors.newWorkStealingPool()

Ich habe ein [url=viewtopic.php?t=15738]Problem[/url] in meiner Bewerbung und künstlichem Beispiel wird so aussehen: < /p>
[code]fun main(args: Array) {
val start = System.currentTimeMillis()
Internal().doWork()
println("Duration is ${(System.currentTimeMillis() - start)/1000} sec")
}

class Internal {

fun doWork() {

val pool = ThreadPoolExecutor(
3, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
ArrayBlockingQueue(1000),
)

val future = CompletableFuture.supplyAsync(
{
// 1 subtask
val future1 = CompletableFuture.supplyAsync(
{
(1..10).map {
CompletableFuture.supplyAsync(SingleExternalCall(), pool)
}.sumOf { it.join() }
},
pool,
)
// 2 subtask
val future2 = CompletableFuture.supplyAsync(
{
(1..5).map {
CompletableFuture.supplyAsync(SingleExternalCall(), pool)
}.sumOf { it.join() }
},
pool,
)
// aggregate
future1.join() + future2.join()
},
pool,
)
println(future.join())
}

class SingleExternalCall : Supplier {

override fun get(): Int {
Thread.sleep(5000)
return counter.incrementAndGet().toInt()
}
}

companion object {

private val counter = AtomicLong()
}
}
< /code>
Wenn Sie versuchen, sie auszuführen - hängt die Anwendung.
Die Ursache ist für mich jetzt klar. Zunächst werden 3 Threads erstellt.sumOf { it.join() }[/code]
[*] Der dritte Thread wird auf Aggregation zukünftig warten1.join () + Future2.Join ()

Alle anderen Aufgaben werden alle anderen Aufgaben in My -Beispielen, sodass die Taskwarteschlange sind, nicht so lange als das Betrag von Taks in My -Beispielen). Applicaiton verfügt also über keine Ressourcen, um nützliche Arbeiten zu erstellen. Aber wie viele? In einem Moment werden wir durch die Menge an Thread
[*] Wir können versuchen, Executors zu verwenden. Jedes Mal, wenn wir einen zusätzlichen Thread benötigen (wenn es keinen kostenlosen Thread im Inneren gibt), können wir versuchen, Executors zu verwenden. Newworkstealingpool ()

konzeptionell sieht konzeptionell aus. Aufgaben.[code]join[/code] und sieht so aus, als ob es erforderlich ist, die gesamte Anwendung umzuschreiben (ich möchte sie vermeiden, wenn es möglich ist)
Also möchte ich wissen:

[*] Ist es eine gute Idee, Forkjoinpool //code>//code>/[code]Executors.newWorkStealingPool()[/code] In meinem Beispiel?
Wenn ja - welche API ist die beste?>