{"id":12831,"date":"2013-05-16T16:00:46","date_gmt":"2013-05-16T13:00:46","guid":{"rendered":"http:\/\/www.javacodegeeks.com\/?p=12831"},"modified":"2013-05-16T09:06:11","modified_gmt":"2013-05-16T06:06:11","slug":"asynchronous-cdi-events","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2013\/05\/asynchronous-cdi-events.html","title":{"rendered":"Asynchronous CDI Events"},"content":{"rendered":"

Few days ago, during our regular code review, one of my colleagues raised a question what would happen \u2014 and if it\u2019s even possible \u2014 when a CDI Observer (so a method with parameter annotated @Observes<\/code>) would be invoked multiple times at the same time for different event instances. In other words, after producing few events, is it possible that the following method will be processed by more than one thread at the same time:
\n 
\n 
\n 
\n 
\n <\/p>\n

public void observe(@Observes MyEvent myEvent) { ... }<\/pre>\n
After thinking about it I\u2019ve decided to run a few tests and describe results in this post.<\/p>\n
First results: it occurred that CDI Events are fired in synchronous mode<\/strong> which was a bit surprise for me. Why?<\/p>\n
Up to this time, I\u2019ve seen it this way: CDI observers allows me to very cleanly separate event producer from event consumer, so I don\u2019t have any hard-coded registering of listeners, maintaining a list of listeners and manually informing them. The CDI container does everything for me.<\/p>\n
Therefore, if we have cleanly separated producers from consumers I thought that there exists some kind of event bus running in specialized, thread executors pool which is responsible for mediation between registered events and invoked observers methods. I guess I based this assumption on other event\/listeners solutions like Google Guava EventBus<\/a>. They give you a chance to define if you want to use synchronous (default, EventBus<\/a>) or asynchronous event dispatchers (AsyncEventBus<\/a>.)<\/p>\n
Moreover, if EJBs are both: producer and consumer, I assume it would have the same features as asynchronous EJB calls when it comes to transactions. The only possible JTA transaction attribute for asynchronous event observer would be: REQUIRED<\/code>, REQUIRES_NEW<\/code> or NOT_SUPPORTED<\/code>.<\/p>\n
Now that\u2019s all how I expected<\/strong> it to work which seems to be quite different from the current status<\/strong>. The real life shows that CDI events are synchronous.<\/p>\n
There is an issue for making asynchronous events available in CDI 1.1<\/a> but I\u2019m not sure what is the current status of this feature and didn\u2019t find a word about it in CDI 1.1 (part of Java EE 7).<\/p>\n
Let\u2019s see how we can deal with it on our own.<\/p>\n
Table of contents<\/h4>\n
    \n
  1. Default Synchronous Events<\/a><\/li>\n
  2. Solution 1 \u2013 CDI Producer and Singleton EJB as Receiver<\/a><\/li>\n
  3. Solution 2 \u2013 Use Singleton EJB as Receiver With Read Lock<\/a><\/li>\n
  4. Solution 3 \u2013 EJB Producer and CDI Consumer<\/a><\/li>\n
  5. Solution 4 \u2013 EJB Producer and EJB Consumer<\/a><\/li>\n
  6. Solution 4 vs Solution 2<\/a><\/li>\n
  7. Solution 5 \u2013 EJB Producer and CDI Consumer II<\/a><\/li>\n
  8. Solution 6 \u2013 CDI With JMS<\/a><\/li>\n
  9. Conclusion<\/a><\/li>\n<\/ol>\n
    Default Synchronous Events<\/h2>\n
    Let\u2019s start with basic example showing the problem. Take a look at the code \u2013 first, the CDI Bean producer:<\/p>\n
    @Path(\"\/produce\")\r\npublic class EventGenerator {\r\n\r\n    @Inject\r\n    private Logger logger;\r\n\r\n    @Inject\r\n    private Event<MyEvent> events;\r\n\r\n    @Path(\"\/cdiBean\/{eventsNum}\")\r\n    @GET\r\n    public String generateEvents(@PathParam(\"eventsNum\") int numberOfEventsToGenerate) {\r\n        for (int i = 0; i < numberOfEventsToGenerate; i++) {\r\n            MyEvent event = new MyEvent(i);\r\n\r\n            logger.info(\"Generating Event: \" + event);\r\n            events.fire(event);\r\n        }\r\n        return \"Finished. Generated \" + numberOfEventsToGenerate + \" events.\";\r\n    }\r\n}<\/pre>\n
    MyEvent<\/code> is just some event object which is not really important here. It stores event sequence number that we pass while instantiation.<\/p>\n
    Consumer is a pretty simple CDI Bean:<\/p>\n
    public class EventConsumer {\r\n\r\n    @Inject\r\n    private Logger logger;\r\n\r\n    public void consumeEvent(@Observes MyEvent myEvent) throws InterruptedException {\r\n        logger.info(\"Receiving event: \" + myEvent);\r\n\r\n        TimeUnit.MILLISECONDS.sleep(500);\r\n    }\r\n}<\/pre>\n
    Note that I\u2019ve inserted a thread sleep to simulate some long-running event receiver process.<\/p>\n
    Now, let\u2019s run this example by invoking a REST command this EventProducer<\/code> is exposing. The result (running JBoss EAP 6.1 Alpha<\/strong>) would be similar to this:<\/p>\n
    \n
    14:15:59,196 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=0<\/strong>]
    \n14:15:59,197 [com.piotrnowicki.EventConsumer] (http-\/127.0.0.1:8080-1) Receiving event: MyEvent[seqNo=0<\/strong>]
    \n14:15:59,697 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=1<\/strong>]
    \n14:15:59,698 [com.piotrnowicki.EventConsumer] (http-\/127.0.0.1:8080-1) Receiving event: MyEvent[seqNo=1<\/strong>]
    \n14:16:00,199 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=2<\/strong>]
    \n14:16:00,200 [com.piotrnowicki.EventConsumer] (http-\/127.0.0.1:8080-1) Receiving event: MyEvent[seqNo=2<\/strong>]
     
    <\/div> <\/div><\/p>\n<\/blockquote>\n
    It shows the synchronous nature of CDI events \u2013 event producing and consuming takes place in the same thread and one after another.<\/p>\n
    So, how to achieve asynchronous events with CDI?<\/p>\n
    Solution 1 \u2013 CDI Producer and Singleton EJB as Receiver<\/h2>\n
    Producer stays at it was \u2013 a pure CDI bean:<\/p>\n
    @Path(\"\/produce\") public class EventGenerator {\r\n\r\n    @Path(\"\/cdiBean\/{eventsNum}\")\r\n    @GET\r\n    public String generateEvents(@PathParam(\"eventsNum\") int numberOfEventsToGenerate) { ... }\r\n}<\/pre>\n
    Now if you turn your receiver into @Singleton<\/strong> EJB and mark observes method as @Asynchronous<\/strong> like this:<\/p>\n
    @Singleton\r\npublic class EventConsumer {\r\n\r\n    @Asynchronous\r\n    public void consumeEvent(@Observes MyEvent myEvent) throws InterruptedException { ...  }\r\n}<\/pre>\n
    You\u2019ll get the following results:<\/p>\n
    \n
    14:21:19,341 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=0]
    \n14:21:19,343 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=1]
    \n14:21:19,343 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=2]
    \n14:21:19,347<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 2) Receiving event: MyEvent[seqNo=1]
    \n14:21:19,848<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 1) Receiving event: MyEvent[seqNo=0]
    \n14:21:20,350<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 3) Receiving event: MyEvent[seqNo=2]<\/p>\n<\/blockquote>\n
    Events are produced one after another and in separate threads Singleton EJB is serving them one after another (take a look at time of event processing.) That\u2019s because of implicit write-lock for every business methods of Singleton EJB. So this is:<\/p>\n
    Asynchronous: yes<\/strong>
    \nThread-safe observer method: yes<\/strong><\/p>\n
    Solution 2 \u2013 Use Singleton EJB as Receiver With Read Lock<\/h2>\n
    This approach is very similar to Solution 1, however, it gives you a much higher throughput because all events processing takes place in parallel.<\/p>\n
    Our producer stays the same \u2013 it\u2019s a CDI bean:<\/p>\n
    @Path(\"\/produce\")\r\npublic class EventGenerator {\r\n\r\n    @Path(\"\/cdiBean\/{eventsNum}\")\r\n    @GET\r\n    public String generateEvents(@PathParam(\"eventsNum\") int numberOfEventsToGenerate) { ... }\r\n}<\/pre>\n
    Our consumer has @Lock(READ)<\/code> added to its observes method; this makes the magic of being able to serve multiple events at the same time:<\/p>\n
    @Singleton\r\npublic class EventConsumer {\r\n\r\n    @Asynchronous\r\n    @Lock(LockType.READ)\r\n    public void consumeEvent(@Observes MyEvent myEvent) throws InterruptedException { ... }\r\n}<\/pre>\n
    This is what you can get as a result:<\/p>\n
    \n
    14:24:44,202 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=0]
    \n14:24:44,204 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=1]
    \n14:24:44,205 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=2]
    \n14:24:44,207<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 4) Receiving event: MyEvent[seqNo=0]
    \n14:24:44,207<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 6) Receiving event: MyEvent[seqNo=2]
    \n14:24:44,207<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 5) Receiving event: MyEvent[seqNo=1]<\/p>\n<\/blockquote>\n
    Different threads serving events at the same time are giving you a bigger throughput. So this is:<\/p>\n
    Asynchronous: yes<\/strong>
    \nThread-safe observer method: no<\/strong><\/p>\n
    Solution 3 \u2013 EJB Producer and CDI Consumer<\/h2>\n
    CDI allows you to observe events during specific stages of transaction. You specify it using @Observes(during=TransactionPhase...)<\/code>. In our case, we would like the CDI to stack all those events and invoke our observer only after the transaction ends. To do so we need only to add the above attribute to our CDI Bean observer:<\/p>\n
    public class EventConsumer { \r\n   public void consumeEvent(@Observes(during = TransactionPhase.AFTER_COMPLETION) MyEvent myEvent) { ... } \r\n}<\/pre>\n
    Now we just need to make sure we have a running transaction in EventGenerator<\/code> method. We can do it quickly by transforming our CDI Bean into @Stateless<\/code> EJB and using its implicit REQUIRED<\/code> TransactionAttribute like this:<\/p>\n
    @Stateless\r\n@Path(\"\/produce\")\r\npublic class EventGenerator {\r\n\r\n    @Path(\"\/cdiBean\/{eventsNum}\")\r\n    @GET\r\n    public String generateEvents(@PathParam(\"eventsNum\") int numberOfEventsToGenerate) { ... }\r\n}<\/pre>\n
    This is the result we might end with:<\/p>\n
    \n
    14:39:06,776 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=0]
    \n14:39:06,776 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=1]
    \n14:39:06,776 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=2]
    \n14:39:06,778<\/strong> [com.piotrnowicki.EventConsumer] (http-\/127.0.0.1:8080-1) Receiving event: MyEvent[seqNo=2]
    \n14:39:07,279<\/strong> [com.piotrnowicki.EventConsumer] (http-\/127.0.0.1:8080-1) Receiving event: MyEvent[seqNo=1]
    \n14:39:07,780<\/strong> [com.piotrnowicki.EventConsumer] (http-\/127.0.0.1:8080-1) Receiving event: MyEvent[seqNo=0]<\/p>\n<\/blockquote>\n
    EJB EventGenerator<\/code> starts a transaction and CDI bean observer will be invoked in serialized way only after the transaction completes.<\/p>\n
    Asynchronous: yes<\/strong>
    \nThread-safe observer method: yes<\/strong><\/p>\n
    Solution 4 \u2013 EJB Producer and EJB Consumer<\/h2>\n
    This is very similar to Solution 3. Our generator stays the same (Stateless EJB):<\/p>\n
    @Stateless\r\n@Path(\"\/produce\")\r\npublic class EventGenerator {\r\n\r\n    @Path(\"\/cdiBean\/{eventsNum}\")\r\n    @GET\r\n    public String generateEvents(@PathParam(\"eventsNum\") int numberOfEventsToGenerate) { ... }\r\n}<\/pre>\n
    And changes are made to EventConsumer<\/code> which is right now:<\/p>\n
    @Singleton\r\npublic class EventConsumer {\r\n\r\n    @Asynchronous\r\n    @Lock(LockType.READ)\r\n    public void consumeEvent(@Observes(during = TransactionPhase.AFTER_COMPLETION) MyEvent myEvent) throws InterruptedException { ...  }\r\n}<\/pre>\n
    The result might be as follows:<\/p>\n
    \n
    14:44:09,363 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=0]
    \n14:44:09,464 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=1]
    \n14:44:09,564 [com.piotrnowicki.EventGenerator] (http-\/127.0.0.1:8080-1) Generating Event: MyEvent[seqNo=2]
    \n14:44:09,670<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 8) Receiving event: MyEvent[seqNo=2]
    \n14:44:09,670<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 2) Receiving event: MyEvent[seqNo=1]
    \n14:44:09,670<\/strong> [com.piotrnowicki.EventConsumer] (EJB default \u2013 1) Receiving event: MyEvent[seqNo=0]<\/p>\n<\/blockquote>\n
    We\u2019ve used two features here \u2013 one is that the event consumer method is asynchronous and the second one is that the consumer will not be notified before the producer transaction completes. This gives us:<\/p>\n
    Asynchronous: yes<\/strong>
    \nThread-safe observer method: no<\/strong><\/p>\n
    Solution 4 vs Solution 2<\/h2>\n
    Those two solutions seems to be the same. They only differ with consumer\u2019s annotation: @Observes<\/code> vs @Observes(during = TransactionPhase.AFTER_COMPLETION)<\/code>. Moreover they act the same for our test case: they are asynchronous<\/strong> and multiple threads can be processing event receivers at the same time<\/strong>. However, there is one big difference between them.<\/p>\n
    In our test case we fired events one after another. Imagine that there is some other operations between events firing. In such case:<\/p>\n