Applying Reactive Programming with RxJava at GOTO Chicago 2015

Rarely do we have a chance to rewrite an application from scratch with the newest techniques. Generally we must work within our existing architectures evolving and refactoring years of code. This can make it seem reactive programming either can’t be introduced or doesn’t play a role.

Netflix has been doing reactive programming with RxJava in production for several years and only recently embarked on “greenfield” development efforts that are fully async. This talk will leverage the experience of introducing reactive programming into existing imperative, blocking codebases to demonstrate how it can be done and when it can make sense to do so. Not all benefits of reactive programming can be obtained without a greenfield, fully async architecture, but many can. Subjects to be covered will include the mental shift from imperative to declarative, working with blocking IO such as JDBC and RPC, service composition, debugging and unit testing.

Presented at GOTO Chicago 2015 http://gotocon.com/chicago-2015/presentation/Applying%20Reactive%20Programming%20with%20Rx

Video: https://www.youtube.com/watch?v=8OcCSQS0tug

Transcript

1. Ben Christensen Edge Engineering at Netflix @benjchristensen http://techblog.netflix.com/ GOTO Chicago - May 2015 Applying Reactive Programming with RxJava
7. async + callbacks
8. service composition + error handling developer productivity
9. async + callbacks
10. Future<T> Future<List<T>>
11. Future<T> Future<List<T>>
12. → getListOfListOfMovies → forEachList → forEachMovie → getBookmark → getRating → getMetadata → getSimilars → etc
13. Future<T> Future<List<T>> Future<List<Future<T>>>
14. Observable<T> Future<T> Future<List<T>> Future<List<Future<T>>>
15. Everything is a Stream
16. List<Order> getOrders(int customerId) List<Product> getProducts(Order order) ShippingStatus getShippingStatus(Order o, Product p) getOrders → getProducts → getShippingStatus
17. List<Order> getOrders(int customerId) List<Product> getProducts(Order order) ShippingStatus getShippingStatus(Order o, Product p) getOrders → getProducts → getShippingStatus Blocking APIs
18. getOrders → getProducts → getShippingStatus
19. IPC via Apache HTTP Memcached Cassandra Tomcat and Servlets
20. IPC via Apache HTTP Memcached Cassandra Tomcat and Servlets All Blocking
21. Many things could not realistically change
22. We could change our service layer
23. Async Facade
24. Async Facade Blocking Code Servlet Sync/Async Bridge
25. Observable<Order> getOrders(int customerId) Observable<Product> getProducts(Order order) Observable<ShippingStatus> getShippingStatus(Order o, Product p) getOrdersAsync → getProductsAsync → getShippingStatusAsync getOrders → getProducts → getShippingStatus
26. Observable<Order> getOrders(int customerId) Observable<Product> getProducts(Order order) Observable<ShippingStatus> getShippingStatus(Order o, Product p) getOrdersAsync → getProductsAsync → getShippingStatusAsync Observable APIs getOrders → getProducts → getShippingStatus
27. getOrdersAsync(1) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }) .forEach(s -> System.out.println(Thread.currentThread() + " [Async] Shipping Status: " + s));
28. getOrdersAsync(1) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }) .forEach(s -> System.out.println(Thread.currentThread() + " [Async] Shipping Status: " + s)); an async Observable
29. getOrdersAsync(1) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }) .forEach(s -> System.out.println(Thread.currentThread() + " [Async] Shipping Status: " + s)); an async Observable stream
30. getOrdersAsync(1) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }) .forEach(s -> System.out.println(Thread.currentThread() + " [Async] Shipping Status: " + s)); another async Observable stream
31. getOrdersAsync(1) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }) .forEach(s -> System.out.println(Thread.currentThread() + " [Async] Shipping Status: " + s));
32. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) ShippingStatus getShippingStatus(Order o, Product p)
33. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) return Observable.defer(() -> { return Observable.just(getShippingStatus(o, p)); }).subscribeOn(Schedulers.io());
34. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) return Observable.defer(() -> { return Observable.just(getShippingStatus(o, p)); }).subscribeOn(Schedulers.io());
35. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) return Observable.defer(() -> { return Observable.just(getShippingStatus(o, p)); }).subscribeOn(Schedulers.io());
36. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) return Observable.defer(() -> { return Observable.just(getShippingStatus(o, p)); }).subscribeOn(Schedulers.io());
37. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) return Observable.defer(() -> { return Observable.just(getShippingStatus(o, p)); }).subscribeOn(Schedulers.io());
38. Observable<ShippingStatus> getShippingStatusAsync(Order o, Product p) return Observable.defer(() -> { return Observable.just(getShippingStatus(o, p)); }).subscribeOn(Schedulers.io());
39. List<Order> orders = getOrders(1); Observable<Order> orders = getOrdersAsync(1);
40. List<Order> orders = getOrders(1); Observable<Order> orders = getOrdersAsync(1);
41. List<Order> orders = getOrders(1); Observable<Order> orders = getOrdersAsync(1);
42. return Observable.defer(() -> { return Observable.from(getOrders(customerId)); }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
43. return Observable.defer(() -> { return Observable.from(getOrders(customerId)); }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
44. Observable<Order> orders = getOrdersAsync(1); return Observable.defer(() -> { return Observable.from(getOrders(customerId)); }).subscribeOn(Schedulers.io());
45. return Observable.defer(() -> { return Observable.from(getOrders(customerId)); }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
46. return Observable.defer(() -> { return Observable.from(getOrders(customerId)); }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
47. Wrapped with Threads using subscribeOn
48. Wrapped with Threads JDBC?
49. List<Order> orders = getOrders(1); try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); ArrayList<Order> orders = new ArrayList<>(); while (rs.next()) { orders.add(new Order(rs.getInt(1))); } rs.close(); return orders; } catch (Exception e) { throw new RuntimeException(e); }
50. List<Order> orders = getOrders(1); try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); ArrayList<Order> orders = new ArrayList<>(); while (rs.next()) { orders.add(new Order(rs.getInt(1))); } rs.close(); return orders; } catch (Exception e) { throw new RuntimeException(e); }
51. Observable<Order> orders = getOrdersAsync(1); return Observable.<Order> create(o -> { try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); while (rs.next() && !o.isUnsubscribed()) { o.onNext(new Order(rs.getInt(1))); } rs.close(); o.onCompleted(); } catch (Exception e) { o.onError(e); } }).subscribeOn(Schedulers.io());
52. Observable<Order> orders = getOrdersAsync(1); return Observable.<Order> create(o -> { try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); while (rs.next() && !o.isUnsubscribed()) { o.onNext(new Order(rs.getInt(1))); } rs.close(); o.onCompleted(); } catch (Exception e) { o.onError(e); } }).subscribeOn(Schedulers.io());
53. return Observable.<Order> create(o -> { try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); while (rs.next() && !o.isUnsubscribed()) { o.onNext(new Order(rs.getInt(1))); } rs.close(); o.onCompleted(); } catch (Exception e) { o.onError(e); } }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
54. return Observable.<Order> create(o -> { try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); while (rs.next() && !o.isUnsubscribed()) { o.onNext(new Order(rs.getInt(1))); } rs.close(); o.onCompleted(); } catch (Exception e) { o.onError(e); } }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
55. return Observable.<Order> create(o -> { try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); while (rs.next() && !o.isUnsubscribed()) { o.onNext(new Order(rs.getInt(1))); } rs.close(); o.onCompleted(); } catch (Exception e) { o.onError(e); } }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
56. return Observable.<Order> create(o -> { try (Connection c = Database.getConnection()) { ResultSet rs = c.createStatement() .executeQuery("select * from orders"); while (rs.next() && !o.isUnsubscribed()) { o.onNext(new Order(rs.getInt(1))); } rs.close(); o.onCompleted(); } catch (Exception e) { o.onError(e); } }).subscribeOn(Schedulers.io()); Observable<Order> orders = getOrdersAsync(1);
57. or with “reactive pull” backpressure …
58. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
59. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
60. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
61. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
62. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
63. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
64. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
65. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
66. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
67. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
68. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
69. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
70. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
71. return Observable.create(AbstractOnSubscribe.<Order, ResultSet> create(s -> { ResultSet rs = s.state(); try { if (rs.next()) { s.onNext(new Order(rs.getInt(1))); } else { s.onCompleted(); } } catch (SQLException e) { s.onError(e); } }, s -> { Connection c = Database.getConnection(); try { Statement stmt = c.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(Integer.MIN_VALUE); return stmt.executeQuery("select * from orders"); } catch (SQLException e) { s.onError(e); return null; } }, rs -> rs.close() } // ignoring try/catch for slide brevity )).subscribeOn(Schedulers.io());
72. Observable.defer Observable.from/just Observable.subscribeOn Observable.create
73. Async Facade Blocking Code Servlet Sync/Async Bridge
74. protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { // perform service composition asynchronously getShippingStatusForCustomerOrdersAsync(req.getParameter(“customerId")) .toBlocking() // block on servlet thread .forEach(status -> { try { // write output using blocking IO resp.getWriter().write(status.toString()); } catch (Exception e) { throw new RuntimeException("unable to write", e); } }); } catch (Exception e) { // toBlocking and forEach will convert async onError events into thrown exceptions resp.setStatus(500); // write and log error ... } }
75. protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { // perform service composition asynchronously getShippingStatusForCustomerOrdersAsync(req.getParameter("customerId")) .toBlocking() // block on servlet thread .forEach(status -> { try { // write output using blocking IO resp.getWriter().write(status.toString()); } catch (Exception e) { throw new RuntimeException("unable to write", e); } }); } catch (Exception e) { // toBlocking and forEach will convert async onError events into thrown exceptions resp.setStatus(500); // write and log error ... } }
76. protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { // perform service composition asynchronously getShippingStatusForCustomerOrdersAsync(req.getParameter("customerId")) .toBlocking() // block on servlet thread .forEach(status -> { try { // write output using blocking IO resp.getWriter().write(status.toString()); } catch (Exception e) { throw new RuntimeException("unable to write", e); } }); } catch (Exception e) { // toBlocking and forEach will convert async onError events into thrown exceptions resp.setStatus(500); // write and log error ... } }
77. Observable<ShippingStatus> getShippingStatusForCustomerOrdersAsync(String customerId) { return getOrdersAsync(Integer.parseInt(customerId)) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }); }
78. Observable<ShippingStatus> getShippingStatusForCustomerOrdersAsync(String customerId) { return getOrdersAsync(Integer.parseInt(customerId)) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }); }
79. Observable<ShippingStatus> getShippingStatusForCustomerOrdersAsync(String customerId) { return getOrdersAsync(Integer.parseInt(customerId)) .limit(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }); }
80. protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { // perform service composition asynchronously getShippingStatusForCustomerOrdersAsync(req.getParameter("customerId")) .toBlocking() // block on servlet thread .forEach(status -> { try { // write output using blocking IO resp.getWriter().write(status.toString()); } catch (Exception e) { throw new RuntimeException("unable to write", e); } }); } catch (Exception e) { // toBlocking and forEach will convert async onError events into thrown exceptions resp.setStatus(500); // write and log error ... } }
81. protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { // perform service composition asynchronously getShippingStatusForCustomerOrdersAsync(req.getParameter("customerId")) .toBlocking() // block on servlet thread .forEach(status -> { try { // write output using blocking IO resp.getWriter().write(status.toString()); } catch (Exception e) { throw new RuntimeException("unable to write", e); } }); } catch (Exception e) { // toBlocking and forEach will convert async onError events into thrown exceptions resp.setStatus(500); // write and log error ... } }
82. protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { // perform service composition asynchronously getShippingStatusForCustomerOrdersAsync(req.getParameter("customerId")) .toBlocking() // block on servlet thread .forEach(status -> { try { // write output using blocking IO resp.getWriter().write(status.toString()); } catch (Exception e) { throw new RuntimeException("unable to write", e); } }); } catch (Exception e) { // toBlocking and forEach will convert async onError events into thrown exceptions resp.setStatus(500); // write and log error ... } }
83. Observable.toBlocking
84. Observable.toBlocking for bridging between async and sync
85. So how did it go adding reactive to an existing codebase?
86. First attempt took 3 tries
87. What the *&!@#$^!!!
88. Tech Worked …
89. … but …
90. Needed to relearn idiomatic solutions
91. Iterable<String> is = Arrays.asList("a", "b", "c"); ArrayList<String> ts = new ArrayList<>(); for(String s : is) { ts.add(s + "-modified"); } // do stuff with ts ... Observable<String> os = Observable.just("a", "b", "c"); os.map(s -> s + "-modified"); // do stuff with os ...
92. try { for (String s : is) { ts.add(s + "-modified"); } } catch (Exception e) { // do something with exception } os.map(s -> s + "-modified") .onErrorResumeNext(exception -> { // do something with exception return Observable.empty(); });
93. Needed to invest in documentation
96. Needed to allow for time and mistakes
97. Also … Unit Testing & Debugging
98. Async Unit Tests
99. TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS) .take(5) .map(i -> { return i + " value"; }).subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value", "2 value", "3 value", "4 value"));
100. TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS) .take(5) .map(i -> { return i + " value"; }).subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value", "2 value", "3 value", "4 value"));
101. TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS) .take(5) .map(i -> { return i + " value"; }).subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value", "2 value", "3 value", "4 value"));
102. TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS) .take(5) .map(i -> { return i + " value"; }).subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value", "2 value", "3 value", "4 value"));
103. TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS) .take(5) .map(i -> { return i + " value"; }).subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value", "2 value", "3 value", "4 value"));
104. TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS) .take(5) .map(i -> { return i + " value"; }).subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value", "2 value", "3 value", "4 value"));
105. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
106. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
107. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
108. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
109. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
110. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
111. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
112. TestScheduler test = Schedulers.test(); TestSubscriber<String> ts = new TestSubscriber<>(); Observable.interval(200, TimeUnit.MILLISECONDS, test) .map(i -> { return i + " value"; }).subscribe(ts); test.advanceTimeBy(200, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value")); test.advanceTimeTo(1000, TimeUnit.MILLISECONDS); ts.assertReceivedOnNext(Arrays.asList("0 value", "1 value”, "2 value", "3 value", "4 value"));
113. Async Debugging is Hard
114. java.lang.RuntimeException: failure! at jdbc.Test.lambda$19(Test.java:63) at jdbc.Test$$Lambda$8/888452645.call(Unknown Source) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55) at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76) at rx.observables.AbstractOnSubscribe$SubscriptionState.accept(AbstractOnSubscribe.java:533) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.doNext(AbstractOnSubscribe.java:367) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.request(AbstractOnSubscribe.java:345) at rx.internal.operators.OperatorSubscribeOn$1$1$1$1.request(OperatorSubscribeOn.java:88) at rx.Subscriber.setProducer(Subscriber.java:177) at rx.Subscriber.setProducer(Subscriber.java:171) at rx.internal.operators.OperatorSubscribeOn$1$1$1.setProducer(OperatorSubscribeOn.java:81) at rx.observables.AbstractOnSubscribe.call(AbstractOnSubscribe.java:191) at rx.observables.AbstractOnSubscribe$LambdaOnSubscribe.call(AbstractOnSubscribe.java:274) at rx.Observable.unsafeSubscribe(Observable.java:7495) at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: jdbc.Product.class at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:58) ... 20 more Ugly Stacktraces
115. java.lang.RuntimeException: failure! at jdbc.Test.lambda$19(Test.java:63) at jdbc.Test$$Lambda$8/888452645.call(Unknown Source) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55) at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76) at rx.observables.AbstractOnSubscribe$SubscriptionState.accept(AbstractOnSubscribe.java:533) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.doNext(AbstractOnSubscribe.java:367) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.request(AbstractOnSubscribe.java:345) at rx.internal.operators.OperatorSubscribeOn$1$1$1$1.request(OperatorSubscribeOn.java:88) at rx.Subscriber.setProducer(Subscriber.java:177) at rx.Subscriber.setProducer(Subscriber.java:171) at rx.internal.operators.OperatorSubscribeOn$1$1$1.setProducer(OperatorSubscribeOn.java:81) at rx.observables.AbstractOnSubscribe.call(AbstractOnSubscribe.java:191) at rx.observables.AbstractOnSubscribe$LambdaOnSubscribe.call(AbstractOnSubscribe.java:274) at rx.Observable.unsafeSubscribe(Observable.java:7495) at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: jdbc.Product.class at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:58) ... 20 more Ugly Stacktraces
116. java.lang.RuntimeException: failure! at jdbc.Test.lambda$19(Test.java:63) at jdbc.Test$$Lambda$8/888452645.call(Unknown Source) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55) at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76) at rx.observables.AbstractOnSubscribe$SubscriptionState.accept(AbstractOnSubscribe.java:533) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.doNext(AbstractOnSubscribe.java:367) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.request(AbstractOnSubscribe.java:345) at rx.internal.operators.OperatorSubscribeOn$1$1$1$1.request(OperatorSubscribeOn.java:88) at rx.Subscriber.setProducer(Subscriber.java:177) at rx.Subscriber.setProducer(Subscriber.java:171) at rx.internal.operators.OperatorSubscribeOn$1$1$1.setProducer(OperatorSubscribeOn.java:81) at rx.observables.AbstractOnSubscribe.call(AbstractOnSubscribe.java:191) at rx.observables.AbstractOnSubscribe$LambdaOnSubscribe.call(AbstractOnSubscribe.java:274) at rx.Observable.unsafeSubscribe(Observable.java:7495) at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: jdbc.Product.class at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:58) ... 20 more Missing Async Call Chain
117. java.lang.RuntimeException: failure! at jdbc.Test.lambda$19(Test.java:63) at jdbc.Test$$Lambda$8/888452645.call(Unknown Source) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55) at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76) at rx.observables.AbstractOnSubscribe$SubscriptionState.accept(AbstractOnSubscribe.java:533) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.doNext(AbstractOnSubscribe.java:367) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.request(AbstractOnSubscribe.java:345) at rx.internal.operators.OperatorSubscribeOn$1$1$1$1.request(OperatorSubscribeOn.java:88) at rx.Subscriber.setProducer(Subscriber.java:177) at rx.Subscriber.setProducer(Subscriber.java:171) at rx.internal.operators.OperatorSubscribeOn$1$1$1.setProducer(OperatorSubscribeOn.java:81) at rx.observables.AbstractOnSubscribe.call(AbstractOnSubscribe.java:191) at rx.observables.AbstractOnSubscribe$LambdaOnSubscribe.call(AbstractOnSubscribe.java:274) at rx.Observable.unsafeSubscribe(Observable.java:7495) at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: jdbc.Product.class at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:58) ... 20 more Stack Pollution
118. java.lang.RuntimeException: failure! at jdbc.Test.lambda$19(Test.java:63) at jdbc.Test$$Lambda$8/888452645.call(Unknown Source) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55) at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76) at rx.observables.AbstractOnSubscribe$SubscriptionState.accept(AbstractOnSubscribe.java:533) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.doNext(AbstractOnSubscribe.java:367) at rx.observables.AbstractOnSubscribe$SubscriptionProducer.request(AbstractOnSubscribe.java:345) at rx.internal.operators.OperatorSubscribeOn$1$1$1$1.request(OperatorSubscribeOn.java:88) at rx.Subscriber.setProducer(Subscriber.java:177) at rx.Subscriber.setProducer(Subscriber.java:171) at rx.internal.operators.OperatorSubscribeOn$1$1$1.setProducer(OperatorSubscribeOn.java:81) at rx.observables.AbstractOnSubscribe.call(AbstractOnSubscribe.java:191) at rx.observables.AbstractOnSubscribe$LambdaOnSubscribe.call(AbstractOnSubscribe.java:274) at rx.Observable.unsafeSubscribe(Observable.java:7495) at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: jdbc.Product.class at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:58) ... 20 more RxJava Tries
119. Stepping Severely Limited
120. Stepping Severely Limited
121. Stepping Severely Limited
122. orders.flatMap(o -> { return getProductsAsync(o) .doOnNext(t -> System.out.println("product: " + t)) .flatMap(p -> { return getShippingStatusAsync(o, p) .map(s -> { return s; }); }); }) Debugging via Logging
123. orders.flatMap(o -> { return getProductsAsync(o) .doOnNext(t -> System.out.println("product: " + t)) .flatMap(p -> { return getShippingStatusAsync(o, p) .map(s -> { return s; }); }); }) :-( Debugging via Logging
124. Async Debugging is Hard … a work in progress …
125. How about once all done?
126. Async Facade Blocking Code Servlet Sync/Async Bridge Observable APIs Worked Well
128. class  VideoService  {        def  VideoList  getPersonalizedListOfMovies(userId);        def  VideoBookmark  getBookmark(userId,  videoId);        def  VideoRating  getRating(userId,  videoId);        def  VideoMetadata  getMetadata(videoId);   } class  VideoService  {        def  Observable<VideoList>  getPersonalizedListOfMovies(userId);        def  Observable<VideoBookmark>  getBookmark(userId,  videoId);        def  Observable<VideoRating>  getRating(userId,  videoId);        def  Observable<VideoMetadata>  getMetadata(videoId);   } ... with an observable api: replaced the blocking api ...
129. Async Facade Blocking Code Servlet Sync/Async Bridge Challenges Mixing Approaches
130. Async Facade Blocking Code Servlet Sync/Async Bridge too many threads
131. Async Facade Blocking Code Servlet Sync/Async Bridge hard to tune & shed load
132. Async Facade Blocking Code Servlet Sync/Async Bridge easy to block async code
133. getOrdersAsync(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }); getOrdersAsync(1) .flatMap(o -> { return getProductsAsync(o) .map(p -> { return getShippingStatus(o, p); }); }); concurrent → sequential →
134. getOrdersAsync(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }); getOrdersAsync(1) .flatMap(o -> { return getProductsAsync(o) .map(p -> { return getShippingStatus(o, p); }); }); concurrent → sequential →
135. getOrdersAsync(1) .flatMap(o -> { return getProductsAsync(o) .flatMap(p -> { return getShippingStatusAsync(o, p); }); }); getOrdersAsync(1) .flatMap(o -> { return getProductsAsync(o) .map(p -> { return getShippingStatus(o, p); }); }); Async Facade only has Observable APIs
136. So what about going “fully reactive”?
138. “cause it’s better”
139. “cause it’s better” :-/
140. “it’s worse”
141. “it doesn’t matter”
142. green threads, native threads, fibers, actors, event loops, theoretically equivalent, OS kernels, etc, etc, etc …
143. C10k and beyond
144. What to bet future on?
145. What to bet future on?
146. WSPerfLab: https://github.com/Netflix-Skunkworks/WSPerfLab
147. WSPerfLab: https://github.com/Netflix-Skunkworks/WSPerfLab
148. WSPerfLab: https://github.com/Netflix-Skunkworks/WSPerfLab
149. WSPerfLab: https://github.com/Netflix-Skunkworks/WSPerfLab
150. WSPerfLab: https://github.com/Netflix-Skunkworks/WSPerfLab
151. WSPerfLab: https://github.com/Netflix-Skunkworks/WSPerfLab 154ms
152. Intel(R) Core(TM) i5-2400 CPU @ 3.10GHz: 4 cores, 1 thread per core OpenJDK 8 with frame pointer patch RxNetty 0.4.8 with Netty 4.0.25.Final vs Tomcat 7.0.45
153. 0% 25% 50% 75% 100% 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat CPU Usage in Percent concurrent clients Testing drove both tests near 100% cpu usage
154. 0ms 0.225ms 0.45ms 0.675ms 0.9ms 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat CPU Consumption per Request concurrent clients • Netty has lower CPU consumption per request • Netty keeps getting faster under load, Tomcat gets slower
155. 0rps 1250rps 2500rps 3750rps 5000rps 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Throughput concurrent clients • Netty achieves higher throughput • Mostly due to lower CPU consumption per request
156. 0ms 70ms 140ms 210ms 280ms 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Average Latency concurrent clients Theoretical best possible for test case is 154ms
157. 0ms 750ms 1500ms 2250ms 3000ms 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Max Latency concurrent clients Tomcat latency degradation is far more severe than Netty
158. 0ipc 0.175ipc 0.35ipc 0.525ipc 0.7ipc 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat CPU Instructions per Cycle concurrent clients Netty increases instructions per cycle as load increases
159. 0 400000 800000 1200000 1600000 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Thread Migrations concurrent clients significant divergence
160. 0 400000 800000 1200000 1600000 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Thread Migrations concurrent clients Under light load both are similar
161. 0 400000 800000 1200000 1600000 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Thread Migrations concurrent clients Increased load causes less migrations, improving IPC and CPU usage per request with event loop architecture
162. Why? • More efficient framework code • object allocation rates and GC costs • Event loop architecture • reduced thread migrations • cache warmth → memory locality → instructions per cycle → lower cpu consumption per request • Thread pool architecture • lock contention • thread migrations
163. Why? • More efficient framework code • object allocation rates and GC costs • Event loop architecture • reduced thread migrations • cache warmth → memory locality → instructions per cycle → lower cpu consumption per request • Thread pool architecture • lock contention • thread migrations
164. Why? • More efficient framework code • object allocation rates and GC costs • Event loop architecture • reduced thread migrations • cache warmth → memory locality → instructions per cycle → lower cpu consumption per request • Thread pool architecture • lock contention • thread migrations
165. Why? • More efficient framework code • object allocation rates and GC costs • Event loop architecture • reduced thread migrations • cache warmth → memory locality → instructions per cycle → lower cpu consumption per request • Thread pool architecture • lock contention • thread migrations • Many more details …
166. With current JVM and Linux, event loops have efficiency benefits.
167. Simpler But Not Easier
168. 0ms 750ms 1500ms 2250ms 3000ms 50 100 150 200 250 300 350 400 450 500 550 600 650 700 750 800 850 900 950 100010501100 RxNetty Tomcat Simpler But Not Easier
169. A fully async architecture does have benefits.
170. Reactive programming can be incrementally applied.
171. Mixed Codebases
172. Concurrency via Async Service Composition
173. Leverage RxJava Schedulers and Threads
174. Leverage RxJava Schedulers and Threads (or use Hystrix)
175. Concurrency is still non-trivial
176. Rx doesn’t trivialize it Concurrency is still non-trivial
178. Reactive Programming in the Netflix API with RxJava http://techblog.netflix.com/2013/02/rxjava-netflix-api.html Optimizing the Netflix API http://techblog.netflix.com/2013/01/optimizing-netflix-api.html Reactive Extensions (Rx) http://www.reactivex.io Reactive Streams https://github.com/reactive-streams/reactive-streams Ben Christensen @benjchristensen RxJava https://github.com/ReactiveX/RxJava @RxJava RxJS http://reactive-extensions.github.io/RxJS/ @ReactiveX jobs.netflix.com Dragon Image: http://www.clipartpanda.com/clipart_images/you-can-use-this-clip-art-on-2194929 Greenfield Image: https://kvaes.wordpress.com/2013/06/05/lingo-explained-greenfield-vs-brownfield/ NY City Image: https://thesandincalifornia.files.wordpress.com/2012/05/newyorkcity-madness.jpg Futuristic City: http://i.imgur.com/rPTkr9D.jpg