Resilience4j @CircuitBreaker Anotasyonu applicaiton.properties

Örnek

Şöyle yaparız

resilience4j.circuitbreaker.instances.default.failure-rate-threshold=60

Örnek

Şöyle yaparız

management:
  health:
    circuitbreakers:
      enabled: true
  endpoints:
    web:
      exposure:
        include: health
  endpoint:
    health:
      show-details: always
resilience4j:
  circuitbreaker:
    instances:
      callerService:
        registerHealthIndicator: true
        eventConsumerBufferSize: 10
        failureRateThreshold: 10
        minimumNumberOfCalls: 5
        automaticTransitionFromOpenToHalfOpenEnabled: true
        waitDurationInOpenState: 5s
        permittedNumberOfCallsInHalfOpenState: 3
        slidingWindowSize: 10
        slidingWindowType: COUNT_BASED

Açıklaması şöyle

- registerHealthIndicator: true: This line is telling the service to register a health indicator, which can be used to monitor the overall health of the service.

- eventConsumerBufferSize: 10: This is the size of the buffer that will store events from the Circuit Breaker.

- failureRateThreshold: 10: This is the threshold for the failure rate of calls. If the failure rate of calls exceeds this threshold, the Circuit Breaker will trip.

- minimumNumberOfCalls: 5: This is the minimum number of calls that the Circuit Breaker will monitor before it starts tripping.

- automaticTransitionFromOpenToHalfOpenEnabled: true: This line enables automatic transition from the open state to the half-open state after the wait duration has elapsed.

- waitDurationInOpenState: 5s: This is the duration that the Circuit Breaker will wait in the open state before attempting to automatically transition to the half-open state.

- permittedNumberOfCallsInHalfOpenState: 3: This is the maximum number of calls that the Circuit Breaker will allow in the half-open state.

- slidingWindowSize: 10: This is the size of the sliding window that the Circuit Breaker will use to monitor calls.

- slidingWindowType: COUNT_BASED: This line specifies the type of sliding window that the Circuit Breaker will use, which in this case is a count-based sliding window. There are other options for slidingWindowType like TIME_BASED, TIME_BASED_RESET etc which can be used for specific use cases.

Örnek

application.properties dosyası şöyledir

resilience4j:
    circuitbreaker:
        instances:
            mockService:
                slidingWindowSize: 3
                slidingWindowType: COUNT_BASED
                #waitDurationInOpenState: 5
                waitInterval: 10000
                failureRateThreshold: 50
                permittedNumberOfCallsInHalfOpenState: 5
                registerHealthIndicator: true
                #register-health-indicator: true
                allowHealthIndicatorToFail: true
        configs:
            default:
                registerHealthIndicator: true

SpringKafka Request Reply

Giriş

Şeklen şöyle

Kafka normalde asenkron çalışır. Ancak Request Reply için SpringBoot destek veriyor. Açıklaması şöyle.

How Spring solves the problem is by automatically setting a correlation ID in the producer record which is returned as-is by the @SendTo annotation at the consumer end.

Request Reply için şunları yapmak gerekir.

1. ReplyingKafkaTemplate direkt kullanılır 

2. Consumer tarafta @KafkaListener ve @SendTo birlikte kullanılır. @SendTo metodun döndürdüğü nesneyi belirtilen topic'e gönderir.

Producer tarafında şöyle yaparız. repliesContainer() ile ReplyingKafkaTemplate nesnesine cevapların hangi topic'ten okunacağını belirtiriz.

@Bean
public ProducerFactory<String, String> producerFactory() {
   ...
}

 
@Bean
public ReplyingKafkaTemplate<String, String, String> replyKafkaTemplate(
  ProducerFactory<String, String> pf, 
  ConcurrentMessageListenerContainer<String, String> repliesContainer) {
  return new ReplyingKafkaTemplate<>(pf, repliesContainer);
}
 
@Bean
public ConcurrentMessageListenerContainer<String, String> repliesContainer(
  ConsumerFactory<String, String> cf) {
  ContainerProperties containerProperties = new ContainerProperties("replies");
  return new ConcurrentMessageListenerContainer<>(cf, containerProperties);
}

Producer tarafındaki Servisi şöyle yaparız

@Service
public class RequestReplyService {
 
