HomeEducationConstructing a Reactive Occasion Streaming Structure - DZone | The Global Today

Constructing a Reactive Occasion Streaming Structure – DZone | The Global Today

This text outlines an answer for streaming occasions from Kafka, forwarding them to Redis utilizing its Stream API, and studying particular person streams from Redis by way of its streaming API. The added complexity on this situation is the necessity to stream occasions from an HTTP endpoint utilizing Server-Despatched Occasions (SSE) whereas making certain that solely occasions related to a selected shopper ID are processed and despatched.

Downside Assertion

Many firms have an present Kafka infrastructure the place occasions are being produced. Our purpose is to arrange a system that subscribes to Kafka messages however solely processes occasions related to a selected shopper ID. These filtered occasions needs to be forwarded to Redis utilizing its Stream API. Moreover, we have to set up an HTTP endpoint for Server-Despatched Occasions (SSE) that enables the required shopper to obtain real-time occasion updates.

Resolution Structure Overview

The structure consists of the next elements:

  • Kafka: A distributed occasion streaming platform that lets you publish and subscribe to streams of data (occasions).
  • Spring Boot: A framework for constructing Java functions. We’ll use it to create Kafka customers and Redis Stream producers. Subscribes to Kafka messages, filters occasions based mostly on the shopper ID, and forwards related occasions to Redis Streams.
  • Redis: A high-performance, in-memory knowledge retailer. We’ll use its Streams characteristic to deal with occasion streams. Shops the streamed occasions utilizing its Streams API.
  • Docker: A containerization platform. We’ll use Docker and Docker-Compose to create containers for Kafka, Redis, and our Spring Boot utility. We’ll make the most of this for a neighborhood POT, POC.
  • HTTP Server-Despatched Occasions (SSE) Endpoint: Offers real-time occasion updates to the shopper, filtering occasions based mostly on the shopper ID.

Redis Streams 

Redis Streams is a characteristic in Redis that gives a approach to deal with real-time knowledge streams with varied use instances. Listed here are some eventualities the place you would possibly need to use Redis Streams:

  • Actual-Time Occasion Processing: Redis Streams are glorious for processing and storing real-time occasions. You need to use it for issues like logging, monitoring, monitoring person actions, or any use case that entails dealing with a steady stream of occasions.
  • Process Queues: For those who want a dependable and distributed activity queue, Redis Streams generally is a nice selection. It lets you push duties right into a stream and have a number of customers course of these duties concurrently.
  • Exercise Feeds: For those who’re constructing a social community or any utility that requires exercise feeds, Redis Streams can effectively deal with the feed knowledge, making certain quick entry and scalability.
  • Message Brokering: Redis Streams can function a light-weight message dealer for microservices or different distributed programs. It could actually deal with message routing and be sure that messages are delivered to customers.
  • Actual-Time Analytics: When it’s essential analyze knowledge in real-time, Redis Streams might be helpful for storing the incoming knowledge after which processing and aggregating it utilizing Redis capabilities.
  • IoT Knowledge Ingestion: For those who’re coping with knowledge from Web of Issues (IoT) gadgets, Redis Streams can deal with the high-throughput and real-time nature of the information generated by these gadgets.
  • Logging and Audit Trails: Redis Streams can be utilized to retailer logs or audit trails in real-time, making it straightforward to investigate and troubleshoot points.
  • Stream Processing: If it’s essential course of a steady stream of information in a selected order (for instance, monetary transactions or sensor readings), Redis Streams might help you handle the information within the order it was obtained.


  • Docker and Docker Compose are put in.
  • Fundamental understanding of Spring Boot, Redis Streams, and Kafka.
  • Java 17 or larger
  • An HTTP shopper of your selection. I used [httpie]

After all, you will get the code here.


1. Set Up Docker Compose for the Backend Infrastructure and the Spring Boot App

Create a `docker-compose.yml` file to outline the companies:

