Introducción y ejemplo de procesar mensajes con Spring Cloud Stream

Escrito por picodotdev el 17/06/2026.
java planeta-codigo
Enlace permanente Comentarios

Las aplicaciones distribuidas emplean comunicación basada en mensajes para lograr un intercambio asíncrono y desacoplado entre servicios. Spring Kafka ofrece una integración directa con Apache Kafka mediante la anotación KafkaListener y KafkaTemplate, sin embargo, cuando los mensajes fluyen entre múltiples topics, esta aproximación de bajo nivel se vuelve compleja. Spring Cloud Stream resuelve este problema simplificando el modelo de desarrollo al usar las interfaces funcionales de Java, Consumer, Function y Supplier para modelar consumidores, procesadores y productores de forma agnística al broker de mensajes, declarativa y altamente configurable.

Spring

Las aplicaciones distribuidas suelen emplear mecanismos de comunicación basados en mensajes. La comunicación basada en mensajes tiene la particularidad deseable en algunos casos de ser una forma de comunicación asíncrona y desacoplada entre el productor del mensaje y el consumidor del mensaje. Esto hace a las aplicaciones procesadores de un flujo o stream constante de mensajes.

Entre las muchas librerías que ofrece el framework de Spring está la integración con el sistema de mensajería de Apache Kafka. Esta librería permite el consumo y producción de mensajes de los topics. Con la anotación KafkaListener se define el punto de entrada en el código de los mensaje de Kafka, con la clase KafkaTemplate permite hacer el envío de mensajes.

Introducción, conceptos y uso básico del broker de mensajes Apache Kafka

Algunas aplicaciones no solo escriben y leen de un topic de Kafka, los mensajes siguen un flujo de procesamiento entre varios topics. Por otro lado, la librería de Spring Kafka ofrece una integración a bajo nivel que se vuelve compleja en integraciones con varios consumidores y productores. Spring Cloud Stream simplifica el modelo de desarrollo de Spring Kafka, al mismo tiempo que lo hace más configurable.

Contenido del artículo

Spring Cloud Stream

Spring Cloud Stream permite en las aplicaciones de Spring utilizar las interfaces funcionales de Java como Consumer, Function y Supplier para modelar consumidores, procesadores y productores de mensajes.

Otra de sus propiedades es que permite la integración con diferentes tipos de sistemas de mensajería como Kafka o RabbitMQ con cambios mínimos en el código, ya que el modelo de programación está basado en interfaces estándar de Java agnósticas de los brokers de menesajes.

Conceptos

Binding

Las instancias de interfaces funcionales que procesan mensajes de Kafka se denominan bindings. No son más que un bean de Spring que implementa una de las interfaces funcionales.

Estas definiciones de bean junto con cierta configuración para definir de que topics leen los mensajes y en cual se escriben permiten el procesado de los mensajes. Las propiedades destination determinan de que topics se leen los mensajes y escriben los mensajes.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


package io.github.picodotdev.blogbitix.springcloudstream.function;

import java.util.Locale;
import java.util.function.Function;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

public class UppercaseFunction implements Function<Message<String>, Message<String>> {

    private static final Logger log = LogManager.getLogger(LoggerConsumer.class);

    @Override
    public Message<String> apply(Message<String> message) {
        if (message == null) {
            return null;
        }
        String upperCase = message.getPayload().toUpperCase(Locale.ROOT);
        log.info("Uppercase message: {}, {}", message.getPayload(), upperCase);
        return MessageBuilder.withPayload(upperCase).build();
    }
}

UppercaseFunction.java

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


package io.github.picodotdev.blogbitix.springcloudstream;

...

@Component
public class Beans {

    @Bean
    Function<Message<String>, Message<String>> uppercase() {
        return new UppercaseFunction();
    }

    @Bean
    Consumer<Message<String>> logger() {
        return new LoggerConsumer();
    }
}

Beans.java

Binding names

El nombre del método de la interfaz funcional determina el nombre del binding. Con cierta convención en los nombres se establecen las propiedades de configuración específica en cada uno de los bindings. El nombre del binding está determinado por el nombre de la funció, in y out corresponding al tipo del binding y finalmente un índice que suele ser 0 para la función típica con un solo elemento de entrada y de salida.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


spring:
  cloud:
    ...
    stream:
      bindings:
        uppercase-in-0:
          destination: message
          group: uppercase-function
        uppercase-out-0:
          destination: message-uppercase
        logger-in-0:
          destination: message-uppercase
          group: logger-consumer

application-1.yml

1
2
3


uppercase-in-0
uppercase-out-0
logger-in-0

binding-names.txt

Es posible dar un nombre personalizado al binding y establecer las propiedades de configuración usando ese nuevo nombre.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


spring:
  cloud:
    ...
    function:
        bindings:
            uppercase-in-0: uppercase-in
            uppercase-out-0: uppercase-out
            logger-in-0: logger-in
    stream:
      bindings:
        uppercase-in:
          destination: message
          group: uppercase-function
        uppercase-out:
          destination: message-uppercase
        logger-in:
          destination: message-uppercase
          group: logger-consumer

application-2.yml

Binder

Los binder son los sistemas de mensajería con los que se integran los bindings. Hay varios, uno de ellos es Kafka pero también están RabbitMQ y Apache Pulsar.

Propiedades de configuración

Buena parte del uso de Spring Cloud Streams se realiza en las propiedades de configuración. Se pueden establecer propiedades de configuración que afectan a todos los consumidores y productores o un consumidor o productor específico. Y propiedades específicas del binder.

