HazelcastAPI EntryListener Arayüzü - Kullanmayın

Giriş

Şu satırı dahil ederiz

import com.hazelcast.core.EntryListener;

Not : MapListener arayüzü tercih edilmeli. 

EntryListener  bir sürü başka arayüzden oluşuyor. Kod şöyle

/**
 * Map Entry listener to get notified when a map entry is added,
 * removed, updated, evicted or expired.  Events will fire as a result
 * of operations carried out via the IMap
 * interface.  Events will not fire, for example, for an entry
 * that comes into the Map via the MapLoader lifecycle.
 * 
 * This interface is here for backward compatibility reasons.
 * For a most appropriate alternative please use/check
 * com.hazelcast.map.listener.MapListener interface.
 *
 * @param <K> the type of key.
 * @param <V> the type of value.
 * @see IMap#addEntryListener
 * @see com.hazelcast.map.listener.MapListener
 */
public interface EntryListener<K, V>
        extends EntryAddedListener<K, V>,
        EntryUpdatedListener<K, V>,
        EntryRemovedListener<K, V>,
        EntryEvictedListener<K, V>,
        EntryExpiredListener<K, V>,
        MapClearedListener,
        MapEvictedListener {

}

Not : EntryListener arayüzündeki bir sürü metodu tek bir metoda yönlendiren ve kullanması daha basit olan EntryAdapter tercih edilebilir.

Örnek

EntryListener yerine kalıttığı arayüzleri teker teker kullanmak için şöyle yaparız

map.addEntryListener((EntryExpiredListener<Integer, Integer>) event -> {...}, 1, false);
map.addEntryListener((EntryEvictedListener<Integer, Integer>) event -> {...}, 2, false);

Örnek

Şöyle yaparız

import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.EntryListener;
import com.hazelcast.map.MapEvent;
public class MapEntryListener implements EntryListener {
  @Override
  public void entryAdded(EntryEvent entryEvent) {
    logger.info("key {} and value {}",entryEvent.getKey(),entryEvent.getValue());
  }
  @Override
  public void entryEvicted(EntryEvent entryEvent) {
    logger.info("Map Entry was evicted : {}",entryEvent);
  }
  @Override
  public void entryRemoved(EntryEvent entryEvent) {
    logger.info("Object with key {} removed from map.",entryEvent.getKey());
  }
  @Override
  public void entryUpdated(EntryEvent entryEvent) {
    logger.info("key {} updated from {} to {}.", entryEvent.getKey(),
      entryEvent.getOldValue(),entryEvent.getValue());
  }
  @Override
  public void mapCleared(MapEvent mapEvent) {
    logger.info("Map was cleared : {}",mapEvent);
  }
  @Override
  public void mapEvicted(MapEvent mapEvent) {
    logger.info("Map was evicted: {}",mapEvent);
  }
  @Override
  public void entryExpired(EntryEvent entryEvent) {
  }
}
IMap<String,String> hazelcastMap = ...;
hazelcastMap.addEntryListener(mapEntryListener,true);

HazelcastAPI DefaultQueryCache Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.map.impl.querycache.subscriber.DefaultQueryCache;

QueryCache

  InternalQueryCache

    AbstractInternalQueryCache

      DefaultQueryCache

add metodu

Callstack şöyle. Map'e ekleme işlemi yapıldıktan sonra ismi hz.client_1_event-N olan başka bir thread tarafından tetikleniyor.

add:101, DefaultQueryCacheRecordStore (com.hazelcast.map.impl.querycache.subscriber)
setInternal:104, DefaultQueryCache (com.hazelcast.map.impl.querycache.subscriber)
set:85, DefaultQueryCache (com.hazelcast.map.impl.querycache.subscriber)
handle:98, SubscriberAccumulatorHandler (com.hazelcast.map.impl.querycache.subscriber)
addQueryCache:181, SubscriberAccumulator (com.hazelcast.map.impl.querycache.subscriber)
accumulate:70, SubscriberAccumulator (com.hazelcast.map.impl.querycache.subscriber)
onEvent:64, SubscriberListener (com.hazelcast.map.impl.querycache.subscriber)
handleQueryCacheBatchEvent:261, ClientQueryCacheEventService$QueryCacheHandler (com.hazelcast.client.map.impl.querycache.subscriber)
handle:159, ContinuousQueryAddListenerCodec$AbstractEventHandler (com.hazelcast.client.impl.protocol.codec)
handleEventMessageOnCallingThread:190, ClientListenerServiceImpl (com.hazelcast.client.impl.spi.impl.listener)
run:361, ClientListenerServiceImpl$ClientEventProcessor (com.hazelcast.client.impl.spi.impl.listener)
process:245, StripedExecutor$Worker (com.hazelcast.internal.util.executor)
run:228, StripedExecutor$Worker (com.hazelcast.internal.util.executor)

