Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Reactive programming in Java

Reactive programming in Java

Reactive Programming is a new programming paradigm providing an higher level abstraction to build event-driven, scalable, resilient and responsive applications. Its main idea is extending the observer pattern to support sequences of data/events and adds operators allowing to compose those sequences together in a declarative and functional style. In this talk we'll go over reactive programming principles and then dive into implementation level details showing in a live coding session how to develop a reactive application using RxJava, a library for composing behaviors applied to event streams.

Mario Fusco

May 08, 2022
Tweet

More Decks by Mario Fusco

Other Decks in Programming

Transcript

  1. by Mario Fusco [email protected] @mariofusco Reactive Programming for a demanding

    world: building event-driven and responsive applications with RxJava
  2. Why reactive? What changed? ➢ Usage patterns: Users expect millisecond

    response times and 100% uptime A few years ago largest applications had tens of servers and gigabytes of data Seconds of response time and hours of offline maintenance were acceptable Today ➢ Big Data: usually measured in Petabytes and increasing with an extremely high frequency ➢ Heterogeneous environment: applications are deployed on everything from mobile devices to cloud-based clusters running thousands of multi-core processors Today's demands are simply not met by yesterday’s software architectures!
  3. The Reactive Manifesto The system responds in a timely manner

    if at all possible. Responsiveness is the cornerstone of usability The system stays responsive in the face of failure The system stays responsive under varying workload. It can react to changes in the input rate by increasing or decreasing the resources allocated to service these inputs The system rely on asynchronous message passing to establish a boundary between components that ensures loose coupling, isolation and location transparency Responsive Resilient Message Driven Elastic
  4. The Reactive Streams Initiative Reactive Streams is an initiative to

    provide a standard for asynchronous stream processing with non-blocking back pressure on the JVM Problem Handling streams of (live) data in an asynchronous and possibly non-blocking way Scope Finding a minimal API describing the operations available on Reactive Streams Implementors Akka Streams Reactor Composable RxJava Ratpack
  5. Rethinking programming the Reactive way ➢ Reactive programming is a

    programming paradigm about data-flow ➢ Think in terms of discrete events and streams of them ➢ React to events and define behaviors combining them ➢ The system state changes over time based on the flow of events ➢ Keep your data/events immutable Never block!
  6. Reactive programming is programming with asynchronous data streams ➢ A

    stream is a sequence of ongoing events ordered in time ➢ Events are processed asynchronously, by defining a function that will be executed when an event arrives
  7. Streams are not collections Streams are ➢ potentially unbounded in

    length ➢ focused on transformation of data ➢ time-dependent ➢ ephemeral ➢ traversable only once «You cannot step twice into the same stream. For as you are stepping in, other waters are ever flowing on to you.» — Heraclitus
  8. RxJava Reactive Extension for async programming ➢ A library for

    composing asynchronous and event-based programs using observable sequences for the Java VM ➢ Supports Java 6 or higher and JVM-based languages such as Groovy, Clojure, JRuby, Kotlin and Scala ➢ Includes a DSL providing extensive operations for streams transformation, filtering and recombination ➢ Implements pure “push” model ➢ Decouple events production from consumption ➢ Allows blocking only for back pressure ➢ First class support for error handling, scheduling & flow control ➢ Used by Netflix to make the entire service layer asynchronous http://reactivex.io http://github.com/ReactiveX/RxJava
  9. Marble diagrams: Representing events' streams ... A stream is a

    sequence of ongoing events ordered in time. It can emit three different things: 1. a value (of some type) 2. an error 3. "completed" signal
  10. Observable The Observable interface defines how to access asynchronous sequences

    of multiple items single value multiple values synchronous T getData() Iterable<T> getData() asynchronous Future<T> getData() Observable<T> getData() An Observable is the asynchronous/push “dual” to the synchronous/pull Iterable Iterable (pull) Obesrvable (push) retrieve data T next() onNext(T) signal error throws Exception onError(Exception) completion !hasNext() onCompleted()
  11. Observable as async Stream // Stream<Stock> containing 100 Stocks getDataFromLocalMemory()

    .skip(10) .filter(s -> s.getValue > 100) .map(s ­ -> s.getName() + “: ” + s.getValue()) .forEach(System.out::println); // Observable<Stock> emitting 100 Stocks getDataFromNetwork() .skip(10) .filter(s -> s.getValue > 100) .map(s ­ -> s.getName() + “: ” + s.getValue()) .forEach(System.out::println);
  12. Observable and Concurrency An Observable is sequential → No concurrent

    emissions Scheduling and combining Observables enables concurrency while retaining sequential emission
  13. Reactive Programming requires a mental shift from sync to async

    from pull to push from imperative to functional
  14. How is the Observable implemented? ➢ Maybe it executes its

    logic on subscriber thread? ➢ Maybe it delegates part of the work to other threads? ➢ Does it use NIO? ➢ Maybe it is an actor? ➢ Does it return cached data? Observer does not care! public interface Observer<T> { void onCompleted(); void onError(Throwable var1); void onNext(T var1); }
  15. Non-Opinionated Concurrency Observable Observer Calling Thread Callback Thread onNext Work

    synchronously on calling thread Observable Observer Calling Thread Callback Thread onNext Work asynchronously on separate thread Thread pool Observable Observer Calling Thread Callback Threads onNext Work asynchronously on multiple threads Thread pool Could be an actor or an event loop
  16. public class TempInfo { public static final Random random =

    new Random(); public final String town; public final int temp; public TempInfo(String town, int temp) { this.town = town; this.temp = temp; } public static TempInfo fetch(String temp) { return new TempInfo(temp, random.nextInt(70) - 20); } @Override public String toString() { return String.format(town + " : " + temp); } } Fetching town's temperature
  17. Creating Observables ... public static Observable<TempInfo> getTemp(String town) { return

    Observable.just(TempInfo.fetch(town)); } public static Observable<TempInfo> getTemps(String... towns) { return Observable.from(Stream.of(towns) .map(town -> TempInfo.fetch(town)) .collect(toList())); } public static Observable<TempInfo> getFeed(String town) { return Observable.create(subscriber -> { while (true) { subscriber.onNext(TempInfo.fetch(town)); Thread.sleep(1000); } }); } ➢ … with just a single value ➢ … from an Iterable ➢ … from another Observable
  18. Combining Observables public static Observable<TempInfo> getFeed(String town) { return Observable.create(

    subscriber -> Observable.interval(1, TimeUnit.SECONDS) .subscribe(i -> subscriber .onNext(TempInfo.fetch(town)))); } public static Observable<TempInfo> getFeeds(String... towns) { return Observable.merge(Arrays.stream(towns) .map(town -> getFeed(town)) .collect(toList())); } ➢ Subscribing one Observable to another ➢ Merging more Observables
  19. Managing errors and completion public static Observable<TempInfo> getFeed(String town) {

    return Observable.create(subscriber -> Observable.interval(1, TimeUnit.SECONDS) .subscribe(i -> { if (i > 5) subscriber.onCompleted(); try { subscriber.onNext(TempInfo.fetch(town)); } catch (Exception e) { subscriber.onError(e); } })); } Observable<TempInfo> feed = getFeeds("Milano", "Roma", "Napoli"); feed.subscribe(new Observer<TempInfo>() { public void onCompleted() { System.out.println("Done!"); } public void onError(Throwable t) { System.out.println("Got problem: " + t); } public void onNext(TempInfo t) { System.out.println(t); } });
  20. Hot & Cold Observables HOT emits immediately whether its Observer

    is ready or not examples mouse & keyboard events system events stock prices time COLD emits at controlled rate when requested by its Observers examples in-memory Iterable database query web service request reading file
  21. Dealing with a slow consumer Push (reactive) when consumer keeps

    up with producer Switch to Pull (interactive) when consumer is slow observable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1); } @Override public void onCompleted() { /* handle sequence-complete */ } @Override public void onError(Throwable e) { /* handle error */ } @Override public void onNext(T n) { // do something with the emitted item request(1); // request another item } }); When you subscribe to an Observable, you can request reactive pull backpressure
  22. Backpressur e Reactive pull backpressure isn’t magic Backpressure doesn’t make

    the problem of an overproducing Observable or an underconsuming Subscriber go away. It just moves the problem up the chain of operators to a point where it can be handled better.