La lista completa de propiedades que permiten los consumidores y productores de Kafka son las siguientes.

Binder

Con la siguientes propiedades se especifica la configuración de conexión a los brokers de Kafka.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: kafka-1.localhost:9092
          configuration:
            security:
              protocol: SASL_SSL
            sasl:
              mechanism: PLAIN
              jaas:
                config: org.apache.kafka.common.security.plain.PlainLoginModule required username="${username}" password="${password}";
        default:
          consumer:
            property: value
          producer:
            property: value

application-3.yml

Consumer

En el consumidor es posible querer especificar el serializer y deserializer de los mensajes, la configuración de procesamiento en batches o el grupo del consumidor en caso de varias instancias.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


spring:
  cloud:
    stream:
      kafka:
        bindings:
          input-in-0:
            consumer:
              configuration:
                key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
                value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer

application-4.yml

Producer

En el productor también es posible especificar el serializer para la key y value del mensaje de Kafka, usando las propiedades de Kafka para los productores. Hay otras propiedades que definen Spring Cloud Stream.

Gestión de errores

Por defecto, la gestión de errores al procesar un mensaje es loggearlo y descartar. Es posible implementar handlers personalizados en caso de errores, basta con definir un consumer de ErrorMessage.

1
2
3
4
5
6


@Bean
public Consumer<ErrorMessage> customErrorHandler(){
    return errorMessage -> {
        logger.error(...);
    };
}

CustomErrorHandler.java

1
2
3
4
5
6


spring:
  cloud:
    stream:
      bindings:
        uppercase-in-0:
          error-handler-definition: customErrorHandler

application-5.yml

En caso de no querer descartar un mensaje cuando falla su procesamiento es posible enviarlo a un topic especial con el patrón dead-letter-queue para su revisión y posterior procesamiento. La siguiente configuración crea un topic automáticamente para el consumidor.

1
2
3
4
5
6
7
8


spring:
  cloud:
    stream:
      kafka:
        bindings:
          uppercase-in-0:
            consumer:
              auto-bind-dlq: true

application-6.yml

Con la siguiente propiedad se configura el número máximo de intentos de procesado antes de enviarlo al topic dead-letter-queue. En este ejemplo el procesado se realiza 3 veces.

1
2
3
4
5
6
7
8


spring:
  cloud:
    stream:
      bindings:
        uppercase-in-0:
          consumer:
            auto-bind-dlq: true
            max-attempts: 3

application-7.yml

Observabilidad

Para implementar métricas de observabilidad hay que añadir las siguientes dependencias y la aplicación se autoconfigura.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


dependencies {
    implementation(platform("org.springframework.cloud:spring-cloud-dependencies:2025.1.2"))

    implementation("org.springframework.boot:spring-boot-starter")
    implementation("org.springframework.boot:spring-boot-starter-web")

    ...

    implementation("org.springframework.cloud:spring-cloud-stream")
    implementation("org.springframework.cloud:spring-cloud-stream-binder-kafka")

    //implementation("org.springframework.boot:spring-boot-starter-actuator")
    //implementation("io.micrometer:micrometer-tracing-bridge-otel")

    testImplementation("org.springframework.boot:spring-boot-starter-test")
    testImplementation("org.springframework.cloud:spring-cloud-stream-test-binder")
}

build.gradle.kts

Pruebas unitarias

Spring Cloud Stream ofrece la posibilidad de probar de forma unitaria a los consumidores y productores. Ya sea la función directamente o como un binding de Spring Cloud Stream.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30


package io.github.picodotdev.blogbitix.springcloudstream.function;

import org.junit.jupiter.api.Test;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import static org.assertj.core.api.Assertions.assertThat;

class UppercaseFunctionTest {

    private final UppercaseFunction function = new UppercaseFunction();

    @Test
    void convertsPayloadToUpperCase() {
        Message<String> upperCase = function.apply(MessageBuilder.withPayload("hello world").build());
        assertThat(upperCase.getPayload()).isEqualTo("HELLO WORLD");
    }

    @Test
    void leavesAlreadyUppercasePayloadUnchanged() {
        Message<String> upperCase = function.apply(MessageBuilder.withPayload("HELLO WORLD").build());
        assertThat(upperCase.getPayload()).isEqualTo("HELLO WORLD");
    }

    @Test
    void handlesEmptyPayload() {
        Message<String> upperCase = function.apply(MessageBuilder.withPayload("").build());
        assertThat(upperCase.getPayload()).isEmpty();
    }
}

UppercaseFunctionTest.java

El código fuente completo del ejemplo puedes descargarlo del repositorio de ejemplos de Blog Bitix alojado en GitHub y probarlo en tu equipo ejecutando siguiente comando:
./gradlew run

Referencia:

Introduction to Spring Cloud Stream

Introducción y ejemplo de procesar mensajes con Spring Cloud Stream

Spring Cloud Stream

Conceptos

Binding

Binding names

Binder

Propiedades de configuración

Binder

Consumer

Producer

Gestión de errores

Observabilidad

Pruebas unitarias

Parchear una clase de una librería Java para solucionar un error

Novedades y nuevas características de Java 25

Generar clientes REST con su interfaz OpenAPI

Novedades y nuevas características de Java 21

Novedades de Java 20

Cifrar y descifrar datos usando algoritmos de clave asimétrica con Java