SpringCloud Stream Kullanımı - Abstraction Layer On Top Of the Messaging Middleware

Giriş

SpringCloud Stream (SCS) alt katmanda kullanılan kuyruk teknolojisini soyutlar. Açıklaması şöyle

SCS framework provides an abstraction layer for event-driven systems to communicate over asynchronous messages. It hides the underlying middlewares from the application so it can use unified programming model to implement services.

SCS Reactive programlama kullanır. Açıklaması şöyle

SCS supports the reactive programming model. It uses Reactor library which allows us to write asynchronous, non blocking, declarative code.

Hangi Messaging Middleware Desteklenir

Açıklaması şöyle

Spring maintains the binder implementations for RabbitMQ, Kafka, Kafka Streams and Amazon Kinesis while Google PubSub, Azure EventHub, Apache RocketMQ and Solace PubSub binders are maintained by the corresponding organisations.

Maven

Kafka için şu satırı dahil ederiz

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

Bileşenler

Şeklen şöyle

Default Binder Tanımlanır

Eğer projede birden fazla binder varsa ve her bir binding için binder tanımlı değilse, kullanılacak binder tanımlanır. Ayrıca custom serializer/deserializer kullanmak istiyorsak yine binder tanımlamak gerekir.

Örnek - custom serializer/deserializer

Şöyle yaparız

spring:
  cloud:
    stream:
      default-binder: kafka
      kafka:
        binder:
          auto-create-topics: true
          consumer-properties:
            key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value.deserializer: org.springframework.kafka.support.serializer
                                .JsonDeserializer
          producer-properties:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: org.springframework.kafka.support.serializer.JsonSerializer

Örnek - custom serializer/deserializer

Şöyle yaparız

spring:
  cloud:
    function:
      definition: consumer;producer
    stream:
      kafka:
        bindings:
          producer-out-0:
            producer:
              configuration:
                value.serializer: com.foo.MessageSerializer
          consumer-in-0:
            consumer:
              configuration:
                value.deserializer: com.foo.MessageDeSerializer
        binder:
          brokers: localhost:9092
            
      bindings:
        producer-out-0:
            destination : first-topic
            producer:
                useNativeEncoding: true # Enables using the custom serializer
        consumer-in-0:
            destination : first-topic
            consumer:
              use-native-decoding: true # Enables using the custom deserializer

Şöyle yaparız

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

public class MessageSerializer implements Serializer<Message> {

  private final ObjectMapper objectMapper = new ObjectMapper();

  @Override
  public byte[] serialize(String topic, Message data) {
    try {
      return objectMapper.writeValueAsBytes(data);
    } catch (JsonProcessingException e) {
      throw new SerializationException(e);
    }
  }
}

Şöyle yaparız

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

public class MessageDeSerializer implements Deserializer<Message> {

  private final ObjectMapper objectMapper = new ObjectMapper();

  @Override
  public Message deserialize(String topic, byte[] data) {
    try {
      return objectMapper.readValue(new String(data), Message.class);
    } catch (IOException e) {
      throw new SerializationException(e);
    }
  }
}

Eski Anotasyonlar

SpringCloud Stream ilk çıktığında anotasyon kullanıyordu. Daha sonra bu anotasyonlar bırakıldı ve "Functional Style" kullanılmaya başlandı. Deprecate edilen anotasyolar ve sınıflar şöyle

- @EnableBinding 

- @Input

- @Output 

- @StreamListener

- @StreamListener + @SendTo : Processor'da kullanılır

Açıklaması şöyle

If you are used to building applications based on Spring framework, then you should be familiar with the use of annotation. You can simply tag features to classes and methods by adding annotations. For example, annotation @Streamlistener is to specify the method to be triggered by incoming messages. While annotation can still be used for development for a while, such approach is already deprecated and it is no longer recommended since Spring Cloud Stream version 3.x.

Function Definition Tanımlanır

application.properties dosyasında spring.cloud.stream.function.definition değeri altında Producer, Consumer, Processor işlevi gören bean'ler tanımlanır.

StreamBridge yazısına da bakabilirsiniz

Producer Bean

Producer Bean yazısına taşıdım

Consumer Bean

Consumer Bean yazısına taşıdım

Processor Bean

Processor Bean yazısına taşıdım

Flux.generate metodu

Giriş

Açıklaması şöyle

- Accepts a Consumer<SynchronousSink<T>>

- Consumer is invoked again and again based on the downstream demand

- Consumer can emit only one element at the max with an optional complete/error signal.

- Publisher produces elements based on the downstream demand