Gönderilen event nesnesi şöyle. Bu listenin büyüklüğü QueryCacheConfig.setBatchSize() ile belirleniyor.

Collection<com.hazelcast.map.impl.querycache.event.QueryCacheEventData>

HazelcastAPI QueryCacheConfig Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.QueryCacheConfig;

Kodu şöyle

public class QueryCacheConfig implements IdentifiedDataSerializable {

  //By default, after reaching this minimum size, node immediately 
  //sends buffered events to QueryCache.
  public static final int DEFAULT_BATCH_SIZE = 1;

  //By default, only buffer last DEFAULT_BUFFER_SIZE 
  //events fired from a partition.
  public static final int DEFAULT_BUFFER_SIZE = 16;

  //Default value of delay seconds which an event wait in the buffer of a node, 
  //before sending to QueryCache.
  public static final int DEFAULT_DELAY_SECONDS = 0;

  //By default, also cache values of entries besides keys.
  public static final boolean DEFAULT_INCLUDE_VALUE = true;

  //By default, execute an initial population query prior to creation of the  QueryCache.
  public static final boolean DEFAULT_POPULATE = true;

  //Default value of coalesce property.
  public static final boolean DEFAULT_COALESCE = false;

  //Do not serialize given keys by default.
  public static final boolean DEFAULT_SERIALIZE_KEYS = false;

  //By default, hold values of entries in {@code QueryCache} as binary.
  public static final InMemoryFormat DEFAULT_IN_MEMORY_FORMAT = InMemoryFormat.BINARY; 
  ... 
}

HazelcastAPI QueryCache Arayüzü - Continuous Query Cache

Giriş

Şu satırı dahil ederiz

import com.hazelcast.map.QueryCache;

Açıklaması şöyle. Eğer birden fazla IMap nesnesine sorgu atmak istersek SQL kullanmak daha iyi.

QueryCache allows caching of the results of a query to improve the performance of read operations on the data. 

QueryCache also provides support for real-time updates to the cached data. When the data in the cache is updated, QueryCache automatically updates the cached result set, ensuring that the cached data is always up-to-date.

Bu arayüzü gerçekleştiren sınıf DefaultQueryCache. Hem client hem de server tarafında kullanılabilir. QueryCache  aslında filtrelenmiş bir IMap nesnesi gibidir.

constructor

Şöyle yaparız

String cacheName = ...;
QueryCacheConfig queryCacheConfig = new QueryCacheConfig(cacheName);
ClientConfig clientConfig = ...;
clientConfig.addQueryCacheConfig(mapName, queryCacheConfig);

IMap<Integer, Integer> map = ...;

QueryCache<Integer, Integer> queryCache = map
  .getQueryCache(cacheName, Predicates.alwaysTrue(), true);

Predicate için call stack şöyle. PutOperation esnasında Predicate çalışır ve gerekiyorsa QueryCacheEventPublisher ile gönderilir.