  private final ReplyingKafkaTemplate<String, Request, Reply> kafkaTemplate;
 
  public Reply process(Request request) throws ExecutionException, InterruptedException {
    ProducerRecord<String, Request> record = 
      new ProducerRecord<>("request-topic", request);
    RequestReplyFuture<String, Request, Reply> future = kafkaTemplate
      .sendAndReceive(record);

    // confirm if producer produced successfully
    SendResult<String, Model> sendResult = future.getSendFuture().get();

    // get consumer record
    ConsumerRecord<String, Reply> consumerRecord = future.get();
    // return consumer value
    return consumerRecord.value();
  }
}

Consumer tarafındaki servisi şöyle yaparız

@Service
public class RequestReplyService {
 
  @KafkaListener(topics = "request-topic")
  @SendTo
  public Reply process(Request request) {
    String message = "Received request: " + request.getMessage();
    return new Reply(message);
  }
}

SpringData ListCrudRepository Arayüzü

Giriş
Şu satırı dahil ederiz.

import org.springframework.data.repository.ListCrudRepository;

Kodu şöyle

public interface ListCrudRepository<T, ID> extends CrudRepository<T, ID> {
  List<T> findAll()
  List<T> findAllById(Iterable<ID> ids)
  <S extends T> List<S> saveAll(Iterable<S> entities);
}  

saveAll metodu

CrudRepository arayüzünün saveAll() metodu Iterable dönerken, bu metod List dönüyor

SpringCache CachingConfigurerSupport Sınıfı

Giriş
Şu satırı dahil ederiz.

import org.springframework.cache.annotation.CachingConfigurerSupport;

Birden fazla Cache kullanacaksan bir tanesini @Primary olarak işaretlemeye gerek kalmaz. cacheManager() tarafından döndürülen nesne @Primary kabul edilir. Şöyle yaparız

@Configuration
@EnableCaching
public class MultipleCacheManagerConfig extends CachingConfigurerSupport {

  @Bean
  public CacheManager cacheManager() {
    CaffeineCacheManager cacheManager = new CaffeineCacheManager("customers", "orders");
    cacheManager.setCaffeine(Caffeine.newBuilder()
      .initialCapacity(200)
      .maximumSize(500)
      .weakKeys()
      .recordStats());
    return cacheManager;
  }

  @Bean
  public CacheManager alternateCacheManager() {
    return new ConcurrentMapCacheManager("customerOrders", "orderprice");
  }
}

SpringPulsar

Bir örnek burada

SpringJMS ActiveMQ Kullanımı

Maven

Şöyle yaparız

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

Producer

Örnek

Elimizde şöyle bir kod olsun

import org.apache.activemq.ActiveMQConnectionFactory;
@Configuration
@EnableJms
public class MQConfigHelper {
  final String BROKER_URL = "tcp://localhost:61616";
  //Enter your ActiveMQ password and username
  final String BROKER_USERNAME = "";
  final String BROKER_PASSWORD = "";

  @Bean
  public ActiveMQConnectionFactory activemqConnectionFactory() {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
    factory.setTrustAllPackages(true);
    factory.setBrokerURL(BROKER_URL);
    factory.setPassword(BROKER_USERNAME);
    factory.setUserName(BROKER_PASSWORD);
    return factory;
  }
  @Bean
  public MessageConverter messageConverter() {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setTargetType(MessageType.TEXT);
    return converter;
  }
  @Bean
  public JmsTemplate templateConfig() {
    JmsTemplate jmsTemplate = new JmsTemplate();
    jmsTemplate.setConnectionFactory(activemqConnectionFactory());
    jmsTemplate.setMessageConverter(messageConverter());
    jmsTemplate.setDeliveryPersistent(true);
    return jmsTemplate;
  }
}

Mesaj gönderen taraf şöyledir

public class CarOrder {...}

public class CustomerOrder {...}

import org.springframework.jms.core.JmsTemplate;
import javax.jms.TextMessage;

