Getting Started with Spring Cloud Stream

This post was co-written with Ben Wilcock, Product and Technical Marketing Manager for Spring at Pivotal. 🔔 A file has been uploaded! 🔔 🔔 A new user was registered! 🔔 🔔 An order was placed! 🔔 These sound like events that many parts of our application architecture might be interested in, right? For example, when an order is placed on our website, we’ll need a call to process the payment, a call to reserve inventory, and a call to begin the process of picking, packaging and shipping the product. For a single order, this isn’t too bad. Our store can make a few requests to these backend services directly and it shouldn’t introduce too much overhead.But what happens if we’re really good at selling our product? Processing 100 orders a second suddenly means our frontend is making three hundred calls per second to our backend services. If we add one more service to that—say, to report to an internal sales dashboard— now that’s four hundred calls per second. That’s a lot of overhead! What if instead, we can simply have our website alert our whole architecture at once? It can yell, “Hey! I made a sale” to our whole stack, and any component that’s interested can take the appropriate action. This means we don’t need to update our frontend as we add additional services, and our new services just need to know what to listen for. Why Spring Cloud Stream? The above is an example of an , where instead of reaching out to each service one by one, our services instead emit a change of state. If a file is uploaded, our file service can emit it out to a messaging platform, and then our Super Duper Image Resizer 3000 service can listen for that and automatically generate differently sized profile images. Pivotal’s own Richard Seroter in detail, and it’s a great read. In his blog post, Richard talks about messaging as a way of reliably delivering events to many consumers quickly and in volume. event-driven architecture wrote about this very topic He also touches on something we want to talk about today: . Spring Cloud Stream We’re big fans of both Kafka and RabbitMQ as event streaming platforms, so for this demo we’ll use Kafka. No matter which you choose to use, making it easy to produce and consume events is important for your developers. I’ve used a lot of frameworks that abstract away from the underlying message queue, but none quite as easy and flexible as Spring Cloud Stream. My teammate Ben Wilcock put together a demo that really shows just how easy it is to get up and running. Let’s take it for a spin—and to follow along, you can . download the full source code here Prepping For The Demo We only need a couple of things for our demo, which are and Docker Compose, and of course your favorite distribution of the JDK (perhaps even , which we sponsor). To keep things easy, the demo includes a Docker Compose config that will set up both Kafka and RabbitMQ, though for our purposes we’ll only be using Kafka. We can spin this up with a simple command: Docker AdoptOpenJDK docker-compose up This will read our file, download the necessary container images, run them, and configure them. After just a few moments, Kafka should be up and running and ready to go. docker-compose.yml Sending Events Our demo is made up of two Spring microservices, one to produce events and one to consume them. In our fictional scenario, the message producer will create a stream of applications for bank loans, and our processor will check if those applications should be approved or declined. Let’s start by producing some messages that will be sent to Kafka, the code for which is in the directory. loansource There are a few files of code here. The file defines a object and the file defines all the states a loan can be in. What’s interesting, though, is the file, which is what’s actually producing our messages. Loan.java loan Statuses.java LoansourceApplication.java As you can imagine, Spring and its dependencies handle a lot of the wiring up of components for us automatically. Let’s take a look at to see how this works. LoansourceApplication.java { () -> { String rName = names.get( Random().nextInt(names.size())); Long rAmount = amounts.get( Random().nextInt(amounts.size())); Loan loan = Loan(UUID.randomUUID().toString(), rName, rAmount); log.info( , loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName()); loan; }; } @Bean Supplier public supplyLoan () return new new new "{} {} for ${} for {}" return is a Java function data type. Because there is only one method that returns this type, Spring Cloud Stream knows exactly what to do next. By default, it will trigger this function once every second and send the result to the default named . What’s nice about this function method is that it only contains business logic, so you can test it using your favorite testing methods. Supplier<> @Bean MessageChannel output We could use the property in the application.properties file to explicitly declare which function bean we want to be bound to binding destinations, but for cases when you only have a single defined, this is not necessary. spring.cloud.function.definition @Bean Likewise, if we wanted to use a different poller interval, we can use the property in the file. The only question that remains is, “How does Spring know it’s Kafka we’re writing to?” For that, we take a look at our : spring.integration.poller.fixed-delay application.properties pom.xml org.springframework.cloud spring-cloud-stream-binder-kafka dependency groupId groupId artifactId artifactId dependency Providing this dependency in our code tells Spring, “I’d like to send these messages to Kafka”. Since our Kafka server is listening on on the default port, we don’t need to provide any additional configuration in our file, but if that’s not the case, providing information such as hostname, port, authentication, etc. localhost application.properties we can of course do so We can run our code and activate the profile, which we’ve configured to be the profile that includes the Kafka SCS binding, and we should see it start producing messages: kafka loansource ./mvnw package spring-boot:run -DskipTests= -Pkafka cd true After a few moments, we’ll see our application start creating new loans and sending them to Kafka: ..LoansourceApplication : PENDING eff9b58-e1f1 d f1d-aa4db8dbb75a $ Donald ..LoansourceApplication : PENDING d507c06c bb a98 f85 f74af36984 $ Jacinda ..LoansourceApplication : PENDING fc86a4-d461 c ce1a258e7 $ Jacinda ..LoansourceApplication : PENDING f3756c-ea9b f-bad2 f1647188b1 $ Vladimir ..LoansourceApplication : PENDING d10f-c1c8 fe8 ce363ef56f $ Theresa 2019 -10 -15. 9 -474 -8 for 10000000 for 2019 -10 -15. -81 -4 -8 -38 for 100 for 2019 -10 -15. 19 -470 -8005 -423 for 100 for 2019 -10 -15. 33 -472 -73 for 10000 for 2019 -10 -15. 1625 -4e75 -8 -10 for 10000000 for If you prefer, you can also see the messages in your browser using . Simply point your browser to and you should see a UI that allows you to look at the messages stored in Kafka: KafDrop localhost:9000 Receiving Events We’ve got half of the equation here, but we also need something to consume and process these events. For this, we’ll look in the directory. For this half of the demo, our loan checker will observe every application and approve or decline it. If approved, an approval message will be sent to the topic otherwise, a denial message will be sent to the topic. loancheck approved declined You can extrapolate from here that other systems down the line could listen for and pick up these messages for further processing. For example, maybe a payout system listens for an approved loan to start processing it. We’ll see the code here is a little different, just pointing to different topics. We see that in , we have the annotation, meaning that all of our definitions for channel bindings are found in the class. LoanCheckApplication.java @EnableBinding(LoanProcessor.class) LoanProcessor In our file, we’ll see we define the we’re listening on is named output, matching the default topic our producer writes to. Additionally, we define two other MessageChannels that we’ll be writing to, and . For each of these, we also define which method to invoke when a message is received on those channels. LoanProcessor.java MessageChannel approved declined { String APPLICATIONS_IN = ; String APPROVED_OUT = ; String DECLINED_OUT = ; (APPLICATIONS_IN) ; (APPROVED_OUT) ; (DECLINED_OUT) ; } @Component public interface LoanProcessor "output" "approved" "declined" @Input SubscribableChannel sourceOfLoanApplications () @Output MessageChannel approved () @Output MessageChannel declined () Finally, we can see how this ties into which method is invoked if we take a look at the file. We’ll see we have a method with the annotation that matches our Input we defined previously: LoanChecker.java checkAndSortLoans @StreamListener (LoanProcessor.APPLICATIONS_IN) { log.info( , loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName()); (loan.getAmount() > MAX_AMOUNT) { loan.setStatus(Statuses.DECLINED.name()); processor.declined().send(message(loan)); } { loan.setStatus(Statuses.APPROVED.name()); processor.approved().send(message(loan)); } } @StreamListener public void checkAndSortLoans (Loan loan) "{} {} for ${} for {}" if else We can start this code up much like we did our , by opening up a separate terminal and running the following: loansource loancheck ./mvnw package spring-boot:run -DskipTests= -Pkafka cd true After a few moments, we’ll start seeing our pending messages come through and then get sorted into or : approved declined ..LoanChecker : PENDING a887cf-ab5f c4-b03b cfc $ Kim ..LoanChecker : APPROVED a887cf-ab5f c4-b03b cfc $ Kim ..LoanChecker : PENDING a15f13fe-fc9a fb-b6f0 a18c0cd $ Angela ..LoanChecker : DECLINED a15f13fe-fc9a fb-b6f0 a18c0cd $ Angela 2019 -10 -15. 95 -48 -556675446 for 1000 for 2019 -10 -15. 95 -48 -556675446 for 1000 for 2019 -10 -15. -40 -24106 for 100000000 for 2019 -10 -15. -40 -24106 for 100000000 for Wrapping Up Spring Cloud Stream provides an extremely powerful abstraction for potentially complicated messaging platforms, turning the act of producing messages into just a couple lines of code. Should your infrastructure needs change and you need to migrate to a new messaging platform, not a single line of code changes other than your pom file. No matter if you’re using Kafka, RabbitMQ, or a cloud provider’s solution such as GCP Pub/Sub or Azure Event Hub, Spring Cloud Stream means it’s simple and quick to get up and running. About the Author Brian is a Principal Product Marketing Manager on the Technical Marketing team at Pivotal, with a focus on technical educational content for Pivotal customers as well as the Cloud Foundry, BOSH, and Knative communities. Prior to Pivotal, Brian worked on both the development and operations of software, with a heavy focus on Cloud Foundry and BOSH at companies in many industries including finance, entertainment and technology. He loves learning and experimenting in many fields of technology, and more importantly sharing the lessons learned along the way.