apply:59, TruePredicate (com.hazelcast.query.impl.predicates)
eval:54, QueryEventFilter (com.hazelcast.map.impl.query)
evaluateQueryEventFilter:77, AbstractFilteringStrategy (com.hazelcast.map.impl.event)
processQueryEventFilter:126, DefaultEntryEventFilteringStrategy (com.hazelcast.map.impl.event)
doFilter:85, DefaultEntryEventFilteringStrategy (com.hazelcast.map.impl.event)
getCQCEventTypeOrNull:152, QueryCacheEventPublisher (com.hazelcast.map.impl.event)
convertQueryCacheEventDataOrNull:122, QueryCacheEventPublisher (com.hazelcast.map.impl.event)
addEventToQueryCache:85, QueryCacheEventPublisher (com.hazelcast.map.impl.event)
postPublishEvent:275, MapEventPublisherImpl (com.hazelcast.map.impl.event)
publishEvent:232, MapEventPublisherImpl (com.hazelcast.map.impl.event)
publishEvent:196, MapEventPublisherImpl (com.hazelcast.map.impl.event)
publishEvent:184, MapEventPublisherImpl (com.hazelcast.map.impl.event)
afterRunInternal:87, BasePutOperation (com.hazelcast.map.impl.operation)
afterRun:225, MapOperation (com.hazelcast.map.impl.operation)
afterRun:368, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
call:297, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
run:258, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
run:486, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
process:197, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
process:137, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
executeRun:123, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
run:102, HazelcastManagedThread (com.hazelcast.internal.util.executor)

Örnek

Şöyle yaparız. Burada maaşı yüz binden büyük olan çalışanlar QueryCache  içine alınıyor. Eğer alttaki IMap değişirse, sonuçlar QueryCache nesnesine de yansıtılıyor

IMap<String, Employee> employees = ...
QueryCache<String, Employee> highSalaryEmployeesCache = employees
  .getQueryCache("highSalaryEmployeesCache", new SqlPredicate("salary > 100000"));

Collection<Employee> highSalaryEmployees = highSalaryEmployeesCache.values();
for (Employee employee : highSalaryEmployees) {
  System.out.println(employee.getName());
}

size metodu

QueryCache nesnesinin içindeki nesne sayısını döner

Hazelcast Jet JobMetrics Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.core.metrics.JobMetrics;

HazelcastAPI AbstractInvocationFuture Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.spi.impl.AbstractInvocationFuture;

Açıklamasında şöyle yazıyor

Custom implementation of CompletableFuture.

Kalıtım şöyle

AbstractInvocationFuture

  InvocationFuture

  ClientInvocationFuture

Client tarafında şöyle kullanılıyor

new ClientInvocation(client, request, objectName).invoke().joinInternal();

joinInternal metodu

Kodu şöyle. Burada Future cevap gelinceye kadar spin lock gibi sürekli işlemci harcayarak dönüyor.

@Override
public V joinInternal() {
  final Object response = registerWaiter(Thread.currentThread(), null);
  if (response != UNRESOLVED) {
    // no registration was done since a value is available.
    return resolveAndThrowForJoinInternal(response);
  }

  boolean interrupted = false;
  try {
    do {
      manageParking(0);
      if (isDone()) {
        return resolveAndThrowForJoinInternal(state);
       } else if (Thread.interrupted()) {
         interrupted = true;
         onInterruptDetected();
       }
     } while (true);
   } finally {
     restoreInterrupt(interrupted);
   }
}

manageParking metodu

Kodu şöyle. LockSupport.park() ile  bloke olmadan sadece tekrar schedule ediliyor

private void manageParking(long timeoutNanos) {
  try {
    // if the caller thread is a ForkJoinWorkerThread
    if (ForkJoinTask.inForkJoinPool()) {
      ForkJoinPool.managedBlock(new ManagedBlocker(timeoutNanos));
    } else if (timeoutNanos == 0) {
      park();
    } else if (timeoutNanos > 0) {
      parkNanos(timeoutNanos);
    }
  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();
  }
}

resolveAndThrowIfException metodu

Soyut bir metod. InvocationFuture ve ClientInvocationFuture tarafından override edilir.

HazelcastAPI SocketInterceptorConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.SocketInterceptorConfig;

setImplementation metodu

Şöyle yaparız

SocketInterceptorConfig socketInterceptorConfig = new SocketInterceptorConfig();
MySocketInterceptor mySocketInterceptor = ...
socketInterceptorConfig.setImplementation(mySocketInterceptor)
  .setEnabled(true);

Config config = ...;
config.getNetworkConfig().setSocketInterceptorConfig(socketInterceptorConfig);

HazelcastAPI AdvancedNetworkConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.AdvancedNetworkConfig;

se

HazelcastAPI SecurityConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.SecurityConfig;

Bu nesne etkinleştirildikten sonra çeşitli nesnelerin döndürdüğü

public List<Permission> permissions()

kontrolleri yapılmaya başlanır. 

Örnek

JOB submit için şöyle yaparız

