Compare commits
No commits in common. "master" and "add-key" have entirely different histories.
|
|
@ -1,102 +0,0 @@
|
||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
||||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
|
||||||
<modelVersion>4.0.0</modelVersion>
|
|
||||||
<parent>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-parent</artifactId>
|
|
||||||
<version>3.3.0</version>
|
|
||||||
<relativePath/> <!-- lookup parent from repository -->
|
|
||||||
</parent>
|
|
||||||
<groupId>ru.ldeloff</groupId>
|
|
||||||
<artifactId>KafkaStream</artifactId>
|
|
||||||
<version>0.0.1-SNAPSHOT</version>
|
|
||||||
<name>KafkaStream</name>
|
|
||||||
<description>KafkaStream</description>
|
|
||||||
<properties>
|
|
||||||
<java.version>22</java.version>
|
|
||||||
<spring-cloud.version>2023.0.1</spring-cloud.version>
|
|
||||||
</properties>
|
|
||||||
<dependencyManagement>
|
|
||||||
<dependencies>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.cloud</groupId>
|
|
||||||
<artifactId>spring-cloud-dependencies</artifactId>
|
|
||||||
<version>${spring-cloud.version}</version>
|
|
||||||
<type>pom</type>
|
|
||||||
<scope>import</scope>
|
|
||||||
</dependency>
|
|
||||||
</dependencies>
|
|
||||||
</dependencyManagement>
|
|
||||||
<dependencies>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.cloud</groupId>
|
|
||||||
<artifactId>spring-cloud-stream</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.cloud</groupId>
|
|
||||||
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-webflux</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>io.github.resilience4j</groupId>
|
|
||||||
<artifactId>resilience4j-spring-boot2</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-aop</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.projectlombok</groupId>
|
|
||||||
<artifactId>lombok</artifactId>
|
|
||||||
<optional>true</optional>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-test</artifactId>
|
|
||||||
<scope>test</scope>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.cloud</groupId>
|
|
||||||
<artifactId>spring-cloud-stream-test-binder</artifactId>
|
|
||||||
<scope>test</scope>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-web</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.cloud</groupId>
|
|
||||||
<artifactId>spring-cloud-function-web</artifactId>
|
|
||||||
</dependency>
|
|
||||||
</dependencies>
|
|
||||||
|
|
||||||
<build>
|
|
||||||
<plugins>
|
|
||||||
<plugin>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
|
||||||
<configuration>
|
|
||||||
<excludes>
|
|
||||||
<exclude>
|
|
||||||
<groupId>org.projectlombok</groupId>
|
|
||||||
<artifactId>lombok</artifactId>
|
|
||||||
</exclude>
|
|
||||||
</excludes>
|
|
||||||
</configuration>
|
|
||||||
</plugin>
|
|
||||||
</plugins>
|
|
||||||
</build>
|
|
||||||
|
|
||||||
</project>
|
|
||||||
|
|
@ -1,13 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream;
|
|
||||||
|
|
||||||
|
|
||||||
import org.springframework.boot.SpringApplication;
|
|
||||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
|
||||||
|
|
||||||
|
|
||||||
@SpringBootApplication
|
|
||||||
public class KafkaStreamApplication {
|
|
||||||
public static void main(String[] args) {
|
|
||||||
SpringApplication.run(KafkaStreamApplication.class, args);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,18 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.config;
|
|
||||||
|
|
||||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
|
||||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
public class CircuitBreakerConfig {
|
|
||||||
private final CircuitBreakerRegistry circuitBreakerRegistry;
|
|
||||||
|
|
||||||
public CircuitBreakerConfig(CircuitBreakerRegistry circuitBreakerRegistry){
|
|
||||||
this.circuitBreakerRegistry = circuitBreakerRegistry;
|
|
||||||
}
|
|
||||||
|
|
||||||
public CircuitBreaker circuitBreaker(String cbName) {
|
|
||||||
return circuitBreakerRegistry.circuitBreaker(cbName);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,34 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.config;
|
|
||||||
|
|
||||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
|
||||||
import io.github.resilience4j.circuitbreaker.event.CircuitBreakerOnStateTransitionEvent;
|
|
||||||
import org.springframework.cloud.stream.binding.BindingsLifecycleController;
|
|
||||||
import org.springframework.cloud.stream.endpoint.BindingsEndpoint;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
public class CircuitBreakerImpl {
|
|
||||||
private final BindingsEndpoint bindingsEndpoint;
|
|
||||||
|
|
||||||
protected CircuitBreakerImpl(BindingsEndpoint bindingsEndpoint,
|
|
||||||
CircuitBreakerConfig circuitBreakerConfig) {
|
|
||||||
this.bindingsEndpoint = bindingsEndpoint;
|
|
||||||
CircuitBreaker circuitBreaker = circuitBreakerConfig.circuitBreaker("consumerCircuitBreaker");
|
|
||||||
circuitBreaker.getEventPublisher().onStateTransition(this::onStateChangeEvent);
|
|
||||||
}
|
|
||||||
|
|
||||||
private void onStateChangeEvent(CircuitBreakerOnStateTransitionEvent event) {
|
|
||||||
switch (event.getStateTransition().getToState()) {
|
|
||||||
case OPEN:
|
|
||||||
System.out.println("consumerAutoTestCircuitBreaker1 open");
|
|
||||||
bindingsEndpoint.changeState("consumerAutoTestCircuitBreaker1-in-0", BindingsLifecycleController.State.STOPPED);
|
|
||||||
break;
|
|
||||||
case CLOSED:
|
|
||||||
System.out.println("consumerAutoTestCircuitBreaker1 closed");
|
|
||||||
case HALF_OPEN:
|
|
||||||
System.out.println("consumerAutoTestCircuitBreaker1 half_open");
|
|
||||||
bindingsEndpoint.changeState("consumerAutoTestCircuitBreaker1-in-0", BindingsLifecycleController.State.STARTED);
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,55 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.config;
|
|
||||||
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
import ru.ldeloff.kafkastream.model.Model;
|
|
||||||
import ru.ldeloff.kafkastream.service.ModelService;
|
|
||||||
|
|
||||||
import java.util.function.Consumer;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class KafkaConsumerConfig {
|
|
||||||
private final ModelService modelService;
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public Consumer<Model> consumerAuto1() {
|
|
||||||
return message -> {
|
|
||||||
// System.out.println("Consume auto 1: " + message);
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public Consumer<Model> consumerManual1() {
|
|
||||||
return message -> {
|
|
||||||
System.out.println("Consume manual 1: " + message);
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public Consumer<Model> consumerAutoTestCircuitBreaker1() {
|
|
||||||
return message -> {
|
|
||||||
System.out.println("Consume Auto CB 1: " + message);
|
|
||||||
try {
|
|
||||||
modelService.sendModel(message);
|
|
||||||
} catch (Exception e) {
|
|
||||||
System.out.println(e.getClass());
|
|
||||||
}
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public Consumer<Model> consumerAuto2() {
|
|
||||||
return message -> {
|
|
||||||
// System.out.println("Consume auto 2: " + message);
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public Consumer<Model> consumerManual2() {
|
|
||||||
return message -> {
|
|
||||||
System.out.println("Consume manual 2: " + message);
|
|
||||||
};
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,68 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.config;
|
|
||||||
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import org.springframework.cloud.stream.function.StreamBridge;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
import org.springframework.kafka.support.KafkaHeaders;
|
|
||||||
import org.springframework.messaging.Message;
|
|
||||||
import org.springframework.messaging.support.MessageBuilder;
|
|
||||||
import ru.ldeloff.kafkastream.model.Model;
|
|
||||||
|
|
||||||
import java.util.function.Supplier;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class KafkaProducerConfig {
|
|
||||||
|
|
||||||
private int i = 0;
|
|
||||||
private final StreamBridge streamBridge;
|
|
||||||
|
|
||||||
// Периодически отправляет сообщения
|
|
||||||
@Bean
|
|
||||||
public Supplier<Message<Model>> producerAuto1() {
|
|
||||||
return () -> {
|
|
||||||
Model message = new Model(++i);
|
|
||||||
// System.out.println("Produce auto 1:" + message);
|
|
||||||
return MessageBuilder.withPayload(message)
|
|
||||||
.setHeader(KafkaHeaders.KEY, String.valueOf(message.getId()))
|
|
||||||
.build();
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
// Ручная отправка
|
|
||||||
public void sendMessage1(Model message) {
|
|
||||||
System.out.println("Produce manual (kafka 1): " + message);
|
|
||||||
streamBridge.send("producerManual1-out-0",
|
|
||||||
MessageBuilder.withPayload(message)
|
|
||||||
.setHeader(KafkaHeaders.KEY, String.valueOf(message.getId()))
|
|
||||||
.build());
|
|
||||||
}
|
|
||||||
|
|
||||||
// Для Circuit Breaker
|
|
||||||
@Bean
|
|
||||||
public Supplier<Message<Model>> producerAutoTestCircuitBreaker1() {
|
|
||||||
return () -> {
|
|
||||||
Model message = new Model(++i);
|
|
||||||
System.out.println("Produce auto (CB) 1:" + message);
|
|
||||||
return MessageBuilder.withPayload(message)
|
|
||||||
.setHeader(KafkaHeaders.KEY, String.valueOf(message.getId()))
|
|
||||||
.build();
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/* Кафка 2 */
|
|
||||||
@Bean
|
|
||||||
public Supplier<Model> producerAuto2() {
|
|
||||||
return () -> {
|
|
||||||
Model message = new Model(++i);
|
|
||||||
// System.out.println("Produce auto 2:" + message);
|
|
||||||
return message;
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
public void sendMessage2(Model message) {
|
|
||||||
System.out.println("Produce manual (kafka 2): " + message);
|
|
||||||
streamBridge.send("producerManual2-out-0", message);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,20 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.config;
|
|
||||||
|
|
||||||
import org.springframework.cloud.stream.binding.BindingsLifecycleController;
|
|
||||||
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
|
|
||||||
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
|
|
||||||
import org.springframework.cloud.stream.endpoint.BindingsEndpoint;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
|
|
||||||
import java.util.List;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
public class KafkaStreamConfig {
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public BindingsEndpoint bindingsEndpoint(List<InputBindingLifecycle> inputBindings,
|
|
||||||
List<OutputBindingLifecycle> outputBindings) {
|
|
||||||
return new BindingsEndpoint(new BindingsLifecycleController(inputBindings, outputBindings));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,28 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.config;
|
|
||||||
|
|
||||||
import io.netty.channel.ChannelOption;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
import org.springframework.http.HttpHeaders;
|
|
||||||
import org.springframework.http.MediaType;
|
|
||||||
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
|
|
||||||
import org.springframework.web.reactive.function.client.WebClient;
|
|
||||||
import reactor.netty.http.client.HttpClient;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
public class WebConfig {
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public WebClient webClient() {
|
|
||||||
return WebClient.builder()
|
|
||||||
.baseUrl("http://localhost:3000")
|
|
||||||
.clientConnector(new ReactorClientHttpConnector(
|
|
||||||
HttpClient.create()
|
|
||||||
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 100)
|
|
||||||
.wiretap(true)
|
|
||||||
)
|
|
||||||
)
|
|
||||||
.defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
|
|
||||||
.build();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,43 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.controller;
|
|
||||||
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import org.springframework.cloud.stream.binding.BindingsLifecycleController;
|
|
||||||
import org.springframework.cloud.stream.endpoint.BindingsEndpoint;
|
|
||||||
import org.springframework.web.bind.annotation.GetMapping;
|
|
||||||
import org.springframework.web.bind.annotation.RestController;
|
|
||||||
import ru.ldeloff.kafkastream.config.KafkaProducerConfig;
|
|
||||||
import ru.ldeloff.kafkastream.model.Model;
|
|
||||||
|
|
||||||
@RestController
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class Controller {
|
|
||||||
int i = 0;
|
|
||||||
private final KafkaProducerConfig kafkaProducerConfig;
|
|
||||||
private final BindingsEndpoint bindingsEndpoint;
|
|
||||||
|
|
||||||
// Ручная отправка сообщений
|
|
||||||
@GetMapping(value = "/test1")
|
|
||||||
public String test1() {
|
|
||||||
kafkaProducerConfig.sendMessage1(new Model(++i));
|
|
||||||
return "test OK, i = " + i;
|
|
||||||
}
|
|
||||||
|
|
||||||
@GetMapping(value = "/test2")
|
|
||||||
public String test2() {
|
|
||||||
kafkaProducerConfig.sendMessage2(new Model(++i));
|
|
||||||
return "test OK, i = " + i;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Переключатели состояний consumerAutoTestCircuitBreaker1-in-0
|
|
||||||
@GetMapping(value = "/test3")
|
|
||||||
public String test3() {
|
|
||||||
bindingsEndpoint.changeState("consumerAutoTestCircuitBreaker1-in-0", BindingsLifecycleController.State.STOPPED);
|
|
||||||
return "test OK";
|
|
||||||
}
|
|
||||||
|
|
||||||
@GetMapping(value = "/test4")
|
|
||||||
public String test4() {
|
|
||||||
bindingsEndpoint.changeState("consumerAutoTestCircuitBreaker1-in-0", BindingsLifecycleController.State.STARTED);
|
|
||||||
return "test OK";
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,19 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.model;
|
|
||||||
|
|
||||||
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();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,27 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream.service;
|
|
||||||
|
|
||||||
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
|
|
||||||
import io.github.resilience4j.retry.annotation.Retry;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import org.springframework.stereotype.Service;
|
|
||||||
import org.springframework.web.reactive.function.BodyInserters;
|
|
||||||
import org.springframework.web.reactive.function.client.WebClient;
|
|
||||||
import ru.ldeloff.kafkastream.model.Model;
|
|
||||||
|
|
||||||
@Service
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class ModelService {
|
|
||||||
private final WebClient webClient;
|
|
||||||
|
|
||||||
@CircuitBreaker(name = "consumerCircuitBreaker")
|
|
||||||
@Retry(name = "consumerCircuitBreaker")
|
|
||||||
public void sendModel(Model model) {
|
|
||||||
System.out.println("Sending Model: " + model);
|
|
||||||
Model model1 = webClient.post()
|
|
||||||
.uri("/models")
|
|
||||||
.bodyValue(BodyInserters.fromValue(model))
|
|
||||||
.retrieve()
|
|
||||||
.bodyToMono(Model.class).block();
|
|
||||||
System.out.println(model1);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,13 +0,0 @@
|
||||||
package ru.ldeloff.kafkastream;
|
|
||||||
|
|
||||||
import org.junit.jupiter.api.Test;
|
|
||||||
import org.springframework.boot.test.context.SpringBootTest;
|
|
||||||
|
|
||||||
@SpringBootTest
|
|
||||||
class KafkaStreamApplicationTests {
|
|
||||||
|
|
||||||
@Test
|
|
||||||
void contextLoads() {
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,59 +0,0 @@
|
||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
||||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
|
||||||
<modelVersion>4.0.0</modelVersion>
|
|
||||||
<parent>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-parent</artifactId>
|
|
||||||
<version>3.3.0</version>
|
|
||||||
<relativePath/> <!-- lookup parent from repository -->
|
|
||||||
</parent>
|
|
||||||
<groupId>ru.ldeloff</groupId>
|
|
||||||
<artifactId>Server</artifactId>
|
|
||||||
<version>0.0.1-SNAPSHOT</version>
|
|
||||||
<name>Server</name>
|
|
||||||
<description>Server</description>
|
|
||||||
<properties>
|
|
||||||
<java.version>17</java.version>
|
|
||||||
</properties>
|
|
||||||
<dependencies>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-web</artifactId>
|
|
||||||
</dependency>
|
|
||||||
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
|
||||||
<optional>true</optional>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.projectlombok</groupId>
|
|
||||||
<artifactId>lombok</artifactId>
|
|
||||||
<optional>true</optional>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-starter-test</artifactId>
|
|
||||||
<scope>test</scope>
|
|
||||||
</dependency>
|
|
||||||
</dependencies>
|
|
||||||
|
|
||||||
<build>
|
|
||||||
<plugins>
|
|
||||||
<plugin>
|
|
||||||
<groupId>org.springframework.boot</groupId>
|
|
||||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
|
||||||
<configuration>
|
|
||||||
<excludes>
|
|
||||||
<exclude>
|
|
||||||
<groupId>org.projectlombok</groupId>
|
|
||||||
<artifactId>lombok</artifactId>
|
|
||||||
</exclude>
|
|
||||||
</excludes>
|
|
||||||
</configuration>
|
|
||||||
</plugin>
|
|
||||||
</plugins>
|
|
||||||
</build>
|
|
||||||
|
|
||||||
</project>
|
|
||||||
|
|
@ -1,13 +0,0 @@
|
||||||
package ru.ldeloff.server;
|
|
||||||
|
|
||||||
import org.springframework.boot.SpringApplication;
|
|
||||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
|
||||||
|
|
||||||
@SpringBootApplication
|
|
||||||
public class ServerApplication {
|
|
||||||
|
|
||||||
public static void main(String[] args) {
|
|
||||||
SpringApplication.run(ServerApplication.class, args);
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,15 +0,0 @@
|
||||||
package ru.ldeloff.server.controller;
|
|
||||||
|
|
||||||
import org.springframework.web.bind.annotation.PostMapping;
|
|
||||||
import org.springframework.web.bind.annotation.RestController;
|
|
||||||
import ru.ldeloff.server.model.Model;
|
|
||||||
|
|
||||||
@RestController
|
|
||||||
public class ModelController {
|
|
||||||
@PostMapping(value = "/models")
|
|
||||||
public Model test2(Model model) {
|
|
||||||
int id = model.getId();
|
|
||||||
model.setId(id + 1000000);
|
|
||||||
return model;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,6 +0,0 @@
|
||||||
server:
|
|
||||||
port: 3000
|
|
||||||
spring:
|
|
||||||
application:
|
|
||||||
name: Server
|
|
||||||
|
|
||||||
69
pom.xml
69
pom.xml
|
|
@ -13,5 +13,74 @@
|
||||||
<version>0.0.1-SNAPSHOT</version>
|
<version>0.0.1-SNAPSHOT</version>
|
||||||
<name>DemoKafkaStream</name>
|
<name>DemoKafkaStream</name>
|
||||||
<description>DemoKafkaStream</description>
|
<description>DemoKafkaStream</description>
|
||||||
|
<properties>
|
||||||
|
<java.version>22</java.version>
|
||||||
|
<spring-cloud.version>2023.0.1</spring-cloud.version>
|
||||||
|
</properties>
|
||||||
|
<dependencyManagement>
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.cloud</groupId>
|
||||||
|
<artifactId>spring-cloud-dependencies</artifactId>
|
||||||
|
<version>${spring-cloud.version}</version>
|
||||||
|
<type>pom</type>
|
||||||
|
<scope>import</scope>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
</dependencyManagement>
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter</artifactId>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.cloud</groupId>
|
||||||
|
<artifactId>spring-cloud-stream</artifactId>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.cloud</groupId>
|
||||||
|
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.projectlombok</groupId>
|
||||||
|
<artifactId>lombok</artifactId>
|
||||||
|
<optional>true</optional>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-test</artifactId>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.cloud</groupId>
|
||||||
|
<artifactId>spring-cloud-stream-test-binder</artifactId>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-web</artifactId>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.cloud</groupId>
|
||||||
|
<artifactId>spring-cloud-function-web</artifactId>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
|
||||||
|
<build>
|
||||||
|
<plugins>
|
||||||
|
<plugin>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||||
|
<configuration>
|
||||||
|
<excludes>
|
||||||
|
<exclude>
|
||||||
|
<groupId>org.projectlombok</groupId>
|
||||||
|
<artifactId>lombok</artifactId>
|
||||||
|
</exclude>
|
||||||
|
</excludes>
|
||||||
|
</configuration>
|
||||||
|
</plugin>
|
||||||
|
</plugins>
|
||||||
|
</build>
|
||||||
|
|
||||||
</project>
|
</project>
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,108 @@
|
||||||
|
package ru.ldeloff.demokafkastream;
|
||||||
|
|
||||||
|
|
||||||
|
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.annotation.Bean;
|
||||||
|
import org.springframework.kafka.support.KafkaHeaders;
|
||||||
|
import org.springframework.messaging.Message;
|
||||||
|
import org.springframework.messaging.support.MessageBuilder;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
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 */
|
||||||
|
// Консьюмер
|
||||||
|
@Bean
|
||||||
|
public Consumer<Model> consumerAuto1() {
|
||||||
|
return message -> {
|
||||||
|
// System.out.println("Consume auto 1: " + message);
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// Продюсер (сам периодически отправляет сообщения)
|
||||||
|
@Bean
|
||||||
|
public Supplier<Message<Model>> producerAuto1() {
|
||||||
|
return () -> {
|
||||||
|
Model message = new Model(++i);
|
||||||
|
// System.out.println("Produce auto 1:" + message);
|
||||||
|
return MessageBuilder.withPayload(message)
|
||||||
|
.setHeader(KafkaHeaders.KEY, String.valueOf(message.getId()))
|
||||||
|
.build();
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// Вручную отправим сообщение
|
||||||
|
@GetMapping(value = "/test1")
|
||||||
|
public String test1() {
|
||||||
|
Model message = new Model(++i);
|
||||||
|
System.out.println("Produce manual (kafka 1): " + message);
|
||||||
|
streamBridge.send("producerManual1-out-0",
|
||||||
|
MessageBuilder.withPayload(message)
|
||||||
|
.setHeader(KafkaHeaders.KEY, String.valueOf(message.getId()))
|
||||||
|
.build());
|
||||||
|
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);
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
package ru.ldeloff.server.model;
|
package ru.ldeloff.demokafkastream;
|
||||||
|
|
||||||
import lombok.Getter;
|
import lombok.Getter;
|
||||||
import lombok.NoArgsConstructor;
|
import lombok.NoArgsConstructor;
|
||||||
|
|
@ -1,12 +1,10 @@
|
||||||
spring:
|
spring:
|
||||||
application:
|
application:
|
||||||
name: DemoKafkaStream
|
name: DemoKafkaStream
|
||||||
main:
|
|
||||||
allow-bean-definition-overriding: true
|
|
||||||
|
|
||||||
cloud:
|
cloud:
|
||||||
function:
|
function:
|
||||||
definition: producerAuto1;consumerAuto1;producerManual1;consumerManual1;producerAuto2;consumerAuto2;producerManual2;consumerManual2;producerAutoTestCircuitBreaker1;consumerAutoTestCircuitBreaker1
|
definition: producerAuto1;consumerAuto1;producerManual1;consumerManual1;producerAuto2;consumerAuto2;producerManual2;consumerManual2
|
||||||
stream:
|
stream:
|
||||||
bindings:
|
bindings:
|
||||||
producerAuto1-out-0:
|
producerAuto1-out-0:
|
||||||
|
|
@ -23,14 +21,6 @@ spring:
|
||||||
binder: kafka-1
|
binder: kafka-1
|
||||||
destination: kafka_1_topic_manual
|
destination: kafka_1_topic_manual
|
||||||
|
|
||||||
producerAutoTestCircuitBreaker1-out-0:
|
|
||||||
binder: kafka-1
|
|
||||||
destination: kafka_1_topic_auto_CB
|
|
||||||
consumerAutoTestCircuitBreaker1-in-0:
|
|
||||||
binder: kafka-1
|
|
||||||
destination: kafka_1_topic_auto_CB
|
|
||||||
group: consumerAutoTestCircuitBreaker
|
|
||||||
|
|
||||||
producerAuto2-out-0:
|
producerAuto2-out-0:
|
||||||
binder: kafka-2
|
binder: kafka-2
|
||||||
destination: kafka_2_topic_auto
|
destination: kafka_2_topic_auto
|
||||||
|
|
@ -69,50 +59,3 @@ spring:
|
||||||
binder:
|
binder:
|
||||||
brokers:
|
brokers:
|
||||||
- 192.168.10.101:39092
|
- 192.168.10.101:39092
|
||||||
|
|
||||||
#Resilience4j circuit breaker config
|
|
||||||
resilience4j.circuitbreaker:
|
|
||||||
instances:
|
|
||||||
consumerCircuitBreaker:
|
|
||||||
slidingWindowType: COUNT_BASED
|
|
||||||
minimumNumberOfCalls: 1
|
|
||||||
permittedNumberOfCallsInHalfOpenState: 10
|
|
||||||
failureRateThreshold: 50
|
|
||||||
automaticTransitionFromOpenToHalfOpenEnabled: true
|
|
||||||
slidingWindowSize: 10
|
|
||||||
waitDurationInOpenState: 10s
|
|
||||||
registerHealthIndicator: true
|
|
||||||
recordExceptions: org.springframework.web.reactive.function.client.WebClientRequestException
|
|
||||||
ignoreExceptions:
|
|
||||||
|
|
||||||
#Resilience4j retry config
|
|
||||||
resilience4j.retry:
|
|
||||||
instances:
|
|
||||||
consumerCircuitBreaker:
|
|
||||||
maxAttempts: 3
|
|
||||||
waitDuration: 500
|
|
||||||
retryExceptions: org.springframework.web.reactive.function.client.WebClientRequestException
|
|
||||||
ignoreExceptions:
|
|
||||||
metrics:
|
|
||||||
enabled: true
|
|
||||||
legacy:
|
|
||||||
enabled: true
|
|
||||||
|
|
||||||
management:
|
|
||||||
endpoints:
|
|
||||||
web:
|
|
||||||
exposure:
|
|
||||||
include: bindings,health
|
|
||||||
|
|
||||||
health:
|
|
||||||
binders:
|
|
||||||
enabled: true
|
|
||||||
circuitbreakers:
|
|
||||||
enabled: true
|
|
||||||
ratelimiters:
|
|
||||||
enabled:true
|
|
||||||
|
|
||||||
endpoint:
|
|
||||||
health:
|
|
||||||
show-details: ALWAYS
|
|
||||||
|
|
||||||
|
|
@ -1,10 +1,10 @@
|
||||||
package ru.ldeloff.server;
|
package ru.ldeloff.demokafkastream;
|
||||||
|
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import org.springframework.boot.test.context.SpringBootTest;
|
import org.springframework.boot.test.context.SpringBootTest;
|
||||||
|
|
||||||
@SpringBootTest
|
@SpringBootTest
|
||||||
class ServerApplicationTests {
|
class DemoKafkaStreamApplicationTests {
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
void contextLoads() {
|
void contextLoads() {
|
||||||
Loading…
Reference in New Issue