model: '3.8'
      picture: confluentinc/cp-zookeeper:newest
         ZOOKEEPER_TICK_TIME: 2000
         - 22181:2181


      picture: confluentinc/cp-kafka:newest
         - zookeeper
         - 29092:29092
         - 9092:9092
         - 9093:9093
         KAFKA_BROKER_ID: 1
         KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181


      picture: redis:6.2-alpine
      #picture: redis:5.0.3-alpine
      restart: all the time
         - '6379:6379'
      #command: redis-server --save 20 1 --loglevel warning --requirepass eYVX7EwVmmxKPCDmwMtyKVge8oLd2t81
      command: redis-server /usr/native/and so forth/redis/redis.conf --loglevel verbose --save 20 1
         - cache:/knowledge
         - ./redis.conf:/usr/native/and so forth/redis/redis.conf
         - $PWD/redis-data:/var/lib/redis

         # - REDIS_REPLICATION_MODE=grasp


      driver: native

         driver: default
            - subnet:

Create the yaml for the appliance ‘sse-demo.yml’.

model: "3.8"
    picture: "sse/spring-sse-demo:newest"
      - "8080:8080"
      #- "51000-52000:51000-52000"

      - native.env

      - REDIS_HOST=
      - REDIS_PORT=6379


      identify: docker_node_net

2. Create Spring Boot Software

Create a Spring Boot utility with the required dependencies:

git checkout 

In `pom.xml,` add the required dependencies for Kafka and Redis integration.

3. Implement Kafka Shopper and Redis Stream Producer

Create a Kafka client that listens to Kafka occasions and sends them to Redis Streams. This part additionally consumes the Redis streams for the HTTP shoppers:

package deal com.taptech.sse.occasion;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.taptech.sse.utils.DurationSupplier;
import com.taptech.sse.utils.ObjectMapperFactory;
import com.taptech.sse.config.SSEProperties;
import jakarta.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.occasion.ApplicationStartedEvent;
import org.springframework.context.occasion.EventListener;
import org.springframework.knowledge.redis.core.ReactiveStringRedisTemplate;
import reactor.core.Disposable;
import reactor.core.writer.Flux;
import reactor.core.writer.Mono;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverRecord;
import reactor.util.operate.Tuple2;
import reactor.util.operate.Tuples;
import reactor.util.retry.Retry;

import java.nio.ByteBuffer;
import java.time.Period;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.operate.Perform;

