eric/OnJava8-Examples
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Completely refactored

Browse Source
This commit is contained in:
Bruce Eckel 2017-01-13 07:03:28 -08:00
parent 3e6025f135
commit fac3233fdd
2 changed files with 104 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
lowlevel/DelayQueueDemo.java
View File

@ -63,14 +63,11 @@ public class DelayQueueDemo {
new Random(47).ints(20, 0, 4000) new Random(47).ints(20, 0, 4000)
.mapToObj(DelayedTask::new), .mapToObj(DelayedTask::new),
// Add the summarizing task: // Add the summarizing task:
Stream.of( Stream.of(new DelayedTask.EndTask(4000)))
new DelayedTask.EndTask(4000)))
.collect(Collectors .collect(Collectors
.toCollection(DelayQueue::new)); .toCollection(DelayQueue::new));
DelayQueue<DelayedTask> delayQueue = while(tasks.size() > 0)
new DelayQueue<>(tasks); tasks.take().run();
while(delayQueue.size() > 0)
delayQueue.take().run();
} }
} }
/* Output: /* Output:

198
lowlevel/PriorityBlockingQueueDemo.java
View File

@ -2,148 +2,142 @@
// (c)2017 MindView LLC: see Copyright.txt // (c)2017 MindView LLC: see Copyright.txt
// We make no guarantees that this code is fit for any purpose. // We make no guarantees that this code is fit for any purpose.
// Visit http://OnJava8.com for more book information. // Visit http://OnJava8.com for more book information.
import java.util.concurrent.*;
import java.util.*; import java.util.*;
import java.util.stream.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import onjava.Nap; import onjava.Nap;
class PrioritizedTask implements class Prioritized implements Comparable<Prioritized> {
Runnable, Comparable<PrioritizedTask> { private static AtomicInteger counter =
private SplittableRandom rand = new SplittableRandom(47); new AtomicInteger();
private static int counter = 0; private final int id = counter.getAndAdd(1);
private final int id = counter++;
private final int priority; private final int priority;
protected static List<PrioritizedTask> sequence = private static List<Prioritized> sequence =
new ArrayList<>(); new CopyOnWriteArrayList<>();
public PrioritizedTask(int priority) { public Prioritized(int priority) {
this.priority = priority; this.priority = priority;
sequence.add(this); sequence.add(this);
} }
@Override @Override
public int compareTo(PrioritizedTask arg) { public int compareTo(Prioritized arg) {
return priority < arg.priority ? 1 : return priority < arg.priority ? 1 :
(priority > arg.priority ? -1 : 0); (priority > arg.priority ? -1 : 0);
} }
@Override @Override
public void run() {
new Nap(rand.nextInt(250));
System.out.println(this);
}
@Override
public String toString() { public String toString() {
return String.format("[%1$-3d]", priority) + return String.format(
" Task " + id; "[%d] Prioritized %d", priority, id);
} }
public String summary() { public void displaySequence() {
return "(" + id + ":" + priority + ")";
}
public static
class EndSentinel extends PrioritizedTask {
private ExecutorService exec;
public EndSentinel(ExecutorService e) {
super(-1); // Lowest priority in this program
exec = e;
}
@Override
public void run() {
int count = 0; int count = 0;
for(PrioritizedTask pt : sequence) { for(Prioritized pt : sequence) {
System.out.print(pt.summary()); System.out.printf("(%d:%d)", pt.id, pt.priority);
if(++count % 5 == 0) if(++count % 5 == 0)
System.out.println(); System.out.println();
} }
System.out.println();
System.out.println(this + " Calling shutdownNow()");
exec.shutdownNow();
} }
public static class EndSentinel extends Prioritized {
public EndSentinel() { super(-1); }
} }
} }
class PrioritizedTaskProducer implements Runnable { class Producer implements Runnable {
private SplittableRandom rand = new SplittableRandom(47); private static AtomicInteger seed =
private Queue<Runnable> queue; new AtomicInteger(47);
private ExecutorService exec; private SplittableRandom rand =
public PrioritizedTaskProducer( new SplittableRandom(seed.getAndAdd(10));
Queue<Runnable> q, ExecutorService e) { private Queue<Prioritized> queue;
public Producer(Queue<Prioritized> q) {
queue = q; queue = q;
exec = e; // Used for EndSentinel
} }
@Override @Override
public void run() { public void run() {
// Unbounded queue; never blocks. rand.ints(10, 0, 20)
// Fill it up fast with random priorities: .mapToObj(Prioritized::new)
for(int i = 0; i < 20; i++) { .peek(p -> new Nap(rand.nextInt(
queue.add(new PrioritizedTask(rand.nextInt(10))); PriorityBlockingQueueDemo.NAPTIME)))
new Nap(10); .forEach(p -> queue.add(p));
} queue.add(new Prioritized.EndSentinel());
// Trickle in highest-priority jobs:
for(int i = 0; i < 10; i++) {
new Nap(250);
queue.add(new PrioritizedTask(10));
}
// Add jobs, lowest priority first:
for(int i = 0; i < 10; i++)
queue.add(new PrioritizedTask(i));
// A sentinel to stop all the tasks:
queue.add(new PrioritizedTask.EndSentinel(exec));
System.out.println(
"Finished PrioritizedTaskProducer");
} }
} }
class PrioritizedTaskConsumer implements Runnable { class Consumer implements Runnable {
private PriorityBlockingQueue<Runnable> q; private PriorityBlockingQueue<Prioritized> q;
public PrioritizedTaskConsumer( private SplittableRandom rand =
PriorityBlockingQueue<Runnable> q) { new SplittableRandom(47);
public
Consumer(PriorityBlockingQueue<Prioritized> q) {
this.q = q; this.q = q;
} }
@Override @Override
public void run() { public void run() {
while(true) {
try { try {
while(!Thread.interrupted()) Prioritized pt = q.take();
// Use current thread to run the task: System.out.println(pt);
q.take().run(); if(pt instanceof Prioritized.EndSentinel) {
} catch(InterruptedException e) { pt.displaySequence();
// Acceptable way to exit break;
}
new Nap(rand.nextInt(
PriorityBlockingQueueDemo.NAPTIME));
} catch(InterruptedException e) {
throw new RuntimeException(e);
}
} }
System.out.println(
"Finished PrioritizedTaskConsumer");
} }
} }
public class PriorityBlockingQueueDemo { public class PriorityBlockingQueueDemo {
public static void static final int NAPTIME = 50;
main(String[] args) throws Exception { public static void main(String[] args) {
ExecutorService es = Executors.newCachedThreadPool(); PriorityBlockingQueue<Prioritized> queue =
PriorityBlockingQueue<Runnable> queue =
new PriorityBlockingQueue<>(); new PriorityBlockingQueue<>();
es.execute(new PrioritizedTaskProducer(queue, es)); CompletableFuture.runAsync(new Producer(queue));
es.execute(new PrioritizedTaskConsumer(queue)); CompletableFuture.runAsync(new Producer(queue));
CompletableFuture.runAsync(new Producer(queue));
CompletableFuture.runAsync(new Consumer(queue))
.join();
} }
} }
/* Output: (First and Last 12 Lines) /* Output:
[9 ] Task 5 [12] Prioritized 1
[8 ] Task 7 [17] Prioritized 2
[10 ] Task 20 [15] Prioritized 0
[8 ] Task 8 [18] Prioritized 17
[10 ] Task 21 [17] Prioritized 10
[7 ] Task 4 [16] Prioritized 20
[10 ] Task 22 [16] Prioritized 16
[7 ] Task 19 [15] Prioritized 15
[10 ] Task 23 [14] Prioritized 8
[7 ] Task 11 [14] Prioritized 13
[10 ] Task 24 [13] Prioritized 12
[7 ] Task 1 [12] Prioritized 19
...________...________...________...________... [12] Prioritized 14
[0 ] Task 16 [11] Prioritized 6
(0:5)(1:7)(2:1)(3:0)(4:7) [11] Prioritized 22
(5:9)(6:6)(7:8)(8:8)(9:3) [11] Prioritized 4
(10:0)(11:7)(12:0)(13:5)(14:2) [11] Prioritized 31
(15:4)(16:0)(17:2)(18:1)(19:7) [10] Prioritized 30
(20:10)(21:10)(22:10)(23:10)(24:10) [10] Prioritized 26
(25:10)(26:10)(27:10)(28:10)(29:10) [8] Prioritized 18
(30:0)(31:1)(32:2)(33:3)(34:4) [8] Prioritized 23
(35:5)(36:6)(37:7)(38:8)(39:9) [8] Prioritized 9
(40:-1) [6] Prioritized 24
[-1 ] Task 40 Calling shutdownNow() [3] Prioritized 3
Finished PrioritizedTaskConsumer [2] Prioritized 29
[1] Prioritized 7
[0] Prioritized 27
[0] Prioritized 5
[0] Prioritized 21
[0] Prioritized 11
[-1] Prioritized 25
(0:15)(2:17)(1:12)(3:3)(4:11)
(5:0)(6:11)(7:1)(8:14)(9:8)
(10:17)(11:0)(12:13)(13:14)(14:12)
(15:15)(16:16)(17:18)(18:8)(19:12)
(20:16)(21:0)(22:11)(23:8)(24:6)
(25:-1)(26:10)(27:0)(28:-1)(29:2)
(30:10)(31:11)(32:-1)
*/ */