L_DelOff/DemoKafkaStream
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Разбил по модулям и добавил Circuit Breaker

Browse Source
This commit is contained in:
L_DelOff
2024-06-17 21:31:45 +03:00
parent c75b76e351
commit cca8187441
21 changed files with 597 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
KafkaStream/src/main/java/ru/ldeloff/kafkastream/KafkaStreamApplication.java Normal file
View File

@@ -0,0 +1,13 @@
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);
}
}

18
KafkaStream/src/main/java/ru/ldeloff/kafkastream/config/CircuitBreakerConfig.java Normal file
View File

@@ -0,0 +1,18 @@
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);
}
}

38
KafkaStream/src/main/java/ru/ldeloff/kafkastream/config/CircuitBreakerImpl.java Normal file
View File

@@ -0,0 +1,38 @@
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 CircuitBreaker circuitBreaker;
private final BindingsEndpoint bindingsEndpoint;
protected CircuitBreakerImpl(BindingsEndpoint bindingsEndpoint,
CircuitBreakerConfig circuitBreakerConfig) {
this.bindingsEndpoint = bindingsEndpoint;
this.circuitBreaker = circuitBreakerConfig.circuitBreaker("consumerCircuitBreaker");
circuitBreaker.getEventPublisher().onStateTransition(this::onStateChangeEvent);
}
private void onStateChangeEvent(CircuitBreakerOnStateTransitionEvent event) {
System.out.println("WARNING");
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;
}
}
}

56
KafkaStream/src/main/java/ru/ldeloff/kafkastream/config/KafkaConsumerConfig.java Normal file
View File

@@ -0,0 +1,56 @@
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);
};
}
}

68
KafkaStream/src/main/java/ru/ldeloff/kafkastream/config/KafkaProducerConfig.java Normal file
View File

@@ -0,0 +1,68 @@
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);
}
}

20
KafkaStream/src/main/java/ru/ldeloff/kafkastream/config/KafkaStreamConfig.java Normal file
View File

@@ -0,0 +1,20 @@
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));
}
}

28
KafkaStream/src/main/java/ru/ldeloff/kafkastream/config/WebConfig.java Normal file
View File

@@ -0,0 +1,28 @@
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();
}
}

41
KafkaStream/src/main/java/ru/ldeloff/kafkastream/controller/Controller.java Normal file
View File

@@ -0,0 +1,41 @@
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;
}
@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";
}
}

19
KafkaStream/src/main/java/ru/ldeloff/kafkastream/model/Model.java Normal file
View File

@@ -0,0 +1,19 @@
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();
}
}

27
KafkaStream/src/main/java/ru/ldeloff/kafkastream/service/ModelService.java Normal file
View File

@@ -0,0 +1,27 @@
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);
}
}

118
KafkaStream/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,118 @@
spring:
application:
name: DemoKafkaStream
main:
allow-bean-definition-overriding: true
cloud:
function:
definition: producerAuto1;consumerAuto1;producerManual1;consumerManual1;producerAuto2;consumerAuto2;producerManual2;consumerManual2;producerAutoTestCircuitBreaker1;consumerAutoTestCircuitBreaker1
stream:
bindings:
producerAuto1-out-0:
binder: kafka-1
destination: kafka_1_topic_auto
consumerAuto1-in-0:
binder: kafka-1
destination: kafka_1_topic_auto
producerManual1-out-0:
binder: kafka-1
destination: kafka_1_topic_manual
consumerManual1-in-0:
binder: kafka-1
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:
binder: kafka-2
destination: kafka_2_topic_auto
consumerAuto2-in-0:
binder: kafka-2
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:
type: kafka
environment:
spring:
cloud:
stream:
kafka:
binder:
producer-properties:
key.serializer: "org.apache.kafka.common.serialization.StringSerializer"
brokers:
- 192.168.10.101:29092
kafka-2:
type: kafka
environment:
spring:
cloud:
stream:
kafka:
binder:
brokers:
- 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

13
KafkaStream/src/test/java/ru/ldeloff/kafkastream/KafkaStreamApplicationTests.java Normal file
View File

@@ -0,0 +1,13 @@
package ru.ldeloff.kafkastream;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class KafkaStreamApplicationTests {
@Test
void contextLoads() {
}
}