public class DefaultEventReceiverService implements EventReceiverService 
	non-public static ultimate Logger logger = LoggerFactory.getLogger(DefaultEventReceiverService.class);

	public ultimate static String TEST_STREAM = "";
	public ultimate static String TEST_STREAM_KEY = "";
	public ultimate static String TEST_STREAM_VALUE = "";
	non-public static ultimate String EMPTY_STR = "";
	public static ultimate String CLIENT_STREAM_STARTED = "";
	public static ultimate String HASH_PREFIX = "hash.";

	non-public static ObjectMapper objectMapper = ObjectMapperFactory.createObjectMapper(ObjectMapperFactory.Scope.SINGLETON);

	ReactiveStringRedisTemplate redisTemplate;
	KafkaReceiver<String, String> kafkaReceiver;
	SSEProperties sseProperties;
	StreamReadOptions streamReadOptions;

	public DefaultEventReceiverService(ReactiveStringRedisTemplate redisTemplate, KafkaReceiver<String, String> kafkaReceiver,
									   SSEProperties sseProperties) 
		this.redisTemplate = redisTemplate;
		this.kafkaReceiver = kafkaReceiver;
		this.sseProperties = sseProperties;
		this.streamReadOptions = StreamReadOptions.empty().autoAcknowledge()
				.block(Period.of(sseProperties.getClientHoldSeconds(), ChronoUnit.SECONDS));

	static ultimate Perform<String,String> calculateHashKey = str -> new StringBuilder(HASH_PREFIX).append(str).toString();

	public void init() 

		this.redisTemplate.opsForValue().append(TEST_STREAM_KEY, TEST_STREAM_VALUE).subscribe();


	public Disposable startKafkaConsumer()"############# Beginning Kafka listener.....");
		return kafkaReceiver.obtain()
				.doOnError(error -> logger.error("Error receiving occasion, will retry", error))
				.retryWhen(Retry.fixedDelay(Lengthy.MAX_VALUE, Period.ofSeconds(sseProperties.getTopicRetryDelaySeconds())))
				.doOnNext(file ->"Acquired occasion: key ", file.key()))
				.filterWhen(file -> checkIfStreamBeingAccessed(file))
				.subscribe(file -> file.receiverOffset().acknowledge());

	Mono<Boolean> checkIfStreamBeingAccessed(ReceiverRecord<String,String> file)
		return this.redisTemplate.opsForHash().hasKey(calculateHashKey.apply(file.key()), CLIENT_STREAM_STARTED)
				.doOnNext(val ->"key => 's stream is being accessed ",file.key(),val));

	public Mono<ReceiverRecord<String, String>> handleEvent(ReceiverRecord<String, String> file) 
		return Mono.simply(file)
				.doOnError(ex -> logger.warn("Error processing occasion: key ", file.key(), ex))
				.onErrorResume(ex -> Mono.empty())
				.doOnNext(rec -> logger.debug("Efficiently processed occasion: key ", file.key()))

	public Mono<Tuple2<RecordId, ReceiverRecord<String, String>>> produce(ReceiverRecord<String, String> recRecord) 

		ObjectRecord<String, String> file = StreamRecords.newRecord()
		return this.redisTemplate.opsForStream().add(file)
				.map(recId -> Tuples.of(recId, recRecord));

	Perform<ObjectRecord<String, String>, NotificationEvent> convertToNotificationEvent() 
		return (file) -> 
			NotificationEvent occasion = null;
				occasion = objectMapper.readValue(file.getValue(), NotificationEvent.class);
			 catch (JsonProcessingException e) 
				occasion = new NotificationEvent();
			return occasion;

	non-public Mono<String> createGroup(String workspaceId)
		return redisTemplate.getConnectionFactory().getReactiveConnection().streamCommands()
				.xGroupCreate(ByteBuffer.wrap(workspaceId.getBytes()), workspaceId, ReadOffset.from("0-0"), true)
				.doOnError((error) -> 
					if (logger.isDebugEnabled())
						logger.debug("Couldn't create group.",error);
				.map(okStr -> workspaceId)
				.onErrorResume((error) -> Mono.simply(workspaceId));

	non-public Flux<NotificationEvent> findClientNotificationEvents(Shopper client, StreamOffset<String> streamOffset, DurationSupplier booleanSupplier)
		return this.redisTemplate.opsForStream().learn(String.class, client, streamReadOptions, streamOffset)

	public Flux<NotificationEvent> devour(ultimate String clientId)
		return Flux.from(createGroup(clientId))
				.flatMap(id -> addIdToStream(clientId))
				.map(id -> Tuples.of(StreamOffset.create(clientId, ReadOffset.lastConsumed()),
						Shopper.from(clientId, clientId),
						new DurationSupplier(Period.of(sseProperties.getClientHoldSeconds(), ChronoUnit.SECONDS),
				.flatMap(tuple3 -> findClientNotificationEvents(tuple3.getT2(), tuple3.getT1(), tuple3.getT3()));


	non-public Mono<String> addIdToStream(String id) 
		return this.redisTemplate.opsForHash().put(calculateHashKey.apply(id), CLIENT_STREAM_STARTED, Boolean.TRUE.toString()).map(val -> id);

	public Flux<Boolean> deleteWorkspaceStream(String workspaceId)
		StreamOffset<String> streamOffset = StreamOffset.create(workspaceId, ReadOffset.lastConsumed());
		StreamReadOptions streamReadOptions = StreamReadOptions.empty().noack();
		Shopper client = Shopper.from(workspaceId, workspaceId);

		return this.redisTemplate.opsForStream().learn(String.class, client, streamReadOptions, streamOffset)
				.flatMap(objRecord -> this.redisTemplate.opsForStream().delete(workspaceId,objRecord.getId()).map(val -> objRecord))
				.flatMap(objRecord -> this.redisTemplate.opsForHash().delete(workspaceId));

	public Flux<String> consumeString(String clientId) 
		return this.redisTemplate.opsForStream().learn(String.class, StreamOffset.newest(clientId)).map(ObjectRecord::getValue);

4. Configure Companies in Spring Boot


# Producer properties

# Frequent Kafka Properties


5. Construct Docker Picture for Spring Boot App

Utilizing the `kubernetes-maven-plugin` from jkube, create the picture on your Spring Boot utility:

./mvnw clear package deal -Dmaven.check.skip=true k8s:construct

6. Begin the Companies and Run the Software

From the src/check/sources/docker listing

Begin the companies:

Begin the app:

7. Hook up with a Stream Utilizing One of many IDS within the client-ids.json File 

http --stream GET  clientId==dd07bd51-1ab0-4e69-a0ff-f625fa9e7fc0

8. Generate Some Occasions

You are able to do an HTTP POST to /generateNE

http POST /generateNE

After this, watch as your HTTP shopper receives an occasion for the clientId that it’s subscribed to.


Why would you employ Kafka and Redis? Doesn’t Kafka supply this alone? Many firms have invested in Kafka as a backend message supplier between their programs. Kafka in itself doesn’t deal with message choice very simply. 

Message choice isn’t a typical characteristic offered natively by Kafka for a few causes:

  • Knowledge Measurement and Latency: Kafka is designed for high-throughput, low-latency message processing. Its structure focuses on distributing messages to numerous customers rapidly. Introducing message choice based mostly on arbitrary situations can decelerate the general processing and introduce latency, which works in opposition to Kafka’s major design targets.
  • Idempotency: Kafka depends on the idea of idempotent producers and customers. Which means that if a client or producer retries a message on account of a failure, it shouldn’t lead to duplicate processing. Introducing selective message retrieval would complicate this idempotency assure, probably resulting in unintended duplicate processing.
  • Shopper Offset Monitoring: Kafka maintains client offsets, permitting customers to maintain monitor of the final processed message. If message choice is launched, offsets turn into much less simple, as some messages could be skipped based mostly on choice standards.
  • Decoupled Structure: Kafka is designed to decouple producers from customers. Producers are unaware of client conduct, and customers can independently resolve what messages they need to devour. Message choice would break this decoupling, as producers would wish to know which messages to supply based mostly on particular client wants.
  • Shopper Flexibility: Kafka customers might be extremely versatile by way of message processing. They are often designed to filter, rework, and mixture messages based mostly on their very own standards. Introducing message choice on the Kafka stage would restrict this flexibility and make the system much less adaptable to altering client necessities.
  • Scaling and Parallelism: Kafka’s scalability and parallelism advantages come from the power to distribute messages throughout a number of partitions and permit a number of customers to course of messages in parallel. Selective message retrieval would complicate this parallelism, making it more durable to distribute work effectively.

Whereas Kafka itself would not present native message choice options, it is important to design the customers to deal with message filtering and choice if wanted. Customers might be designed to filter and course of messages based mostly on particular standards, making certain that solely related messages are processed throughout the client utility. This strategy permits Kafka to take care of its core design ideas whereas nonetheless offering the pliability wanted for varied message-processing eventualities.

Kafka couldn’t basically resolve the issue in a straightforward manner, which result in pushing the messages to a different persistent house that would simply choose based mostly on identified standards. This requirement results in the choice to make use of Redis and permit pushing messages on to Redis.

A choice was made to restrict the occasions being pushed into Redis based mostly on whether or not there was a shopper truly anticipating a message. If there have been no shoppers, then Kafka messages have been being filtered out.

.filterWhen(file -> checkIfStreamBeingAccessed(file))

The shopper registers the id in order that the Kafka listener will push the to the Redis stream. 

.flatMap(id -> addIdToStream(clientId))


By following the steps outlined on this doc, we now have efficiently applied an occasion streaming structure that takes occasions from Kafka, filters them based mostly on a selected shopper ID, and forwards the related occasions to Redis utilizing its Stream API. The SSE endpoint permits shoppers to obtain real-time occasion updates tailor-made to their respective shopper IDs. This answer offers an environment friendly and scalable approach to deal with occasion streaming for focused shoppers.

#Constructing #Reactive #Occasion #Streaming #Structure #DZone



Please enter your comment!
Please enter your name here

Most Popular

Skip to toolbar