2.7 Java API - Using GPars from Java - Reference Documentation
Authors: The Whole GPars Gang
Version: 1.0.0
2.7 Java API - Using GPars from Java
Using GPars is very addictive, I guarantee. Once you get hooked you won't be able to code without it. May the world force you to write code in Java, you will still be able to benefit from most of GPars features.Java API specifics
Some parts of GPars are irrelevant in Java and it is better to use the underlying Java libraries directly:- Parallel Collection - use jsr-166y library's Parallel Array directly
- Fork/Join - use jsr-166y library's Fork/Join support directly
- Asynchronous functions - use Java executor services directly
GPars Closures in Java API
To overcome the lack of closures as a language element in Java and to avoid forcing users to use Groovy closures directly through the Java API, a few handy wrapper classes have been provided to help you define callbacks, actor body or dataflow tasks.- groovyx.gpars.MessagingRunnable - used for single-argument callbacks or actor body
- groovyx.gpars.ReactorMessagingRunnable - used for ReactiveActor body
- groovyx.gpars.DataflowMessagingRunnable - used for dataflow operators' body
Actors
The DynamicDispatchActor as well as the ReactiveActor classes can be used just like in Groovy:import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.DynamicDispatchActor; public class StatelessActorDemo { public static void main(String[] args) throws InterruptedException { final MyStatelessActor actor = new MyStatelessActor(); actor.start(); actor.send("Hello"); actor.sendAndWait(10); actor.sendAndContinue(10.0, new MessagingRunnable<String>() { @Override protected void doRun(final String s) { System.out.println("Received a reply " + s); } }); } } class MyStatelessActor extends DynamicDispatchActor { public void onMessage(final String msg) { System.out.println("Received " + msg); replyIfExists("Thank you"); } public void onMessage(final Integer msg) { System.out.println("Received a number " + msg); replyIfExists("Thank you"); } public void onMessage(final Object msg) { System.out.println("Received an object " + msg); replyIfExists("Thank you"); } }
import groovy.lang.Closure; import groovyx.gpars.ReactorMessagingRunnable; import groovyx.gpars.actor.Actor; import groovyx.gpars.actor.ReactiveActor;public class ReactorDemo { public static void main(final String[] args) throws InterruptedException { final Closure handler = new ReactorMessagingRunnable<Integer, Integer>() { @Override protected Integer doRun(final Integer integer) { return integer * 2; } }; final Actor actor = new ReactiveActor(handler); actor.start(); System.out.println("Result: " + actor.sendAndWait(1)); System.out.println("Result: " + actor.sendAndWait(2)); System.out.println("Result: " + actor.sendAndWait(3)); } }
Convenience factory methods
Obviously, all the essential factory methods to build actors quickly are available where you'd expect them.import groovy.lang.Closure; import groovyx.gpars.ReactorMessagingRunnable; import groovyx.gpars.actor.Actor; import groovyx.gpars.actor.Actors;public class ReactorDemo { public static void main(final String[] args) throws InterruptedException { final Closure handler = new ReactorMessagingRunnable<Integer, Integer>() { @Override protected Integer doRun(final Integer integer) { return integer * 2; } }; final Actor actor = Actors.reactor(handler); System.out.println("Result: " + actor.sendAndWait(1)); System.out.println("Result: " + actor.sendAndWait(2)); System.out.println("Result: " + actor.sendAndWait(3)); } }
Agents
import groovyx.gpars.MessagingRunnable; import groovyx.gpars.agent.Agent; public class AgentDemo { public static void main(final String[] args) throws InterruptedException { final Agent counter = new Agent<Integer>(0); counter.send(10); System.out.println("Current value: " + counter.getVal()); counter.send(new MessagingRunnable<Integer>() { @Override protected void doRun(final Integer integer) { counter.updateValue(integer + 1); } }); System.out.println("Current value: " + counter.getVal()); } }
Dataflow Concurrency
Both DataflowVariables and DataflowQueues can be used from Java without any hiccups. Just avoid the handy overloaded operators and go straight to the methods, like bind , whenBound , getVal and other. You may also continue using dataflow tasks passing to them instances of Runnable or Callable just like groovy Closure .import groovyx.gpars.MessagingRunnable; import groovyx.gpars.dataflow.DataflowVariable; import groovyx.gpars.group.DefaultPGroup;import java.util.concurrent.Callable;public class DataflowTaskDemo { public static void main(final String[] args) throws InterruptedException { final DefaultPGroup group = new DefaultPGroup(10); final DataflowVariable a = new DataflowVariable(); group.task(new Runnable() { public void run() { a.bind(10); } }); final DataflowVariable result = group.task(new Callable() { public Object call() throws Exception { return (Integer)a.getVal() + 10; } }); result.whenBound(new MessagingRunnable<Integer>() { @Override protected void doRun(final Integer integer) { System.out.println("arguments = " + integer); } }); System.out.println("result = " + result.getVal()); } }
Dataflow operators
The sample below should illustrate the main differences between Groovy and Java API for dataflow operators.- Use the convenience factory methods accepting list of channels to create operators or selectors
- Use DataflowMessagingRunnable to specify the operator body
- Call getOwningProcessor() to get hold of the operator from within the body in order to e.g. bind output values
import groovyx.gpars.DataflowMessagingRunnable; import groovyx.gpars.dataflow.Dataflow; import groovyx.gpars.dataflow.DataflowQueue; import groovyx.gpars.dataflow.operator.DataflowProcessor;import java.util.Arrays; import java.util.List;public class DataflowOperatorDemo { public static void main(final String[] args) throws InterruptedException { final DataflowQueue stream1 = new DataflowQueue(); final DataflowQueue stream2 = new DataflowQueue(); final DataflowQueue stream3 = new DataflowQueue(); final DataflowQueue stream4 = new DataflowQueue(); final DataflowProcessor op1 = Dataflow.selector(Arrays.asList(stream1), Arrays.asList(stream2), new DataflowMessagingRunnable(1) { @Override protected void doRun(final Object… objects) { getOwningProcessor().bindOutput(2*(Integer)objects[0]); } }); final List secondOperatorInput = Arrays.asList(stream2, stream3); final DataflowProcessor op2 = Dataflow.operator(secondOperatorInput, Arrays.asList(stream4), new DataflowMessagingRunnable(2) { @Override protected void doRun(final Object… objects) { getOwningProcessor().bindOutput((Integer) objects[0] + (Integer) objects[1]); } }); stream1.bind(1); stream1.bind(2); stream1.bind(3); stream3.bind(100); stream3.bind(100); stream3.bind(100); System.out.println("Result: " + stream4.getVal()); System.out.println("Result: " + stream4.getVal()); System.out.println("Result: " + stream4.getVal()); op1.stop(); op2.stop(); } }