- We can get the reference of SynchronousSink. But it might not be really useful as we could emit only one element

generate metodu - SynchronousSink

Örnek

Şöyle yaparız

AtomicInteger atomicInteger = new AtomicInteger();

//Flux generate sequence
Flux<Integer> integerFlux = Flux.generate((SynchronousSink<Integer> synchronousSink) -> {
   System.out.println("Flux generate");
   synchronousSink.next(atomicInteger.getAndIncrement());
});

//observer
integerFlux.delayElements(Duration.ofMillis(50))
        .subscribe(i -> System.out.println("First consumed ::" + i));

Açıklaması şöyle. SynchronousSink önce belli bir sayıda nesne ile doldurulur, daha sonra tüketen taraf istedikçe yeni nesneler eklenir.

Generate method emits elements based on the demand from the downstream. It generates 32 elements first and buffers it. As and when downstream starts processing elements and when the buffer size drops the below threshold, it emits few more elements. This process is repeated again and again. At one point, if the observer stops processing elements, Flux.generate will also stop emitting elements. So this generate method is aware of downstream observers processing speed.

Örnek - Birden Fazla Nesne Yazmaya Çalışmak

Bu yapılamaz. Elimizde şöyle bir kod olsun

AtomicInteger atomicInteger = new AtomicInteger();
Flux<Integer> integerFlux = Flux.generate((SynchronousSink<Integer> synchronousSink) -> {
   System.out.println("Flux generate");
    synchronousSink.next(atomicInteger.getAndIncrement());
    synchronousSink.next(atomicInteger.getAndIncrement());
});

Çıktı olarak şu hatayı alırız

Flux generate

[ERROR] (main) Operator called default onErrorDropped - reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: More than one call to onNext

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: More than one call to onNext

Caused by: java.lang.IllegalStateException: More than one call to onNext

Flux.create metodu - Harici Kaynaklardan Gelen Asenkron Veri İçindir

Giriş

Şu satırı dahil ederizz

import reactor.core.publishe.Flux;

Cold publisher yaratır.  Harici kaynaklardan asenkron olarak gelen veriyi bir FluxSink'e yazarak tüketilmesini sağlar. Açıklaması şöyle. Flux.create() multi-threaded şekilde çalışır. Eğer harici kaynak tek ise yani  tek bir thread tarafından veri dolduruluyorsa Flux.push() kullanılabilir.

- Accepts a Consumer<FluxSink<T>>

- Consumer is invoked only once per subscriber

- Consumer can emit 0..N elements immediately

- Publisher is not aware of downstream state. So we need to provide Overflow strategy as an additional parameter

- We can get the reference of FluxSink using which we could keep on emitting elements as and when required using multiple threads.

Bu metod tarafından yaratılan FluxSink, tüketen taraf yavaş kalırsa bile umursamaz.  Açıklaması şöyle

The downstream will determine how many elements (elements == next signals) it wants and if he can't keep up, those elements which are already emitted will be removed/buffered in some strategy (by default they will be buffered until the downstream will ask for more).

create metodu - FluxSink

Bu metoda geçilen lamda parametre olarak bir tane FluxSink nesnesi alır. Bu nesnenin next() metodu çağrılarak doldurma işlemi gerçekleştirilir.

Örnek

Şöyle yaparızz

Flux<String> articlesFlux = Flux.create((FluxSink<String> sink) -> {
  /* get all the latest article from a server and emit them one by one to downstream. */
  List<String> articals = getArticalsFromServer();
  articals.forEach(sink::next);
});

Örnek

Açıklaması şöyle

Flux.create produces a SerializedSink that is safe to use from multiple threads for next calls

Şöyle yaparız. Burada FluxSink nesnesine hemen bir şey yazılmıyor. Daha sonra ara ara yazılıyor

Map<String, FluxSink<Item>> sinks = new ConcurrentHashMap<>();

// Store a new sink for the given ID
public void start(String id) {
  Flux.create(sink -> sinks.put(id, sink));
}

// Called from different threads
public void publish(String id, Item item) {
  sinks.get(id).next(item); //<----------- Is this safe??
}

Örnek

Elimizde  şöyle bir kod olsun. Burada FluxSink nesnesi en baştan hemen dolduruluyor. Her abone veriyi en baştan tekrar alır

Flux<Integer> integerFlux = Flux.create((FluxSink<Integer> fluxSink) -> {
    IntStream.range(0, 5)
            .peek(i -> System.out.println("going to emit - " + i))
            .forEach(fluxSink::next);
});

