Готовый пример

src/main/java/ru/ldeloff/demokafkastream/DemoKafkaStreamApplication.java

@@ -1,49 +1,114 @@
 package ru.ldeloff.demokafkastream;
 
 
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.*;
+import org.apache.kafka.common.network.Mode;
+import org.apache.logging.log4j.CloseableThreadContext;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.context.MessageSource;
 import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.apache.kafka.common.serialization.StringSerializer;
 
+import java.io.DataInput;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Objects;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 
+@RestController
 @SpringBootApplication
 public class DemoKafkaStreamApplication {
 
     private int i = 0;
 
+    @Autowired
+    private StreamBridge streamBridge;
+
     public static void main(String[] args) {
         SpringApplication.run(DemoKafkaStreamApplication.class, args);
     }
 
-    // Пуляю в кафку 1
+    /* Кафка 1 */
+    // Консьюмер
     @Bean
-    public Consumer<String> consumer1() {
-        return s -> System.out.println("Data Consumer 1 (Kafka-1):" + s);
+    public Consumer<Model> consumerAuto1() {
+        return message -> {;
+    //        System.out.println("Consume auto 1: " + message);
+        };
     }
 
-//    // В данном примере продюссер отправляет сообщения периодически
-//    @Bean
-//    public Supplier<String> producer1() {
-//        return () -> {
-//            System.out.println("Data Supplier 1 (Kafka-1):" + ++i);
-//            return "Data Supplier 1 (Kafka-1):" + i;
-//        };
-//    }
-
-    // Пуляю в кафку 2
+    // Продюсер (сам периодически отправляет сообщения)
     @Bean
-    public Consumer<String> consumer2() {
-        return s -> System.out.println("Data Consumer 2 (Kafka-2):" + s);
+    public Supplier<Model> producerAuto1() {
+        return () -> {
+            Model message = new Model(++i);
+    //        System.out.println("Produce auto 1:" + message);
+            return message;
+        };
     }
 
-//    // В данном примере продюссер отправляет сообщения периодически
-//    @Bean
-//    public Supplier<String> producer2() {
-//        return () -> {
-//            System.out.println("Data Supplier 2 (Kafka-2):" + ++i);
-//            return "Data Supplier 2 (Kafka-2):" + i;
-//        };
-//    }
 
+
+    // Вручную отправим сообщение
+    @GetMapping(value = "/test1")
+    public String test1() {
+        Model message = new Model(++i);
+        System.out.println("Produce manual (kafka 1): " + message);
+        streamBridge.send("producerManual1-out-0", message);
+        return "test OK";
+    }
+
+    // Консьюмер
+    @Bean
+    public Consumer<Model> consumerManual1() {
+        return message -> {
+            System.out.println("Consume manual 1: " + message);
+        };
+    }
+
+
+    /* Кафка 2 */
+    // Продюсер (сам периодически отправляет сообщения)
+    @Bean
+    public Supplier<Model> producerAuto2() {
+        return () -> {
+            Model message = new Model(++i);
+    //        System.out.println("Produce auto 2:" + message);
+            return message;
+        };
+    }
+
+    // Консьюмер
+    @Bean
+    public Consumer<Model> consumerAuto2() {
+        return message -> {
+    //        System.out.println("Consume auto 2: " + message);
+        };
+    }
+
+    // Вручную отправим сообщение
+    @GetMapping(value = "/test2")
+    public String test2() {
+        Model message = new Model(++i);
+        System.out.println("Produce manual (kafka 2): " + message);
+        streamBridge.send("producerManual2-out-0", message);
+        return "test OK";
+    }
+
+    // Консьюмер
+    @Bean
+    public Consumer<Model> consumerManual2() {
+        return message -> {
+            System.out.println("Consume manual 2: " + message);
+        };
+    }
 }

src/main/java/ru/ldeloff/demokafkastream/Model.java

@@ -0,0 +1,19 @@
+package ru.ldeloff.demokafkastream;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+
+@Getter
+@Setter
+@ToString
+@NoArgsConstructor
+public class Model {
+    private int id;
+    private long time;
+    public Model(int id) {
+        this.id = id;
+        this.time = System.currentTimeMillis();
+    }
+}

src/main/java/ru/ldeloff/demokafkastream/controller.java

@@ -1,30 +0,0 @@
-package ru.ldeloff.demokafkastream;
-
-import lombok.RequiredArgsConstructor;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-
-@RestController
-@RequiredArgsConstructor
-public class controller {
-
-    private final StreamBridge streamBridge;
-
-    @GetMapping(value = "/test1")
-    public String test1() {
-        String str = "Produce manual (kafka 1) " + System.currentTimeMillis();
-        System.out.println(str);
-        streamBridge.send("producer1-out-0", str);
-        return "test OK";
-    }
-
-    @GetMapping(value = "/test2")
-    public String test2() {
-        String str = "Produce manual (kafka 2) " + System.currentTimeMillis();
-        System.out.println(str);
-        streamBridge.send("producer2-out-0", str);
-        return "test OK";
-    }
-}

src/main/resources/application.yml

@@ -4,22 +4,36 @@ spring:
 
   cloud:
     function:
-      definition: producer1;consumer1;producer2;consumer2
+      definition: producerAuto1;consumerAuto1;producerManual1;consumerManual1;producerAuto2;consumerAuto2;producerManual2;consumerManual2
     stream:
       bindings:
-        producer1-out-0:
+        producerAuto1-out-0:
           binder: kafka-1
-          destination: kafka_1_topic
-        consumer1-in-0:
+          destination: kafka_1_topic_auto
+        consumerAuto1-in-0:
           binder: kafka-1
-          destination: kafka_1_topic
+          destination: kafka_1_topic_auto
 
-        producer2-out-0:
+        producerManual1-out-0:
+          binder: kafka-1
+          destination: kafka_1_topic_manual
+        consumerManual1-in-0:
+          binder: kafka-1
+          destination: kafka_1_topic_manual
+
+        producerAuto2-out-0:
           binder: kafka-2
-          destination: kafka_2_topic
-        consumer2-in-0:
+          destination: kafka_2_topic_auto
+        consumerAuto2-in-0:
           binder: kafka-2
-          destination: kafka_2_topic
+          destination: kafka_2_topic_auto
+
+        producerManual2-out-0:
+          binder: kafka-2
+          destination: kafka_2_topic_manual
+        consumerManual2-in-0:
+          binder: kafka-2
+          destination: kafka_2_topic_manual
 
       binders:
         kafka-1:
@@ -32,6 +46,7 @@ spring:
                     binder:
                       brokers:
                         - 192.168.10.101:29092
+
         kafka-2:
           type: kafka
           environment:
@@ -42,4 +57,3 @@ spring:
                     binder:
                       brokers:
                         - 192.168.10.101:39092
-