14 Mart 2023 Salı

Log4j2 log4j2.xml ConsoleAppender Tanımlama - Console Tag

Giriş
Açıklaması şöyle
Log4j2 supports a number of System Properties that can be used to configure various items. For example, the log4j2.skipJansi system property can be used to configure if the ConsoleAppender will try to use a Jansi output stream on Windows.
Açıklaması şöyle
writes the data to System.out or System.err with the default begin the first one (a Java best practice when logging in containers)
Örnek
Şöyle yaparız
<Appenders>
  <Console name="Console" target="SYSTEM_OUT">
    <PatternLayout pattern="%d{HH:mm:ss.SSS} - %m %n"/>
  </Console>
</Appenders>
Örnek
Şöyle yaparız. Burada target System.out ama aslında normalde belirtmeye gerek yok. Varsayılan zaten bu
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN" monitorInterval="30" shutdownHook="disable">
   <Appenders>
      <Console name="Console" target="SYSTEM_OUT">
         <PatternLayout pattern="%-5p|%d{ISO8601}{GMT}|%X{token}|%c{1}|%m%ex%n" />
      </Console>
   </Appenders>
   ...
</Configuration>

Log4j2 log4j2.xml FileAppender Tanımlama

Örnek
Şöyle yaparız
<Configuration name="LogToConsole" target="SYSTEM_OUT">
  <Appenders>
    <File name="FileAppender" fileName="logs/app.log.json">
      <EcsLayout serviceName="my-app"/>
    </File>
  </Appenders>

  <Loggers>
    <Root>
      <AppenderRef ref="FileAppender" />
    </Root>
    <Logger name="none.rks.Main" level="debug" additivity="false">
      <AppenderRef ref="FileAppender" />
    </Logger>
  </Loggers>
</Configuration>

ForkJoinPool Work Stealing Nedir?

Giriş
Açıklaması şöyle
If a thread is getting overwhelmed and its internal queue fills up another thread instead of picking up task from the main queue can “steal” task from another threads internal queue.
Şeklen şöyle. B thread'i A thread'inin kuyruğundan iş çalıyor.
Resimdeki her threadin kuyruğunun ismi ForkJoinPool.WorkQueue. Açıklaması şöyle
ForkJoinPool is that it’s created with the following ruling points.Each thread has its own task queue. 

- Each task queue is a cicular array.
- Each thread uses push and pop to add or remove tasks to its own queue.
- ForkJoinPool uses the work-stealing algorithm to balance the workload on different threads. The pool maintains a global work queue that stores externally submitted tasks. Each worker thread will pop tasks from its own task queue. If there are no tasks in its own queue, it will try to randomly steal tasks from the shared work queues or other workers. If it fails to find tasks from both shared queues or other threads, it will go to sleep.
ForkJoinPool.WorkQueue Sınıfı
3 tane temel metod sunuyor. Açıklaması şöyle
push: used by worker thread to push task to the top of its own work queue
pop: used by worker thread to pop task from the top of its own work
poll: used by other thread to steal task from the bottom of the work queue of a different thread

8 Mart 2023 Çarşamba

MongoDB CreateCollectionOptions Sınıfı

Giriş
Şu satırı dahil ederiz
import com.mongodb.client.model.CreateCollectionOptions;
Örnek
Şöyle yaparız
CreateCollectionOptions options = new CreateCollectionOptions(); ValidationOptions validationOptions = new ValidationOptions(); validationOptions.validator(BsonDocument.parse( "{\n" + " $jsonSchema: {\n" + " bsonType: \"object\",\n" + " title: \"Person Object Validation\",\n" + " required: [ \"firstName\", \"lastName\", \"birthYear\" ],\n" + " properties: {" + " \"firstName\": { \"bsonType\": \"string\" }\n" + " \"lastName\": { \"bsonType\": \"string\" }\n" + " \"employed\": { \"bsonType\": \"bool\" }\n" + " }\n" + " }\n" + " }\n" )); options.validationOptions(validationOptions); MongoClient client = MongoClients.create(...); String databaseName = "testDatabase"; String collectionName = "people"; MongoDatabase testDatabase = client.getDatabase(databaseName); testDatabase.createCollection(collectionName, options);

6 Mart 2023 Pazartesi

Kafka Streams API KTable Sınıfı

Giriş
Şu satırı dahil ederiz
import org.apache.kafka.streams.kstream.KTable;
Açıklaması şöyle
A KTable is an abstraction of a changelog stream, where each data record represents an update. More precisely, the value in a data record is interpreted as an “UPDATE” of the last value for the same record key.

A record with a null as value represents a “DELETE” or tombstone for the record’s key.
Açıklaması şöyle
KTable uses the local state for the Kafka instance, so in this case if the one of the Kafka instance goes down, you will lose the data persisted to the local instance because a new Kafka instance is formed on new machine. 
join metodu
Örnek - KTable KTable
Şöyle yaparız
// Key=null,Value=10:00:00AM_GMT
KTable<String, String> table1 = builder.table("kafka-left-topic");

// Key=10:00:00AM, Value=task_1_completed
KTable<String, String> table2 = builder.table("kafka-right-topic");

KTable<String, String> joinedTable = table1.join(table2,
  (value1, value2) -> value1 + "," + value2
);

joinedTable.toStream().foreach((key, value) -> System.out.println(key + ": " + value));
Örnek - KTable + KeyValueMapper
Şöyle yaparız
KeyValueMapper<String, String, String> foreignKeyExtractor =
    (value1, value2) -> value1.split("_")[0];

