By the end of this tutorial you\u2019ll have a simple Spring Boot based Greetings micro-service running that<\/p>\n
- \n
- takes a message from a REST api<\/li>\n
- writes it to a Kafka topic<\/li>\n
- reads it from the topic<\/li>\n
- outputs it to the console<\/li>\n<\/ol>\n
Let\u2019s get started!<\/p>\n
Btw, you\u2019ll find the source code here<\/a>.<\/p>\n
What is Spring Cloud Streaming?<\/h2>\n
Spring Cloud Stream is a framework built upon Spring Boot for building message-driven microservices.<\/p>\n
What is Kafka?<\/h3>\n
Kafka is a popular high performant and horizontally scalable messaging platform originally developed by LinkedIn.<\/p>\n
Installing Kafka<\/h3>\n
Download Kafka from\u00a0here<\/a>\u00a0and untar it:<\/p>\n
>\u00a0tar -xzf kafka_2.11-1.0.0.tgz\r\n> cd kafka_2.11-1.0.0<\/pre>\n
Start Zookeeper and Kafka<\/p>\n
On Windows:<\/p>\n
>\u00a0bin\\windows\\zookeeper-server-start.bat config\\zookeeper.properties\r\n> bin\\windows\\kafka-server-start.bat config\\server.properties<\/pre>\n
On Linux or Mac:<\/p>\n
>\u00a0bin\/zookeeper-server-start.sh config\/zookeeper.properties\r\n> bin\/kafka-server-start.sh config\/server.properties<\/pre>\n
If Kafka is not running and fails to start after your computer wakes up from hibernation, delete the\u00a0
<TMP_DIR>\/kafka-logs<\/code>folder and then start Kafka again.<\/p>\nWhat is Lombok?<\/h2>\n
Lombok is a java framework that automatically generates getters, setters, toString(), builders, loggers, etc. in the code.<\/p>\n
Maven dependencies<\/h2>\n
Go to\u00a0https:\/\/start.spring.io<\/a>\u00a0to create a maven project:<\/p>\n
![\"\"](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2018\/03\/spring-initializr-streamkafka-1024x519.png\")
<\/a><\/p>\n- \n
- Add necessary dependencies:\u00a0
Spring Cloud Stream<\/code>,\u00a0Kafka<\/code>,\u00a0Devtools<\/code>\u00a0(for hot redeploys during development, optional),\u00a0Actuator<\/code>\u00a0(for monitoring application, optional),\u00a0Lombok<\/code>\u00a0(make sure to also have the Lombok plugin installed in your IDE)<\/li>\n- Click the Generate Project button to download the project as a zip file<\/li>\n
- Extract zip file and import the maven project to your favourite IDE<\/li>\n<\/ol>\n
Notice the maven dependencies in the\u00a0
pom.xml<\/code>\u00a0file:<\/p>\n<dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-actuator<\/artifactId>\r\n <\/dependency>\r\n <dependency>\r\n <groupId>org.springframework.cloud<\/groupId>\r\n <artifactId>spring-cloud-stream<\/artifactId>\r\n <\/dependency>\r\n <dependency>\r\n <groupId>org.springframework.cloud<\/groupId>\r\n <artifactId>spring-cloud-starter-stream-kafka<\/artifactId>\r\n <\/dependency>\r\n <!-- Also install the Lombok plugin in your IDE -->\r\n <dependency>\r\n <groupId>org.projectlombok<\/groupId>\r\n <artifactId>lombok<\/artifactId>\r\n <optional>true<\/optional>\r\n <\/dependency>\r\n\r\n <!-- hot reload - press Ctrl+F9 in IntelliJ after a code change while application is running -->\r\n <dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-devtools<\/artifactId>\r\n <optional>true<\/optional>\r\n <\/dependency><\/pre>\n
\u2026 also the\u00a0
<dependencyManagement><\/code>\u00a0section:<\/p>\n<dependencyManagement>\r\n\u00a0 <dependencies>\r\n\u00a0 \u00a0 <dependency>\r\n\u00a0 \u00a0 \u00a0 <!-- Import dependency management from Spring Boot -->\r\n\u00a0 \u00a0 \u00a0 <groupId>org.springframework.boot<\/groupId>\r\n\u00a0 \u00a0 \u00a0 <artifactId>spring-boot-dependencies<\/artifactId>\r\n\u00a0 \u00a0 \u00a0 <version>${spring-boot.version}<\/version>\r\n\u00a0 \u00a0 \u00a0 <type>pom<\/type>\r\n\u00a0 \u00a0 \u00a0 <scope>import<\/scope>\r\n\u00a0 \u00a0 <\/dependency>\r\n\u00a0 \u00a0 <dependency>\r\n\u00a0 \u00a0 \u00a0 <groupId>org.springframework.cloud<\/groupId>\r\n\u00a0 \u00a0 \u00a0 <artifactId>spring-cloud-stream-dependencies<\/artifactId>\r\n\u00a0 \u00a0 \u00a0 <version>${spring-cloud-stream.version}<\/version>\r\n\u00a0 \u00a0 \u00a0 <type>pom<\/type>\r\n\u00a0 \u00a0 \u00a0 <scope>import<\/scope>\r\n\u00a0 \u00a0 <\/dependency>\r\n\u00a0 <\/dependencies>\r\n <\/dependencyManagement><\/pre>\n\u2026 and the\u00a0
<repository><\/code>\u00a0section:<\/div> <\/div><\/p>\n<repository>\r\n\u00a0 <id>spring-milestones<\/id>\r\n\u00a0 <name>Spring Milestones<\/name>\r\n\u00a0 <url>http:\/\/repo.spring.io\/libs-milestone<\/url>\r\n\u00a0 <snapshots>\r\n\u00a0 \u00a0 <enabled>false<\/enabled>\r\n\u00a0 <\/snapshots>\r\n<\/repository><\/pre>\n
Define the Kafka streams<\/h2>\n
package com.kaviddiss.streamkafka.stream;\r\n\r\nimport org.springframework.cloud.stream.annotation.Input;\r\nimport org.springframework.cloud.stream.annotation.Output;\r\nimport org.springframework.messaging.MessageChannel;\r\nimport org.springframework.messaging.SubscribableChannel;\u00a0 \r\n\r\npublic interface GreetingsStreams {\r\n String INPUT = \"greetings-in\";\r\n String OUTPUT = \"greetings-out\";\r\n\r\n @Input(INPUT)\r\n SubscribableChannel inboundGreetings();\r\n\r\n @Output(OUTPUT)\r\n MessageChannel outboundGreetings();\r\n}<\/pre>\nIn order for our application to be able to communicate with Kafka, we\u2019ll need to define an outbound stream to write messages to a Kafka topic, and an inbound stream to read messages from a Kafka topic.<\/p>\n
Spring Cloud provides a convenient way to do this by simply creating an interface that defines a separate method for each stream.<\/p>\n
The\u00a0
inboundGreetings()<\/code>\u00a0method defines the inbound stream to read from Kafka and\u00a0outboundGreetings()<\/code>\u00a0method defines the outbound stream to write to Kafka.<\/p>\nDuring runtime Spring will create a java proxy based implementation of the\u00a0
GreetingsStreams<\/code>\u00a0interface that can be injected as a Spring Bean anywhere in the code to access our two streams.<\/p>\nConfigure Spring Cloud Stream<\/h2>\n
Our next step is to configure Spring Cloud Stream to bind to our streams in the\u00a0
GreetingsStreams<\/code>\u00a0interface. This can be done by creating a\u00a0@Configuration<\/code>\u00a0class\u00a0com.kaviddiss.streamkafka.config.StreamsConfig<\/code>\u00a0with below code:<\/p>\npackage com.kaviddiss.streamkafka.config;\r\n\r\nimport com.kaviddiss.streamkafka.stream.GreetingsStreams;\r\nimport org.springframework.cloud.stream.annotation.EnableBinding;\r\n\r\n@EnableBinding(GreetingsStreams.class)\r\npublic class StreamsConfig {\r\n}<\/pre>\nBinding the streams is done using the\u00a0
@EnableBinding<\/code>\u00a0annotation where the\u00a0GreatingsService<\/code>\u00a0interface is passed to.<\/p>\nConfiguration properties for Kafka<\/h2>\n
By default, the configuration properties are stored in the\u00a0
src\/main\/resources\/application.properties<\/code>\u00a0file.<\/p>\nHowever I prefer to use the YAML format as it\u2019s less verbose and allows to keep both common and environment-specific properties in the same file.<\/p>\n
For now, let\u2019s rename\u00a0
application.properties<\/code>\u00a0to\u00a0application.yaml<\/code>\u00a0and paste below config snippet into the file:<\/p>\nspring:\r\n cloud:\r\n stream:\r\n kafka:\r\n binder:\r\n brokers: localhost:9092\r\n bindings:\r\n greetings-in:\r\n destination: greetings\r\n contentType: application\/json\r\n greetings-out:\r\n destination: greetings\r\n contentType: application\/json<\/pre>\n
The above configuration properties configure the address of the Kafka server to connect to, and the Kafka topic we use for both the inbound and outbound streams in our code. They both must use the same Kafka topic!<\/p>\n
The\u00a0
contentType<\/code>\u00a0properties tell Spring Cloud Stream to send\/receive our message objects as\u00a0String<\/code>s in the streams.<\/p>\nCreate the message object<\/h2>\n
Create a simple\u00a0
com.kaviddiss.streamkafka.model.Greetings<\/code>\u00a0class with below code that will represent the message object we read from and write to the\u00a0greetings<\/code>\u00a0Kafka topic:<\/p>\npackage com.kaviddiss.streamkafka.model;\r\n\r\n\/\/ lombok autogenerates getters, setters, toString() and a builder (see https:\/\/projectlombok.org\/):\r\nimport lombok.Builder;\r\nimport lombok.Getter;\r\nimport lombok.Setter;\r\nimport lombok.ToString;\r\n\r\n@Getter @Setter @ToString @Builder\r\npublic class Greetings {\r\n private long timestamp;\r\n private String message;\r\n}<\/pre>\nNotice how the class doesn\u2019t have any getters and setters thanks to the Lombok annotations. The\u00a0
@ToString<\/code>\u00a0will generate a\u00a0toString()<\/code>\u00a0method using the class\u2019 fields and the\u00a0@Builder<\/code>\u00a0annotation will allow us creating\u00a0Greetings<\/code>\u00a0objects using fluent builder (see below).<\/p>\nCreate service layer to write to Kafka<\/h2>\n
Let\u2019s create the\u00a0
com.kaviddiss.streamkafka.service.GreetingsService<\/code>\u00a0class with below code that will write a\u00a0Greetings<\/code>object to the\u00a0greetings<\/code>\u00a0Kafka topic:<\/p>\npackage com.kaviddiss.streamkafka.service;\r\n\r\nimport com.kaviddiss.streamkafka.model.Greetings;\r\nimport com.kaviddiss.streamkafka.stream.GreetingsStreams;\r\nimport lombok.extern.slf4j.Slf4j;\r\nimport org.springframework.messaging.MessageChannel;\r\nimport org.springframework.messaging.MessageHeaders;\r\nimport org.springframework.messaging.support.MessageBuilder;\r\nimport org.springframework.stereotype.Service;\r\nimport org.springframework.util.MimeTypeUtils;\r\n\r\n@Service\r\n@Slf4j\r\npublic class GreetingsService {\r\n private final GreetingsStreams greetingsStreams;\r\n\r\n public GreetingsService(GreetingsStreams greetingsStreams) {\r\n this.greetingsStreams = greetingsStreams;\r\n }\r\n\r\n public void sendGreeting(final Greetings greetings) {\r\n log.info(\"Sending greetings {}\", greetings);\r\n\r\n MessageChannel messageChannel = greetingsStreams.outboundGreetings();\r\n messageChannel.send(MessageBuilder\r\n .withPayload(greetings)\r\n .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)\r\n .build());\r\n }<\/pre>\nThe\u00a0
@Service<\/code>\u00a0annotation will configure this class as a Spring Bean and inject the\u00a0GreetingsService<\/code>\u00a0dependency via the constructor.<\/p>\nThe\u00a0
@Slf4j<\/code>\u00a0annotation will generate an SLF4J logger field that we can use for logging.<\/p>\nIn the\u00a0
sendGreeting()<\/code>\u00a0method we use the injected\u00a0GreetingsStream<\/code>\u00a0object to send a message represented by the\u00a0Greetings<\/code>\u00a0object.<\/p>\nCreate REST api<\/h2>\n
Now we\u2019ll be creating a REST api endpoint that will trigger sending a message to Kafka using the\u00a0
GreetingsService<\/code>\u00a0Spring Bean:<\/p>\npackage com.kaviddiss.streamkafka.web;\r\n\r\nimport com.kaviddiss.streamkafka.model.Greetings;\r\nimport com.kaviddiss.streamkafka.service.GreetingsService;\r\nimport org.springframework.http.HttpStatus;\r\nimport org.springframework.web.bind.annotation.GetMapping;\r\nimport org.springframework.web.bind.annotation.RequestParam;\r\nimport org.springframework.web.bind.annotation.ResponseStatus;\r\nimport org.springframework.web.bind.annotation.RestController;\u00a0\r\n\r\n@RestController\r\npublic class GreetingsController {\r\n private final GreetingsService greetingsService;\r\n\r\n public GreetingsController(GreetingsService greetingsService) {\r\n this.greetingsService = greetingsService;\r\n }\r\n\r\n @GetMapping(\"\/greetings\")\r\n @ResponseStatus(HttpStatus.ACCEPTED)\r\n public void greetings(@RequestParam(\"message\") String message) {\r\n Greetings greetings = Greetings.builder()\r\n .message(message)\r\n .timestamp(System.currentTimeMillis())\r\n .build();\r\n\r\n greetingsService.sendGreeting(greetings);\r\n }\r\n}<\/pre>\nThe\u00a0
@RestController<\/code>\u00a0annotation tells Spring that this is a Controller bean (the C from MVC). The\u00a0greetings()<\/code>\u00a0method defines an\u00a0HTTP GET \/greetings<\/code>\u00a0endpoint that takes a\u00a0message<\/code>\u00a0request param and passes it to the\u00a0sendGreeting()<\/code>\u00a0method in\u00a0GreetingsService<\/code>.<\/p>\nListening on the greetings Kafka topic<\/h2>\n
Let\u2019s create a\u00a0
com.kaviddiss.streamkafka.service.GreetingsListener<\/code>\u00a0class that will listen to messages on the\u00a0greetings<\/code>Kafka topic and log them on the console:<\/p>\npackage com.kaviddiss.streamkafka.service;\r\n\r\nimport com.kaviddiss.streamkafka.model.Greetings;\r\nimport com.kaviddiss.streamkafka.stream.GreetingsStreams;\r\nimport lombok.extern.slf4j.Slf4j;\r\nimport org.springframework.cloud.stream.annotation.StreamListener;\r\nimport org.springframework.messaging.handler.annotation.Payload;\r\nimport org.springframework.stereotype.Component;\r\n\r\n@Component\r\n@Slf4j\r\npublic class GreetingsListener {\r\n @StreamListener(GreetingsStreams.INPUT)\r\n public void handleGreetings(@Payload Greetings greetings) {\r\n log.info(\"Received greetings: {}\", greetings);\r\n }\r\n}<\/pre>\nThe\u00a0
@Component<\/code>\u00a0annotation similarly to\u00a0@Service<\/code>\u00a0and\u00a0@RestController<\/code>\u00a0defines a Spring Bean.<\/p>\nGreetingsListener<\/code>\u00a0has a single method,\u00a0handleGreetings()<\/code>\u00a0that will be invoked by Spring Cloud Stream with every new\u00a0Greetings<\/code>\u00a0message object on the\u00a0greetings<\/code>\u00a0Kafka topic. This is thanks to the\u00a0@StreamListener<\/code>\u00a0annotation configured for the\u00a0handleGreetings()<\/code>\u00a0method.<\/p>\nRunning the application<\/h2>\n
The last piece of the puzzle is the\u00a0
com.kaviddiss.streamkafka.StreamKafkaApplication<\/code>\u00a0class that was auto-generated by the Spring Initializer:<\/p>\npackage com.kaviddiss.streamkafka;\r\n\r\nimport org.springframework.boot.SpringApplication;\r\nimport org.springframework.boot.autoconfigure.SpringBootApplication;\r\n\r\n@SpringBootApplication\r\npublic class StreamKafkaApplication {\r\n\r\n public static void main(String[] args) {\r\n SpringApplication.run(StreamKafkaApplication.class, args);\r\n }\r\n}<\/pre>\nNo need to make any changes here. You can either run this class as a Java application from your IDE, or run the application from the command line using the Spring Boot maven plugin:<\/p>\n
>\u00a0mvn spring-boot:run<\/pre>\n
Once the application is running, go to\u00a0http:\/\/localhost:8080\/greetings?message=hello<\/a>\u00a0in the browser and check your console.<\/p>\n
Summary<\/h2>\n
I hope you enjoyed this tutorial. Feel free to ask any questions and leave your feedback.<\/p>\n
\n
- Add necessary dependencies:\u00a0