@Component
public class MessageSender {
  String queueName = "carQueue";
  @Autowired
  JmsTemplate jmsTemplate;
  public void sendToQueue() {
   try {
     CarOrder carOrder = new CarOrder("Mercedes GLE", 7, 74);
     processMessageForSending(carOrder);
    } catch (Exception e) {
      ...
    }
  }
  public void processMessageForSending(Object message) 
  throws JsonProcessingException {
    String jsonObj = new ObjectMapper()
      .writer()
      .withDefaultPrettyPrinter()
      .writeValueAsString(message);
      jmsTemplate.send(queueName, messageCreator -> {
        TextMessage txtMessage = messageCreator.createTextMessage();
        txtMessage.setText(jsonObj);
        return txtMessage;
      });
  }
}

Mesaj dinleyen taraf şöyledir

import org.springframework.jms.annotation.JmsListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.TextMessage;
@Component
public class MessageConsumer {
  @JmsListener(destination = "carQueue")
  public void queueMessageReceiver(Message receivedMessage) {
    try {
      processMessageReceived(receivedMessage);
    } catch (Exception e) {
       ...
    }
  }
  public void processMessageReceived(Message messageArg) throws JMSException {
    if (messageArg instanceof TextMessage) {
      TextMessage textMessage = (TextMessage) messageArg;
      String processedMessage = textMessage.getText();
      ... 
     }
  }
}

SpringContext InputStreamResource Sınıfı - MediaType.APPLICATION_OCTET_STREAM İndirme

Giriş

Şu satırı dahil ederiz 

import org.springframework.core.io.InputStreamResource;

Açıklaması şöyle. Yani Resource arayüzünü gerçekleştirir.

It is a subclass of AbstractResource, which means that it implements the Resource interface. The Resource interface provides a number of methods for working with resources, such as getting the resource's URL, filename, and content type.

Açıklaması şöyle. Java'daki InputStream ve Spring'deki Resource arayüzü arasındaki köprüdür.

The purpose of InputStreamResource is to provide a bridge between the Spring Framework's resource handling capabilities and the input stream-based data sources. It allows you to treat an input stream as a resource that can be easily manipulated and managed within the Spring Framework's ecosystem.

constructor - InputStream

Örnek - Video İndirme

Şöyle yaparız. Burada bir ByteBufferBackedInputStream aslında bir InputStream, tek farkı içine bir ByteBuffer alması. Bu nesnesi kullanarak bir InputStreamResource yaratıyoruz ve ResponseEntity.body() metodu ile gönderiyoruz. Cevaptaki Content type APPLICATION_OCTET_STREAM yani ham byte verisi.

@GetMapping("{name}")
ResponseEntity<Resource> getMovieByName(@PathVariable String name) {
  return storage.pull(name)
    .map(Movie::movieByteBuffer)
    .map(ByteBuffer::slice)
    .map(ByteBufferBackedInputStream::new)
    .map(InputStreamResource::new)
    .<ResponseEntity<Resource>>map(resource ->
      ResponseEntity.ok()
        .contentType(MediaType.APPLICATION_OCTET_STREAM)
        .body(resource))
      .orElseGet(() -> ResponseEntity.notFound().build());
}

Örnek - Dosya İndirme

Elimizde veri tabanından sorgu yapıp dosyaya yazan şöyle bir kod olsun

@Override
public File getWholeFile(Long segmentId) throws IOException {
  ...
  File file = ...
  try (PrintWriter printWriter = new PrintWriter(new FileWriter(file))) {
    printWriter.println(HEADER);
    writeRowsToFile(segments, printWriter);
  }
  return file;
}

Şöyle yaparız. Sorgu sonucu olan dosyayı tek parça halinde indirir

@GetMapping("/{id}")
public ResponseEntity<Resource> getFileWhole(@PathVariable("id") Long segmentId) 
throws IOException {
  File file = fileService.getWholeFile(segmentId);
  InputStreamResource resource = new InputStreamResource(new FileInputStream(file));
  HttpHeaders headers = formHeaders(segmentId);
  return ResponseEntity.ok()
    .headers(headers)
    .body(resource);
}

formHeaders şöyle

private HttpHeaders formHeaders(Long segmentId) {
  HttpHeaders headers = new HttpHeaders();
  headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_OCTET_STREAM_VALUE);
  headers.add(HttpHeaders.ACCESS_CONTROL_EXPOSE_HEADERS, CONTENT_DISPOSITION);
  headers.add(CONTENT_DISPOSITION, String.format("attachment; filename=segment_%d.csv",
    segmentId));
  return headers;
}