//First observer. takes 1 ms to process each element
integerFlux.delayElements(Duration.ofMillis(1)).subscribe(
  i -> System.out.println("First :: " + i));

//Second observer. takes 2 ms to process each element
integerFlux.delayElements(Duration.ofMillis(2)).subscribe(
  i -> System.out.println("Second:: " + i));

Çıktı olarak şunu alırız

going to emit - 0
going to emit - 1
going to emit - 2
going to emit - 3
going to emit - 4
going to emit - 0
going to emit - 1
going to emit - 2
going to emit - 3
going to emit - 4
First :: 0
Second:: 0
First :: 1
First :: 2
Second:: 1
First :: 3
First :: 4
Second:: 2
Second:: 3
Second:: 4

Açıklaması şöyle

What we can understand from the above output is,

- Each observer gets its own FluxSink instance which is expected as we create a Cold publisher.

- Create method does not wait for the observer to process the elements. It emits the elements even before observers start processing the elements.

create metodu - FluxSink + OverFlowStrategy

Açıklaması şöyle

BUFFER : Buffer all signals if the downstream can't keep up

DROP : Drop the incoming signal if the downstream is not ready to receive it

ERROR : Signal an IllegalStateException when the downstream can't keep up

IGNORE : Completely ignore the downstream backpressure request

LATEST : Downstream will get only the lates signals from upstream

Örnek

Şöyle yaparız. Burada BUFFER, DROP, ERROR, IGNORE, LATEST seçenekleri kullanılabilir.

Flux<Integer> integerFlux = Flux.create((FluxSink<Integer> fluxSink) -> {
            IntStream.range(0, 5)
                    .peek(i -> System.out.println("going to emit - " + i))
                    .forEach(fluxSink::next);
        }, FluxSink.OverflowStrategy.DROP);

SpringCloud Feign Retrier Arayüzü

Giriş

Bir başka örnek burada

Örnek

Elimizde şöyle bir kod olsun.

class Custom implements Retryer {

  private final int maxAttempts;
  private final long backoff;
  int attempt;

  public Custom() {
    this(2000, 3);
  }

  public Custom(long backoff, int maxAttempts) {
    this.backoff = backoff;
    this.maxAttempts = maxAttempts;
    this.attempt = 1;
  }

  public void continueOrPropagate(RetryableException e) {
    if (attempt++ >= maxAttempts) {
      throw e;
    }

    try {
      Thread.sleep(backoff);
    } catch (InterruptedException ignored) {
      Thread.currentThread().interrupt();
    }
  }

  @Override
  public Retryer clone() {
    return new Custom(backoff, maxAttempts);
  }
}

Şöyle yaparız

@Configuration
public class FeignClientConfig {

  @Bean
  public Retryer retryer() {
    return new Custom();
  }
}

SpringData R2DBC DatabaseClient Sınfı

Giriş

Şu satırı dahil ederiz. Normalde R2dbcRepository sınıfını kullanmak lazım. Eğer custom bir iş yapılacaksa bu sınıf kullanılabilir.

import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.relational.core.query.Update;
import org.springframework.stereotype.Repository;

delete metodu

Örnek

Şöyle yaparız

public Mono<Boolean> deleteBook(UUID id){
  return databaseClient.delete()
    .from(Book.class).matching(Criteria.where("id").is(id)).then().thenReturn(true);
}

insert metodu

Örnek

Şöyle yaparız

public Mono<UUID> createBook(Book book){
  UUID bookId = UUID.randomUUID();
  book.setId(bookId);

  return databaseClient.insert().into(Book.class).using(book).then()
    .thenReturn(bookId);
}

select metodu

Örnek

Şöyle yaparız

@Repository
public class BookRepository {

  @Autowired
  private DatabaseClient databaseClient;

  public Mono<Book> getBook(UUID id){
    return databaseClient.select()
      .from(Book.class)
      .matching(Criteria.where("id").is(id))
      .fetch()
      .one();
  }

  public Flux<Book> getBooks(){
    return databaseClient.select().from(Book.class).fetch().all();
  }
}

sql metodu

Örnek

Şöyle yaparız

@Repository
@RequiredArgsConstructor
public class AuthorCustomRepository {

  private final DatabaseClient databaseClient;

