{"id":5593,"date":"2018-12-02T14:00:25","date_gmt":"2018-12-02T14:00:25","guid":{"rendered":"http:\/\/writeasync.net\/?p=5593"},"modified":"2018-11-25T00:45:38","modified_gmt":"2018-11-25T00:45:38","slug":"a-simple-message-bus-java-edition","status":"publish","type":"post","link":"http:\/\/writeasync.net\/?p=5593","title":{"rendered":"A simple message bus: Java edition"},"content":{"rendered":"

In the previous post<\/a>, we looked at competing message bus implementations in C# and C++. How about we give Java a try now?<\/p>\n

Converting to Java syntax and style conventions, the SendOneSubscriber<\/code> test should look like this:<\/p>\n

\r\npackage com.example;\r\n\r\nimport java.util.function.*;\r\nimport java.util.*;\r\nimport org.junit.Test;\r\n\r\nimport static org.hamcrest.Matchers.*;\r\nimport static org.hamcrest.MatcherAssert.assertThat;\r\n\r\npublic class MessageBusTest {\r\n    \/\/ . . .\r\n\r\n    @Test\r\n    public void testSendOneSubscriber() {\r\n        MessageBus bus = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<MyMessage> subscriber = m -> received.add(m.getText());\r\n\r\n        bus.subscribe(subscriber);\r\n        bus.send(new MyMessage("hello"));\r\n\r\n        assertThat(received, contains("hello"));\r\n    }\r\n\r\n    \/\/ . . .\r\n}\r\n<\/pre>\n
Alas, there is a big problem here. Let’s first review what the API for MessageBus<\/code> should look like given the test above:<\/p>\n
\r\npackage com.example;\r\n\r\nimport java.util.function.*;\r\n\r\npublic class MessageBus {\r\n    public MessageBus() {\r\n        \/* . . . *\/\r\n    }\r\n\r\n    public <M> void subscribe(Consumer<M> subscriber) {\r\n        \/* . . . *\/\r\n    }\r\n\r\n    public <M> void send(M message) {\r\n        \/* . . . *\/\r\n    }\r\n}\r\n<\/pre>\n
Ideally, inside those generic methods we would retrieve the type information about M<\/code> and use this as a lookup key for a HashMap<\/code>. Unfortunately, the implementation of Java generics<\/a> is based on type erasure<\/a>. By definition, we have no runtime information available to us about what type M<\/code> actually was in the source code. Techniques such as TypeLiteral in various Java libraries<\/a> exist to solve related type erasure problems but cannot seem to help us in this instance.<\/p>\n
As yet, I have not found a way to make this type of generically typed API work as is. We will simply have to alter the API and force the caller to pass a type token<\/a> as an additional parameter:<\/p>\n
\r\npackage com.example;\r\n\r\nimport java.util.function.*;\r\n\r\npublic class MessageBus {\r\n    public MessageBus() {\r\n        \/* . . . *\/\r\n    }\r\n\r\n    public <M> void subscribe(Consumer<M> subscriber, Class<M> type) {\r\n        \/* . . . *\/\r\n    }\r\n\r\n    public <M> void send(M message, Class<M> type) {\r\n        \/* . . . *\/\r\n    }\r\n}\r\n<\/pre>\n
The test would change as follows:<\/p>\n
\r\n    \/* . . . *\/\r\n\r\n    @Test\r\n    public void testSendOneSubscriber() {\r\n        MessageBus bus = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<MyMessage> subscriber = m -> received.add(m.getText());\r\n\r\n        bus.subscribe(subscriber, MyMessage.class);\r\n        bus.send(new MyMessage("hello"), MyMessage.class);\r\n\r\n        assertThat(received, contains("hello"));\r\n    }\r\n\r\n    \/* . . . *\/\r\n<\/pre>\n
It’s a bit ugly but should be type safe. For example, it is not possible to compile this code:<\/p>\n
\r\n        Consumer<String> subscriber = m -> \/* . . . *\/;\r\n\r\n        \/\/ ERROR: The method subscribe(M, Class<M>) in the type MessageBus is not\r\n        \/\/ applicable for the arguments (Consumer<String>, Class<Integer>)\r\n        bus.subscribe(subscriber, Integer.class);\r\n<\/pre>\n
Here is the entire test suite translated to Java, using the convenient Hamcrest matcher objects<\/a> for fluent syntax:<\/p>\n
\r\npackage com.example;\r\n\r\nimport java.util.function.*;\r\nimport java.util.*;\r\nimport org.junit.Test;\r\n\r\nimport static org.hamcrest.Matchers.*;\r\nimport static org.hamcrest.MatcherAssert.assertThat;\r\n\r\npublic class MessageBusTest {\r\n    @Test\r\n    public void testSendZeroSubscribers() {\r\n        MessageBus bus = new MessageBus();\r\n\r\n        boolean threwException = false;\r\n        try {\r\n            bus.send(new MyMessage("hello"), MyMessage.class);\r\n        } catch (Exception e) {\r\n            threwException = true;\r\n        }\r\n\r\n        assertThat(threwException, is(false));\r\n    }\r\n\r\n    @Test\r\n    public void testSendOneSubscriber() {\r\n        MessageBus bus = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<MyMessage> subscriber = m -> received.add(m.getText());\r\n\r\n        bus.subscribe(subscriber, MyMessage.class);\r\n        bus.send(new MyMessage("hello"), MyMessage.class);\r\n\r\n        assertThat(received, contains("hello"));\r\n    }\r\n\r\n    @Test\r\n    public void testSendTwoSubscribers() {\r\n        MessageBus bus = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<MyMessage> subscriber1 = m -> received.add(m.getText() + "1");\r\n        Consumer<MyMessage> subscriber2 = m -> received.add(m.getText() + "2");\r\n\r\n        bus.subscribe(subscriber1, MyMessage.class);\r\n        bus.subscribe(subscriber2, MyMessage.class);\r\n        bus.send(new MyMessage("hello"), MyMessage.class);\r\n\r\n        assertThat(received, contains("hello1", "hello2"));\r\n    }\r\n\r\n    @Test\r\n    public void testSendTwoOneSubscriberEach() {\r\n        MessageBus bus = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<MyMessage> subscriber1 = m -> received.add(m.getText() + "1");\r\n        Consumer<MyOtherMessage> subscriber2 = m -> received.add(m.getText() + "2");\r\n\r\n        bus.subscribe(subscriber1, MyMessage.class);\r\n        bus.subscribe(subscriber2, MyOtherMessage.class);\r\n        bus.send(new MyMessage("one-hello"), MyMessage.class);\r\n        bus.send(new MyOtherMessage("two-hello"), MyOtherMessage.class);\r\n\r\n        assertThat(received, contains("one-hello1", "two-hello2"));\r\n    }\r\n\r\n    @Test\r\n    public void testSendTwoSimpleTypesOneSubscriberEach() {\r\n        MessageBus bus = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<String> subscriber1 = m -> received.add("S=" + m);\r\n        Consumer<Integer> subscriber2 = m -> received.add("N=" + m);\r\n\r\n        bus.subscribe(subscriber1, String.class);\r\n        bus.subscribe(subscriber2, Integer.class);\r\n        bus.send("xyz", String.class);\r\n        bus.send(123, Integer.class);\r\n\r\n        assertThat(received, contains("S=xyz", "N=123"));\r\n    }\r\n\r\n    @Test\r\n    public void testSendTwoInstancesThreeSubscribersEachBeforeAndAfter() {\r\n        MessageBus bus1 = new MessageBus();\r\n        MessageBus bus2 = new MessageBus();\r\n        ArrayList<String> received = new ArrayList<String>();\r\n        Consumer<String> subscriber1 = m -> received.add("S1=" + m);\r\n        Consumer<String> subscriber2 = m -> received.add("S2=" + m);\r\n        Consumer<String> subscriber3 = m -> received.add("S3=" + m);\r\n        Consumer<String> subscriber4 = m -> received.add("S4=" + m);\r\n        Consumer<Integer> subscriber5 = m -> received.add("S5=" + m);\r\n        Consumer<Integer> subscriber6 = m -> received.add("S6=" + m);\r\n\r\n        bus1.subscribe(subscriber1, String.class);\r\n        bus2.subscribe(subscriber2, String.class);\r\n        bus1.send("aaa", String.class);\r\n        bus2.send("bbb", String.class);\r\n        bus1.subscribe(subscriber3, String.class);\r\n        bus2.subscribe(subscriber4, String.class);\r\n        bus1.send("ccc", String.class);\r\n        bus2.send("ddd", String.class);\r\n        bus1.subscribe(subscriber5, Integer.class);\r\n        bus2.subscribe(subscriber6, Integer.class);\r\n        bus1.send(1, Integer.class);\r\n        bus2.send(2, Integer.class);\r\n\r\n        assertThat(received, contains("S1=aaa", "S2=bbb", "S1=ccc", "S3=ccc", "S2=ddd", "S4=ddd", "S5=1", "S6=2"));\r\n    }\r\n\r\n    private class MyMessage {\r\n        private final String text;\r\n\r\n        public MyMessage(String text) {\r\n            this.text = text;\r\n        }\r\n\r\n        public String getText() {\r\n            return this.text;\r\n        }\r\n    }\r\n\r\n    private class MyOtherMessage extends MyMessage {\r\n        public MyOtherMessage(String text) {\r\n            super(text);\r\n        }\r\n    }\r\n}\r\n<\/pre>\n
These tests are somewhat closer to the C# implementation than the C++, in large part due to Hamcrest<\/a>. There doesn’t seem to be a nice fluent way to assert for exceptions<\/a>, though. FYI, since JUnit 4.12 includes an older version of Hamcrest<\/a>, I updated my build.gradle<\/code> file for the java-library<\/code> plugin<\/a> to include Hamcrest 2.0 like so:<\/p>\n
\r\ndependencies {\r\n    \/\/ . . .\r\n    \/\/ Use JUnit test framework\r\n    testImplementation 'junit:junit:4.12'\r\n    \r\n    \/\/ Use Hamcrest 2.0\r\n    testImplementation('org.hamcrest:java-hamcrest:2.0.0.0')\r\n}\r\n<\/pre>\n
Finally, let’s look at the implementation code:<\/p>\n
\r\npackage com.example;\r\n\r\nimport java.util.*;\r\nimport java.util.function.*;\r\n\r\npublic class MessageBus {\r\n    private final HashMap<Class<?>, ArrayList<Consumer<Object>>> map;\r\n\r\n    public MessageBus() {\r\n        this.map = new HashMap<Class<?>, ArrayList<Consumer<Object>>>();\r\n    }\r\n\r\n    public <M> void subscribe(Consumer<M> subscriber, Class<M> type) {\r\n        ArrayList<Consumer<Object>> list = this.map.get(type);\r\n        if (list == null) {\r\n            list = new ArrayList<Consumer<Object>>();\r\n            this.map.put(type, list);\r\n        }\r\n\r\n        @SuppressWarnings("unchecked")\r\n        Consumer<Object> f = m -> subscriber.accept((M) m);\r\n        list.add(f);\r\n    }\r\n\r\n    public <M> void send(M message, Class<M> type) {\r\n        ArrayList<Consumer<Object>> list = this.map.get(type);\r\n        if (list != null) {\r\n            for (Consumer<Object> f : list) {\r\n                f.accept(message);\r\n            }\r\n        }\r\n    }\r\n}\r\n<\/pre>\n
Like C++, Java has no multicast delegates, so we’re back to the “list of functions” solution. Other than that, the code is roughly as concise as the C# version. Note the use of the unbounded wildcard type<\/a> Class<?><\/code> which is roughly equivalent to System.Type<\/code> in the C# solution, and the need to suppress the “unchecked” warning<\/a> due to the Object<\/code> to M<\/code> cast (we know it is safe here, but due to type erasure, the runtime does not).<\/p>\n
Of course, the exercise wouldn’t be complete without the ported benchmark. Here it is:<\/p>\n
\r\npackage com.example;\r\n\r\nimport java.util.function.*;\r\n\r\npublic class Program {\r\n    public static void main(String[] args) {\r\n        benchmark(1);\r\n        benchmark(16);\r\n        benchmark(256);\r\n        benchmark(4096);\r\n        benchmark(65536);\r\n        benchmark(262144);\r\n        benchmark(524288);\r\n        benchmark(524288);\r\n        benchmark(524288);\r\n        benchmark(524288);\r\n        benchmark(524288);\r\n    }\r\n\r\n    private static void benchmark(int iterations) {\r\n        LongWrapper count = new LongWrapper();\r\n        Consumer<String> subscriber = m -> count.increment(m.length());\r\n        MessageBus bus = new MessageBus();\r\n\r\n        bus.subscribe(subscriber, String.class);\r\n\r\n        Consumer<Integer> func = n -> {\r\n            for (int i = 0; i < n; ++i) {\r\n                for (char c = 'A'; c <= 'Z'; ++c) {\r\n                    String msg = new String(c + "");\r\n                    bus.send(msg, String.class);\r\n                }\r\n            }\r\n        };\r\n\r\n        long start = System.nanoTime();\r\n\r\n        func.accept(iterations);\r\n\r\n        long end = System.nanoTime();\r\n        double nsecPerOp = 1.0 * (end - start) \/ iterations;\r\n\r\n        System.out.printf("Average operation time: %f ns (var count = %d)\\n", nsecPerOp, count.get());\r\n    }\r\n\r\n    private static class LongWrapper {\r\n        private long value;\r\n\r\n        public LongWrapper() {\r\n            this.value = 0L;\r\n        }\r\n\r\n        public void increment(long inc) {\r\n            this.value += inc;\r\n        }\r\n\r\n        public long get() {\r\n            return this.value;\r\n        }\r\n    }\r\n}\r\n<\/pre>\n
Remember, Java doesn’t have unsigned types<\/a> and local variables cannot be mutated within a lambda expression<\/a> (necessitating the LongWrapper<\/code> workaround that is otherwise done “for free” in the .NET case<\/a>). The results:<\/p>\n
\r\nAverage operation time: 176769.000000 ns (var count = 26)\r\nAverage operation time: 62644.437500 ns (var count = 416)\r\nAverage operation time: 20396.523438 ns (var count = 6656)\r\nAverage operation time: 8717.085205 ns (var count = 106496)\r\nAverage operation time: 1592.115631 ns (var count = 1703936)\r\nAverage operation time: 1398.114811 ns (var count = 6815744)\r\nAverage operation time: 1432.140196 ns (var count = 13631488)\r\nAverage operation time: 852.286091 ns (var count = 13631488)\r\nAverage operation time: 816.319372 ns (var count = 13631488)\r\nAverage operation time: 835.554661 ns (var count = 13631488)\r\nAverage operation time: 869.618492 ns (var count = 13631488)\r\n<\/pre>\n
What we traded in API convenience seems to have paid back in runtime performance. The Java implementation which stabilizes at ~850 ns obviously won’t beat C++ but is apparently faster than the C# implementation by almost 25%.<\/p>\n
Who says Java is slow<\/a>?!<\/p>\n","protected":false},"excerpt":{"rendered":"
In the previous post, we looked at competing message bus implementations in C# and C++. How about we give Java a try now? Converting to Java syntax and style conventions, the SendOneSubscriber test should look like this: package com.example; import… <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[91,104],"tags":[],"class_list":["post-5593","post","type-post","status-publish","format-standard","hentry","category-design","category-performance"],"_links":{"self":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/5593","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=5593"}],"version-history":[{"count":3,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/5593\/revisions"}],"predecessor-version":[{"id":5596,"href":"http:\/\/writeasync.net\/index.php?rest_route=\/wp\/v2\/posts\/5593\/revisions\/5596"}],"wp:attachment":[{"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=5593"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=5593"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/writeasync.net\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=5593"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}