Config config = ...
SecurityConfig securityConfig = config.getSecurityConfig();
securityConfig.setEnabled(true);


String name = null;
String role = "dev;
PermissionConfig permissionConfig = new PermissionConfig(PermissionType.JOB, name, role);
permissionConfig.getActions().addAll(asList(ActionConstants.ACTION_ALL));
securityConfig.addClientPermissionConfig(permissionConfig);

addClientPermissionConfig metodu

PermissionConfig nesnesi ekler

setEnabled metodu

Şöyle yaparız

Config config = ...
config.getSecurityConfig().setEnabled(true);

Hazelcast Jet StreamEventJournalP Sınıfı - Source

Giriş
Şu satırı dahil ederiz 

import com.hazelcast.jet.impl.connector.StreamEventJournalP;

Kalıtım şöyle

Processor
  AbstractProcessor
    StreamEventJournalP

Bu sınıfın içinde ki tane sınıf daha var

1. private static class ClusterMetaSupplier<E, T> implements ProcessorMetaSupplier - Yani bir ProcessorMetaSupplier ve Processor yaratır

2. private static class ClusterProcessorSupplier<E, T> implements ProcessorSupplier {

streamMapSupplier metodu

Çağrı sırası şöyle

Sources.streamMapP()
  StreamEventJournalP.streamMapSupplier()

streamRemoteMapSupplier metodu

Çağrı sırası şöyle

Sources.remoteMapJournal()
  StreamEventJournalP.streamRemoteMapSupplier()

Bir ClusterMetaSupplier nesnesi döner. 

Hazelcast Jet WriteJdbcP Sınıfı - JDBC Sink

Giriş
Şu satırı dahil ederiz 

import com.hazelcast.jet.impl.connector.WriteJdbcP;

Processor açısından bakarsak en önemli metodlar şöyle

1. init()

Yeni bir JDBC Connection nesnesi alır

2. process()

Inbox'tan yeni bir nesne alır ve onu veri tabanına yazar

3. close()

JDBC Connection nesnesini kapatır

Nesneyi Yaratmak

Bu nesneyi yaratmanın 2 yolu var.

1. SqlService.execute ile

2. Sinks vs gibi bir builder kod ile. Bu durumda metaSupplier metodu çağrılır

1. SqlService.execute

Eğer belirtilen SQL cümlesi "DELETE ..." ise DeleteProcessorSupplier tarafından yaratılır

metaSupplier metodu

Kodla yaratacaksak çağrı sırası şöyle

Sinks.jdbc()
  JdbcSinkBuilder.build()
    SinkProcessors.writeJdbcP()
      WriteJdbcP.metaSupplier()

Kod şöyle. ProcessorMetaSupplier döndürür. 

1- ProcessorMetaSupplier sınıfı ProcessorSupplier  döndürür

2. ProcessorSupplier İse Processor Yaratır

3. Yani isimlendirme saçma sapan. Kısaca ProcessorMetaSupplier aslında bir Processor factory. 

Koddaki ProcessorMetaSupplier.preferLocalParallelismOne() ile bir MetaSupplierFromProcessorSupplier nesnesi döndürür. Bu nesne de hep aynı ProcessorSupplier'ı döndürür. Yani DataSource nesnesini ProcessorSupplier yönetiyor

public static <T> ProcessorMetaSupplier metaSupplier(
  String jdbcUrl,
  String updateQuery,
  FunctionEx<ProcessorMetaSupplier.Context, ? extends CommonDataSource> 
    dataSourceSupplier,
  BiConsumerEx<? super PreparedStatement, ? super T> bindFn,
  boolean exactlyOnce,
  int batchLimit
) {
  ...
  return ProcessorMetaSupplier.preferLocalParallelismOne(
    ConnectorPermission.jdbc(jdbcUrl, ACTION_WRITE),
    new ProcessorSupplier() {
      private transient CommonDataSource dataSource;
      @Override
      public void init(@Nonnull Context context) {
        dataSource = dataSourceSupplier.apply(context);
      }
      @Override
      public void close(Throwable error) throws Exception {
        if (dataSource instanceof CloseableDataSource) {
          ((CloseableDataSource) dataSource).close();
        }
      }
      @Override
      public Collection<? extends Processor> get(int localParallelism) {
        return IntStream.range(0, count)
          .mapToObj(i -> new WriteJdbcP<>(updateQuery, dataSource, bindFn,
                                        exactlyOnce, batchLimit))
          .collect(Collectors.toList());
      }
     ...
    });
}

Birinci parametre jdbc url

İkinci parametre SQL

Üçüncü parametre DataSource döndüren metod

Hazelcast Jet Processor Arayüzü

Giriş
Şu satırı dahil ederiz 

import com.hazelcast.jet.core.Processor;

Açıklaması şöyle

When you submit a Jet job, the job is divided into multiple "processing units" called "processors". These processors are then distributed across the nodes in the cluster where the Hazelcast Jet instance is running.

Kalıtım şöyle

Processor

  AbstractProcessor

  AsyncHazelcastWriterP

  KinesisSinkP

  NoopP

  ProcessorWrapper

  RootResultConsumerSink

  WriteBufferedP

  WriteFileP 

  WriteKafkaP

  WriteMapP

  XaSinkProcessorBase

HazelcastAPI EventJournalConfig Sınıfı - Bir Map'e Yapılan Değişiklikleri Okumak İçindir

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.EventJournalConfig;

setCapacity

Örnek

Şöyle yaparız. Burada Config nesnesine addEventJournalConfig() ile ekleniyor

Config config = new Config();
// Add an event journal config for map which has custom capacity of 1000 (default 10_000)
// and time to live seconds as 10 seconds (default 0 which means infinite)
config.addEventJournalConfig(new EventJournalConfig().setEnabled(true)
                                                     .setMapName(MAP_NAME)
                                                     .setCapacity(1_000)
                                                     .setTimeToLiveSeconds(100));

Örnek

Şöyle yaparız. Burada Config nesnesine getMapConfig() ile ekleniyor

Config config = new Config();
config.getMapConfig(MAP_NAME)
  .getEventJournalConfig()
    .setEnabled(true)
    .setCapacity(10_000)
    .setTimeToLiveSeconds(10);

Hazelcast Jet ProcessorMetaSupplier Arayüzü - Job Coordinator'a Gönderilir

Giriş

Açıklaması şöyle

Factory of ProcessorSupplier instances. The starting point of the chain leading to the eventual creation of Processor instances on each cluster member:

1. client creates ProcessorMetaSupplier as a part of the DAG;

2. serializes it and sends to a cluster member;

3. the member deserializes and uses it to create one ProcessorSupplier for each cluster member;

4. serializes each ProcessorSupplier and sends it to its target member;

5. the target member deserializes and uses it to instantiate as many instances of Processor as requested by the parallelism property on the corresponding Vertex.

Before being asked to create ProcessorSuppliers this meta-supplier will be given access to the Hazelcast instance and, in particular, its cluster topology and partitioning services. It can use the information from these services to precisely parameterize each Processor instance that will be created on each member.

1. Kısaca Job başlatmak isteyen client veya member, JobDefinition nesnesini Job Coordinator'a gönderir.
2. JobDefinition içeriğinde ProcessorMetaSupplier'da bulunur. 
3. ProcessorMetaSupplier  Job Coordinator tarafından çalıştırılır ve her bir member için ProcessorSupplier döndüren bir supplier üretir.
4. ProcessorSupplier gerekli member'a gönderilir ve bu member ProcessorSupplier nesnesini kullanarak Processor Yaratır

1. client creates ProcessorMetaSupplier as a part of the DAG;

Örnek bir kod şöyle. Yani ProcessorMetaSupplier  Vertex'e ekleniyor

Planner p = ...
ProcessorMetaSupplier metaSupplier = ...
...
p.addVertex(this, name(), determinedLocalParallelism(), metaSupplier);

Metodlar

get metodu - List<Address>

İmzası şöyle. ProcessorSupplier döndüren bir Function döner. Burada sanırım addresses listesi get() metodunun üzerinde çalıştığı tüm member listesini belirtmek için kullanılıyor. Bu parametre çoğunlukla önemli değil

Function<? super Address, ? extends ProcessorSupplier> get(List<Address> addresses);

Örnek

Şöyle yaparız

public class MyProcessorMetaSupplier implements ProcessorMetaSupplier {

  @Override
  public Function<? super Address, ? extends ProcessorSupplier> get(
    List<Address> addresses) {
    return address -> {
      // Get the index of the member corresponding to this address
      int memberId = getMemberId(addresses, address);

      // Create a custom processor with behavior specific to this member
      Processor processor = new MyProcessor(memberId);

      // Return a processor supplier that provides instances of the custom processor
      return ProcessorSupplier.of(processor);
    };
  }

  // Helper method to get the index of a member based on its address
  private int getMemberId(List<Address> addresses, Address address) {
    int index = addresses.indexOf(address);
    return index < 0 ? -1 : index;
  }
}

of metodu

Sanırım tüm member'lar üzerinde Vertex.LOCAL_PARALLELISM_USE_DEFAULT local parallelism ile çalışır

preferLocalParallelismOne metodu

Sanırım tüm member'lar üzerine 1 local parallelism ile çalışır

preferLocalParallelismOnSingleMember metodu

Rastgele bir member üzerinde belirtilen localParallelism kadar Processor yaratır

randomMember metodu

Rastgelen  bir member'lar üzerine 1 local parallelism ile çalışır

HazelcastClientAPI TcpClientConnectionManager Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.client.impl.connection.tcp.TcpClientConnectionManager;

Bu sınıfın alanlarından birisi şöyle. TcpClientConnection nesnelerini saklar

private final ConcurrentMap<UUID, TcpClientConnection> activeConnections = 
  new ConcurrentHashMap<>();

Bir diğer alana ise şöyle. Cluster üyelerini bulduktan sonra bunlara bağlanmaya çalışır

private final ClusterDiscoveryService clusterDiscoveryService;

doConnectToCluster metodu

Metodun için şöyle. Keşfedilen cluster üzerinde dolaşır

private void doConnectToCluster() {
  CandidateClusterContext currentContext = clusterDiscoveryService.current();

  logger.info("Trying to connect to cluster: " + currentContext.getClusterName());

  // try the current cluster
  if (doConnectToCandidateCluster(currentContext, false)) {
     connectionProcessListenerRunner
       .onClusterConnectionSucceeded(currentContext.getClusterName());
     return;
  }
  ...
}

HazelcastAPI Listeners

Giriş

Hazelcast içinde çok fazla sayıda Listener arayüzü var. Hepsi java.util.EventListener arayüzünden kalıtıyor.

Bazı önemli olanlar şöyle

DistributedObjectListener

LifecycleListener

MapListener : Bu arayüzle şunlar da geliyor.

MapClearedListener, MapEvictedListener, EntryAddedListener, EntryEvictedListener, EntryExpiredListener, EntryRemovedListener, EntryMergedListener, EntryUpdatedListener, 

EntryLoadedListener

MembershipListener

MigrationListener

PartitionLostListener

HazelcastAPI SerializationService Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.internal.serialization.SerializationService;

Kodu şöyle

/**
 * SPI to serialize user objects to Data and back to Object
 * Data is the internal representation of binary data in hazelcast.
 */
public interface SerializationService {
   ...
}

Yani yaptığı tek şey toData(Object obj) veya toObject(Object data)  metodları ile nesneyi Data formatına çevirmek veya tam tersi

Kalıtım şöyle

SerializationService

  InternalSerializationService

  AbstractSerializationService

    DelegatingSerializationService

    SerializationServiceV1

Kendi içinde çalışırken SerializerAdapter arayüzünü kullanır. SerializerAdapter kalıtımı şöyle

SerializerAdapter 

  ByteArraySerializerAdapter : ByteArraySerializer kullanır

  CompactWithSchemaStreamSerializerAdapter : CompactStreamSerializer kullanır

  StreamSerializerAdapter : StreamSerializer kullanır

    CompactStreamSerializerAdapter

StreamSerializer  kalıtımı çok uzun. Bazı ilginç olanlar şöyle

  SingletonSerializer

    JavaSerializer

toData metodu

İmzası şöyle. Data nesnesi döner

<B extends Data> B toData(Object obj);

Örnek

Şöyle yaparız

IMap customerMap = ...

SerializationService ss = ...
System.out.println("Binary size of value is " + 
             ss.toData(customerMap.get(1)).bufferSize() + " bytes");

toObject metodu

İmzası şöyle

<T> T toObject(Object data);

HazelcastAPI HealthMonitor Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.internal.diagnostics.HealthMonitor;

MetricsRegistry sınıfından bir sürü şeyi okuyarak periyodik olarak loglar. 

run metodu

Kod şöyle

@Override
public void run() {
  try {
    while (node.getState() == NodeState.ACTIVE) {
      healthMetrics.update();
      switch (monitorLevel) {
        case NOISY:
          if (healthMetrics.exceedsThreshold()) {
            logDiagnosticsHint();
          }
          logger.info(healthMetrics.render());
          break;
        case SILENT:
          if (healthMetrics.exceedsThreshold()) {
            logDiagnosticsHint();
            logger.info(healthMetrics.render());
          }
          break;
        default:
          throw new 
            IllegalStateException("Unrecognized HealthMonitorLevel: " + monitorLevel);
       }
       try {
         SECONDS.sleep(delaySeconds);
       } catch (InterruptedException e) {
         currentThread().interrupt();
         return;
       }
     }
   } catch (OutOfMemoryError e) {
     OutOfMemoryErrorDispatcher.onOutOfMemory(e);
   } catch (Throwable t) {
     logger.warning("Health Monitor failed", t);
   }
 }

Eğer SILENT ise açıklaması şöyle

These logs are triggered when certain performance indicators exceed a threshold. 

HazelcastAPI NodeEngineImpl Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.spi.impl.NodeEngineImpl;

Kod şöyle. Çoğu servis bu sınıfın içinde

public class NodeEngineImpl implements NodeEngine {

  private final Node node;
  private final SerializationService serializationService;
  private final SerializationService compatibilitySerializationService;
  private final LoggingServiceImpl loggingService;
  private final ILogger logger;
  private final MetricsRegistryImpl metricsRegistry;
  private final ProxyServiceImpl proxyService;
  private final ServiceManagerImpl serviceManager;
  private final ExecutionServiceImpl executionService;
  private final OperationServiceImpl operationService;
  private final EventServiceImpl eventService;
  private final OperationParkerImpl operationParker;
  private final ClusterWideConfigurationService configurationService;
  private final TransactionManagerServiceImpl transactionManagerService;
  private final WanReplicationService wanReplicationService;
  private final Consumer<Packet> packetDispatcher;
  private final SplitBrainProtectionServiceImpl splitBrainProtectionService;
  private final SqlServiceImpl sqlService;
  private final Diagnostics diagnostics;
  private final SplitBrainMergePolicyProvider splitBrainMergePolicyProvider;
  private final ConcurrencyDetection concurrencyDetection;
  private final TenantControlServiceImpl tenantControlService;
  private final DataLinkService dataLinkService;
  ...
}

start metodu

Servisleri başlatır

HazelcastAPI Node Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.instance.impl.Node;

Kod şöyle. HazelcastInstanceImpl sınıfının bir alanıdır. Bazı servisler bu sınıfın içindedir. Bazı servisler de NodeEngineImpl içindedir.

public class Node {

  // name of property used to inject ClusterTopologyIntentTracker in Discovery Service
  public static final String DISCOVERY_PROPERTY_CLUSTER_TOPOLOGY_INTENT_TRACKER =
            "hazelcast.internal.discovery.cluster.topology.intent.tracker";

  private static final int THREAD_SLEEP_DURATION_MS = 500;
  private static final String GRACEFUL_SHUTDOWN_EXECUTOR_NAME = "hz:graceful-shutdown";

  public final HazelcastInstanceImpl hazelcastInstance;
  public final DynamicConfigurationAwareConfig config;
  public final NodeEngineImpl nodeEngine;
  public final ClientEngine clientEngine;
  public final InternalPartitionServiceImpl partitionService;
  public final ClusterServiceImpl clusterService;
  public final MulticastService multicastService;
  public final DiscoveryService discoveryService;
  public final TextCommandService textCommandService;
  public final LoggingServiceImpl loggingService;
  public final Server server;

  /**
   * Member-to-member address only.
   * When the Node is configured with multiple endpoints, this address still 
   * represents ProtocolType.MEMBER
   * For accessing a full address-map, see AddressPicker.getPublicAddressMap()
   */
  public final Address address;
  public final SecurityContext securityContext;
  final ClusterTopologyIntentTracker clusterTopologyIntentTracker;

  private final ILogger logger;
  private final AtomicBoolean shuttingDown = new AtomicBoolean(false);
  private final NodeShutdownHookThread shutdownHookThread;
  private final MemberSchemaService schemaService;
  private final InternalSerializationService serializationService;
  private final InternalSerializationService compatibilitySerializationService;
  private final ClassLoader configClassLoader;
  private final NodeExtension nodeExtension;
  private final HazelcastProperties properties;
  private final BuildInfo buildInfo;
  private final HealthMonitor healthMonitor;
  private final Joiner joiner;
  private final LocalAddressRegistry localAddressRegistry;
  private ManagementCenterService managementCenterService;

  // it can be changed on cluster service reset 
  //see: ClusterServiceImpl#resetLocalMemberUuid
  private volatile UUID thisUuid;
  private volatile NodeState state = NodeState.STARTING;
  ...
}

start metodu

NodeEngineImpl nesnesini ve servisleri başlatır

HazelcastAPI PartitionService Arayüzü

Giriş 

Şu satırı dahil ederiz

import com.hazelcast.partition.PartitionService;

getPartition metodu

Şöyle yaparız

HazelcastInstance instance = ...;
PartitionService partitionService = instance.getPartitionService();

String str = ...,
Partition partition = partitionService.getPartition(str);

Partition nesnesinden tekrar Member nesnesine erişebiliriz. Şöyle yaparız

Member owner = partition.getOwner();

HazelcastQueryAPI IndexStore Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.query.impl.IndexStore;

Kalıtım şöyle

IndexStore

  BitmapIndexStore

  BaseIndexStore

    BaseSingleValueIndexStore

      OrderedIndexStore

      UnorderedIndexStore

IndexStore'un sahibi ise 

Index

  InternalIndex

    AbstractIndex : Bu sınıf içinde IndexStore var

      IndexImpl

OrderedIndexStore Sınıfı

Şu satırı dahil ederiz

import com.hazelcast.query.impl.OrderedIndexStore;

Kodu şöyle. Yani indeks ConcurrentSkipListMap<Comparable, NavigableMap<Data, QueryableEntry> yapısına sahip recordMap değişkeni.

public class OrderedIndexStore extends BaseSingleValueIndexStore {
  public static final Comparator<Data> DATA_COMPARATOR = new DataComparator();

  private final ConcurrentSkipListMap<Comparable, NavigableMap<Data, QueryableEntry>> 
    recordMap = new ConcurrentSkipListMap<>(Comparables.COMPARATOR);

  private final IndexFunctor<Comparable, QueryableEntry> addFunctor;
  private final IndexFunctor<Comparable, Data> removeFunctor;

  private volatile SortedMap<Data, QueryableEntry> recordsWithNullValue;
    ...
}

  

HazelcastQueryAPI LocalIndexStats Arayüzü - Map üzerine Konulan Belirli Bir İndeks İçin İstatistiki Bilgi Verir

Giriş

Şu satırı dahil ederiz

import com.hazelcast.query.LocalIndexStats;

İlişki şöyle

LocalMapStats has LocalIndexStats

Map üzerine konulan belirli bir indeks için istatistiki bilgi verir. Bu arayüzü gerçekleştiren sınıf LocalIndexStatsImpl. Bu sınıf iki kaynaktan dolduruluyor

PerIndexStats

  PartitionPerIndexStats

  GlobalPerIndexStats

Kaynağın sahibi ise 

Index

  InternalIndex

    AbstractIndex : Bu sınıf içinde IndexStore var

      IndexImpl

    

Kaynağı dış dünyaya açan sınıf LocalMapStatsProvider

Örnek

Şöyle yaparız

HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IMap<String, String> customers = hazelcastInstance.getMap("customers");        
addIndex(customers, "name", true); // or add the index using the map config

LocalMapStats mapStatistics = customers.getLocalMapStats();
Map<String, LocalIndexStats> indexStats = mapStatistics.getIndexStats();

LocalIndexStats nameIndexStats = indexStats.get("name");
System.out.println("average name index hit selectivity on this member = "
  + nameIndexStats.getAverageHitSelectivity());

getTotalInsertLatency metodu

Insert işleminin kaç nanosaniye sürdüğünü belirtir.