KTable<String, String> joinedTable = table1.join(table2,foreignKeyExtractor,
  (value1, value2) -> value1 + "," + value2
);
Örnek - KStream + KTable
Şöyle yaparız
KStream<String, GenericRecord> ordersStream = ...
KTable<String, GenericRecord> customersTable = ..

// Create a foreign key extractor to extract the customer_id from the orders stream
KeyValueMapper<GenericRecord, GenericRecord, String> foreignKeyExtractor =
    (order, customer) -> order.get("customer_id").toString();

// Perform the join operation between the orders stream and the customers table
KStream<String, EnrichedOrder> enrichedOrdersStream = ordersStream
    .leftJoin(customersTable, (order, customer) -> new EnrichedOrder(order, customer))
    .selectKey(foreignKeyExtractor);
toStream metodu
KTable nesnesini yeni bir topic'e yazar
Örnek
Şöyle yaparız
import org.apache.kafka.streams.kstream.Printed;

KTable<String, Long> kt0 = ...
kt0.toStream().print(Printed.toSysOut());
Örnek
Şöyle yaparız
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> textLines = builder.stream("streams-plaintext-input");
KTable<String, Long> wordCounts = textLines.flatMapValues(value -> 
  Arrays.asList(pattern.split(value.toLowerCase())))
  .groupBy((key, value) -> value)
  .count();
  wordCounts.toStream().to("streams-wordcount-output", 
    Produced.with(Serdes.String(), Serdes.Long()));

Kafka Streams API KafkaStreams Sınıfı

Giriş
Şu satırı dahil ederiz
import org.apache.kafka.streams.KafkaStreams;
Belirtilen topology nesnesini çalıştırır

start metodu
Örnek
Şöyle yaparız
Properties prop = new Properties();
prop.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-lambda-example");
prop.put(StreamsConfig.CLIENT_ID_CONFIG, "wordcount-lambda-example-client");
prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, 
  Serdes.String().getClass().getName());
prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, 
  Serdes.String().getClass().getName());
prop.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/test");

StreamsBuilder builder = new StreamsBuilder();
...
KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), prop);
kafkaStreams.start();

Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));

Thread.sleep metodu

Giriş
Açıklaması şöyle.
Thread.sleep causes the current thread to suspend execution for a specified period. This is an efficient means of making processor time available to the other threads of an application or other applications that might be running on a computer system.
Açıklaması şöyle.
Causes the currently executing thread to sleep (temporarily cease execution) for the specified number of milliseconds, subject to the precision and accuracy of system timers and schedulers. The thread does not lose ownership of any monitors.
sleep metodu - long millis
Şöyle yaparız.
Thread.sleep (1000);
Şu iki kod aynı işi görür.
Thread.sleep (intervalInMills);
TimeUnit.MILLISECONDS.sleep (intervalInMills);
sleep(0)
Örnek
Şöyle yaparız
int i = 0;
while (i<10_000_000) {
  // business logic

  //prevent long gc
  if (i % 3000 == 0) {
    try {
      Thread.sleep(0);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}
Buradaki amaç şu değil
.. Thread.sleep(0)is not used to actively give up the CPU time slice,
Amaç şu
GC threads are known to have low priority, so Thread.sleep(0) is used to help GC threads try to compete for CPU time slice
Daha detaylı bir açıklama şöyle. GC işlemi rastgele bir yerde başlamaz. Sadece Safepoint noktalarda başlar. Döngüler Safepoint noktası değildir. Ancak Thread.sleep() bir Safepoint noktasıdır
Taking the HotSpot virtual machine as an example, the JVM does not pause at any position in the code instruction stream to start garbage collection, but forces that the execution must reach a safepoint before pausing. In other words, until the safepoint is reached, the JVM will not stop the world for GC.

The JVM will set safepoints on some loop jumps and method calls. However, in order to avoid the heavy burden of too many safepoints, the HotSpot virtual machine also has an optimization measure for loops. If the number of cycles is small, the execution time should not be too long. Therefore, loops that use int or smaller data types as index values will not be placed with safepoint by default. This kind of loop is called a countable loop. Correspondingly, a loop that uses long or a larger range of data types as index values is called an uncounted loop and will be placed at a safepoint.

However, we happen to have a countable loop here, so our code will not be placed at a safepoint. Therefore, the GC thread must wait until the thread finishes executing and can execute until the nearest safepoint. But if you use Thread.sleep(0), you can place a security point in the code.
Açıklamanın  devamı şöyle. Yani 10 milyon defa dönen bir döngüde arada bir Thread.sleep() kullanarak Safepoint noktası oluşturur ve arada bir GC olmasını sağlar. Böylece döngüden çıkınca uzun bir GC olmasının önüne geçilir.
Thread.sleep(0) is not a useless code. The sleep method can be used to place a safepoint in the java code. It can trigger the GC in a long loop in advance to prevent the GC thread from waiting for a long time, thus avoiding the goal of lengthening the GC time. The people who wrote this code are too strong. It’s really awesome.
sleep metodu - long millis, int nanos
Java 21 için açıklaması şöyle
Thread.sleep(long millis, int nanos) can now perform sub-millisecond sleeps on POSIX platforms. Before, non-zero arguments for nanos were rounded up to a full millisecond before.

Be aware that the actual precision still depends on the underlying system!