  public Flux<Author> findAll() {
    return databaseClient.sql("SELECT  "
      + "book.id as bookId, book.title as bookName, book.date_of_parution "
      + "as dateOfParution, "
      + "author.id as authorId, author.name as authorName "
      + "FROM author "
      + "JOIN book on author.id = book.author "
      + "ORDER BY authorId")
    .fetch()
    .all()
    .bufferUntilChanged( result -> result.get("authorId"))
    .map(list -> {
      AuthorBuilder author = Author.builder();
      author.name(String.valueOf(list.get(0).get("authorName")));
      author.id((UUID) list.get(0).get("authorId"));
			
      author.books(
	list.stream()
	.map(map -> {
	  return Book.builder()
	    .title((String) map.get("bookName"))
	    .id((UUID) map.get( "bookId"))
	    .dateOfParution((LocalDate) map.get("dateOfParution"))
	    .build();
	})
        .collect(Collectors.toList())
      );
      return author.build();
    });
  }
}

update metodu

Örnek

Şöyle yaparız

public Mono<String> updateBook(UUID ID,Book book){
  return databaseClient.update().table("books")
    .using(Update.update("name",book.getName())).matching(Criteria.where("id").is(ID))
    .then().thenReturn("Success");
}

SpringData R2DBC AbstractR2dbcConfiguration Sınıfı

getCustomConverters metodu

Örnek

Şöyle yaparız. Converter sınıfları @ReadingConverter veya @WritingConverter olarak işaretlidir

import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;

import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
import io.r2dbc.spi.ConnectionFactory;

@Configuration
@EnableR2dbcRepositories
public class PostgresConfig extends AbstractR2dbcConfiguration {

  @Override
  @Bean
  public ConnectionFactory connectionFactory() {
    return new PostgresqlConnectionFactory(
      PostgresqlConnectionConfiguration.builder()
        .host("localhost")
        .port(5432)
	.username("postgres")
	.password("admin")
	.database("reactive_examples_db")
	.build());
  }

  @Override
  protected List<Object> getCustomConverters() {
    List<Object> converterList = new ArrayList<>();
    converterList.add(new PersonReadConverter());
    converterList.add(new PersonWriteConverter());
    converterList.add(new BookReadConverter());
    return converterList;
  }
}

SpringBoot Actuator - Startup Endpoint

Giriş

Bu endpoint ile uygulamayı başlatırken hangi bean'lerin daha çok vakit aldığını görmek mümkün. Web kullanımını etkinleştirmek için şöyle yaparız

management.endpoints.web.exposure.include=startup

Eğer yml kullanılyorsak  şöyle yaparız

management:
  endpoints:
    web:
      exposure:
        include: 'startup'

ApplicationStartup arayüzünü gerçekleştiren iki tane bean var

1. BufferingApplicationStartup Sınıfı

Örnek

Şöyle yaparız

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.metrics.buffering.BufferingApplicationStartup;

@SpringBootApplication
public class Application {

  public static void main(String[] args) {
    SpringApplication application = new SpringApplication(Application.class);
    application.setApplicationStartup(new BufferingApplicationStartup(10000));
    application.run(args);
  }
}

Sonuçları almak için POST göndermek gerekir. Şöyle yaparız

curl --location --request POST 'http://localhost:8080/actuator/startup'

Çıktı olarak şunu alırız. Burada isimleri studentRepository ve studentsController olan iki tane bean için bilgiler görülebilir. 

{
  "startupStep": {
    "name": "spring.beans.instantiate",
    "id": 123,
    "parentId": 122,
    "tags": [
      {
        "key": "beanName",
        "value": "studentRepository"
      }
    ]
  },
  "startTime": "2021-01-26T06:38:09.234585991Z",
  "endTime": "2021-01-26T06:38:09.380445297Z",
  "duration": "PT0.145859306S"
},
{
  "startupStep": {
  "name": "spring.beans.instantiate",
  "id": 122,
  "parentId": 5,
  "tags": [
    {
      "key": "beanName",
      "value": "studentsController"
    }
  ]
  },
  "startTime": "2021-01-26T06:38:09.231829732Z",
  "endTime": "2021-01-26T06:38:09.382129262Z",
  "duration": "PT0.15029953S"
},

2. FlightRecordingApplicationStartup Sınıfı

Örnek

Şöyle yaparız

public static void main(String[] args) {
  SpringApplication application = new SpringApplication(Application.class);
  application.setApplicationStartup(new FlightRecorderApplicationStartup());
  application.run(args);
}

Uygulamayı başlatmak için şöyle yaparız.  myrecording.jfr dosyası oluşur. Bu dosyayı JDK Mission Control ile açabiliriz.

java -XX:+FlightRecorder \        
        -XX:StartFlightRecording=duration=5s,filename=myrecording.jfr \        
        -jar target/spring-boot-crud-jpa.jar