Welcome to Hazelcast Reference Manual. This manual includes concepts, instructions and samples to guide you on how to use Hazelcast and build Hazelcast applications.
As the reader of this manual, you must be familiar with Java programming language and should have your preferred IDE.
Throughout this manual;
Hazelcast is free provided under Apache 2 license. Hazelcast Enterprise is commercially licensed by Hazelcast, Inc.
Hazelcast is a registered trademark of Hazelcast, Inc. All other trademarks in this manual are held by their respective owners.
Support for Hazelcast is provided via GitHub, Mail Group and StackOverflow.
For information on support for Hazelcast Enterprise, please see hazelcast.com/support/commercial.
Below table shows the conventions used in this manual.
| Convention | Description |
|---|---|
| bold font | - Indicates words in a sentence requiring reader's specific attention. - Also indicates property/parameter values. |
| italic font | - When used with "<" and ">" before and after it, indicates a variable in a command or code syntax that you must replace, e.g. hazelcast-< version>.jar . - Also, Note and Related Information texts are in italic fonts. |
monospace |
- Indicates file, folder, class and library names, code snippets, and inline code words in a sentence. |
| RELATED INFORMATION | - Indicates a resource that is relevant to the topic, usually with a link or cross-reference. |
| NOTE | Indicates information that is of special interest or importance, e.g. an additional action required only in certain circumstances. |
This section provides the new features introduced with Hazelcast 3.3 release.
evictAll
and loadAll
methods have been introduced to be able to evict all entries except the locked ones and that loads all or a set of keys from a configured map store, respectively. Please see Evicting All Entries and Forcing All Keys to be Loaded sections for more information.3.3.2 Fixes
This section lists issues solved for Hazelcast 3.3.2 release.
Map#getEntryView
should check expiration of a key [#3801].HazelcastInstanceNotActiveException
loop during multicast join [#3732].maxIdleTime
expiration when running EntryProcessors. A delay should be added to expiration times on backups [#3710].containsKey()
in transactional context returns wrong value for keys deleted within transaction [#3682].TransactionalMap.values()
returns stale values that was updated within the transaction boundary [#3668].3.3.1 Fixes
This section lists issues solved for Hazelcast 3.3.1 release.
connectionTimeout
property in ClientNetworkConfig is set to Integer.MAX_VALUE
, the client could not connect to cluster since a default 2000 ms. extra value is added to connectionTimeout
while connecting [#3615].MaxSizeConfig
constructor should convert zero size to Integer.MAX_VALUE
[#3579].manifest.mf
files are not correctly set #3537.mancenter.war
from Hazelcast release 3.2.5 cannot be deployed to Glassfish <IP_ADDRESS> and it fails to deploy #3501.evictAll
, locked keys should stay in the map #3473.hazelcast-vm
module, before every test, new server container is started. And after every test, running server is terminated. This behavior causes a long test execution time. Server start-up and termination should be done before and after test class initialization and finalization #3473.IQueue.take()
method should throw InterruptedException, but throws HazelcastException instead #3133.3.3 Fixes
This section lists issues solved for Hazelcast 3.3 release.
java.lang.NullPointerException
is thrown when publishing an event in ClientEndPointImpl [#3407].entryUpdated()
callback of a listener during a transaction always has a null oldValue
in the EntryEvent [#3406].map.replace()
stores replaced items [#3386].setTransactionTimeout()
method is not correctly implemented [#3384].isStarted == false
, resulting in errors down the line [#3350].PagingPredicate.getAnchor
does not return the correct value [#3241].RC3 Fixes
This section lists issues solved for Hazelcast 3.3-RC3 release.
MapStore#store
method for the same key triggered by IMap#flush
[#3338].addAll()
performed on this list does not throw an exception [#3330].java.io.FileNotFoundException
thrown by MapLoaderTest [#3324].ExpirationManager
partition sorting can fail [#3271].getOldValue
and getValue
returns the same value when removing item from IMap [#3198].IMap.delete()
should not call MapLoader.load()
[#3178].connectionMarkedAsNotResponsive
[#3169].WebFilter.HazelcastHttpSession.isNew()
does not check the Hazelcast Session Cache [#3132].ClassCastException: java.lang.Integer
cannot be cast to java.lang.String
Query [#3091].expirationTime
on EntryView is not set [#3038].BasicRecordStoreLoader
cannot handle retry responses [#3033]. await()
on condition of contended lock causes IllegalStateException [#3025]. IMap#keySet
triggers value deserialization [#3008].map.destroy()
throws DistributedObjectDestroyedException [#3001].com.hazelcast.jca.HazelcastConnection::getExecutorService
returns plain ExecutorService [#2986].IMap.getAll()
[#2982].RC2 Fixes
This section lists issues solved for Hazelcast 3.3-RC2 release.
evictAll
should flush to staging area #2969.AddSessionEntryProcessor
[#2955].StripedExecutor
to WanReplicationService [#2947].hazelcast-wm
module [#2927].tls
record size [#2914].RC1 Fixes
This section lists issues solved for Hazelcast 3.3-RC1 release.
wget
on it [#2814].mapCleared
method for EntryListener is needed [#2789].keySet
with predicate should trigger loading of MapStore [#2692].IMap.remove()
followed by IMap.putIfAbsent(key,value)
still returns the old value [#2685].System.nanoTime()
[#2666].AbstractReachabilityHandler
writes to standard output [#2591].IMap.set()
does not not remove a key from write behind deletions queue [#2588].com.hazelcast.core.EntryView#getLastAccessTime
is invalid[#2581].In this section, we list the changes what users should take into account before upgrading to latest Hazelcast from earlier versions.
Map<Integer, String> customers = Hazelcast.getMap( "customers" );
with
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
// or if you already started an instance named "instance1"
// HazelcastInstance hazelcastInstance = Hazelcast.getHazelcastInstanceByName( "instance1" );
Map<Integer, String> customers = hazelcastInstance.getMap( "customers" );
Removal of lite members: With 3.0 there will be no member type as lite member. As 3.0 clients are smart client that they know in which node the data is located, you can replace your lite members with native clients.
Renaming "instance" to "distributed object": Before 3.0 there was a confusion for the term "instance". It was used for both the cluster members and the distributed objects (map, queue, topic, etc. instances). Starting 3.0, the term instance will be only used for Hazelcast instances, namely cluster members. We will use the term "distributed object" for map, queue, etc. instances. So you should replace the related methods with the new renamed ones. As 3.0 clients are smart client that they know in which node the data is located, you can replace your lite members with native clients.
public static void main( String[] args ) throws InterruptedException {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IMap map = hazelcastInstance.getMap( "test" );
Collection<Instance> instances = hazelcastInstance.getInstances();
for ( Instance instance : instances ) {
if ( instance.getInstanceType() == Instance.InstanceType.MAP ) {
System.out.println( "There is a map with name: " + instance.getId() );
}
}
}
with
public static void main( String[] args ) throws InterruptedException {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IMap map = hz.getMap( "test" );
Collection<DistributedObject> objects = hazelcastInstance.getDistributedObjects();
for ( DistributedObject distributedObject : objects ) {
if ( distributedObject instanceof IMap ) {
System.out.println( "There is a map with name: " + distributedObject.getName() );
}
}
}
com.hazelcast.core
from com.hazelcast.partition
.removeListener
methods was taking the Listener object as parameter. But, it causes confusion as same listener object may be used as parameter for different listener registrations. So we have changed the listener API. addListener
methods return you an unique ID and you can remove listener by using this ID. So you should do following replacement if needed:IMap map = hazelcastInstance.getMap( "map" );
map.addEntryListener( listener, true );
map.removeEntryListener( listener );
with
IMap map = hazelcastInstance.getMap( "map" );
String listenerId = map.addEntryListener( listener, true );
map.removeEntryListener( listenerId );
tryRemove(K key, long timeout, TimeUnit timeunit)
returns boolean indicating whether operation is successful.tryLockAndGet(K key, long time, TimeUnit timeunit)
is removed.putAndUnlock(K key, V value)
is removed.lockMap(long time, TimeUnit timeunit)
and unlockMap()
are removed.getMapEntry(K key)
is renamed as getEntryView(K key)
. The returned object's type, MapEntry class is renamed as EntryView.<merge-policy>com.hazelcast.map.merge.PassThroughMergePolicy</merge-policy>
Also MergePolicy interface has been renamed to MapMergePolicy and also returning null from the implemented merge()
method causes the existing entry to be removed.
IQueue
is configured. With Hazelcast 3.0 there will not be backing map configuration for queue. Settings like backup count will be directly configured on queue config. For queue configuration details, please see Queue.pause()
, resume()
, restart()
methods have been removed.AtomicNumber
class has been renamed to IAtomicLong
.await()
operation has been removed. We expect users to use await()
method with timeout parameters.ISemaphore
has been substantially changed. attach()
, detach()
methods have been removed.max-size
eviction policy was cluster_wide_map_size. In 3.x releases, default is PER_NODE. After upgrading, the max-size
should be set according to this new default, if it is not changed. Otherwise, it is likely that OutOfMemory exception may be thrown.| Chapter | Section | Description |
|---|---|---|
| Chapter 1 - Preface | Added as a new chapter. | |
| Chapter 4 - Distributed Data Structures | IdGenerator, ICountDownLatch, IAtomicReference | Added as new sections. |
| MultiMap | Section improved by adding information about how it is implemented and by providing configuration information. | |
| Map | Eviction section improved by adding the detailed policy and parameter explanations. A subsection on how to evict particular map entries added (Evicting Specific Entries). Also Map Overview and Backups sections enhanced with "how it works" information. Added Entry Listener and Map Locks as new sections. Added Evicting All Entries section to explain the method evictAll . |
|
| Map Persistence | Added Forcing All Keys To Be Loaded section to explain the method loadAll . Also, added Read-Through, Write-Through and Write-Behind sections. Section also enhanced by adding MapStore and MapLoader descriptions with a sample code. Thread information related to MapLoader interface added as a note. |
|
| Queue | Section improved with new subsections. QueueStore interface explanations added. Bounded Queue section modified for a cleaner description. | |
| Lock, ISemaphore | Sections updated by adding fairness related warnings. | |
| Chapter 6 - Distributed Computing | Executor Service | HazelcastInstanceAware related information added. |
| Chapter 7 - Distributed Query | Query section improved by adding 'how it works' information. Also, Continuous Query section re-written and sample codes added. Aggregators added as a new section. A note related to the indexing of non-primitive types added to Indexing section. | |
| Chapter 8 - User Defined Services | Added as a new chapter explaining Hazelcast's Service Provider Interface (SPI). | |
| Chapter 9 - Transactions | XA Transactions | Added as a new section. |
| Sample JBoss AS 7 - EAP 6 Web Application Configuration | Added as a new section. | |
| Chapter 10 - Hazelcast JCache Implementation | Added as a new chapter. | |
| Chapter 11 - Integrated Clustering | Tomcat Based Web Session Replication | Added as a new section (Enterprise Only feature). |
| Filter Based Web Session Replication | Updated by adding SpringAwareWebFilter related information. | |
| Chapter 13 - Clients | Java Client | Client configuration related to AWS added and the whole section enhanced. Also, Load Balancer related configuration information added. |
| Chapter 14 - Serialization | All sections under this chapter re-written and enhanced. Also added HazelcastInstanceAware, Stream & ByteArray Serializer, Serializable & Externalizable sections. Chapter introduction enhanced by adding an overview, a section explaining serialization interfaces and a comparison table for these interfaces. | |
| Null Portable Serialization | Added as a new section. | |
| Chapter 15 - Management | Clustered JMX | Added as a new section explaining how to monitor the statistics of distributed objects from a JMX interface (Enterprise Only feature). |
| Cluster Utilities | Member Attributes added as a new section. Also, Cluster-Member Safety Check section added explaining the PartitionService interface of Hazelcast. | |
| Clustered REST | Added as a new section (Enterprise Only feature). | |
| Chapter 17 - Performance | Threading Model | Added as a new section. |
| Chapter 18 - WAN | WAN Replication Queue Size | Added as a new section (Enterprise only feature). |
| Chapter 19 - Configuration | Configuring TCP/IP Cluster | Section improved by adding more detailed configuration options. |
| EC2 Auto Discovery | Section improved by adding AWSClient Class description, Debugging information and more detailed tag explanations. | |
| Ports | Added as a new section explaining how to configure ports which Hazelcast uses to communicate between nodes. Also existing "Restricting Outbound Ports" moved under this new section with the heading "Outbound Ports". | |
| System Property | Updated by adding the descriptions of missing system properties. | |
| Chapter 20 - Network Partitioning | Added as a new chapter. | |
| Chapter 21 - FAQ | Updated with new questions. | |
| Chapter 22 - Glossary | Added as a new chapter. |
Hazelcast is an open source In-Memory Data Grid (IMDG). As such it provides elastically scalable distributed In-Memory computing, widely recognized as the fastest and most scalable approach to application performance, and Hazelcast does so in open source. More importantly it makes distributed computing simple by offering distributed implementations of developer friendly interfaces from Java such as Map, Queue, ExecutorService, Lock, JCache and many more. For example, the Map interface provides an In-Memory Key Value store which confers many of the advantages of NoSQL in terms of developer friendliness and developer productivity.
In addition to distributing data In-Memory, Hazelcast provides a convenient set of APIs to access the CPUs in your cluster for maximum processing speed. Hazelcast is designed to be lightweight and easy to use. Since Hazelcast is delivered as a compact library (JAR) and has no external dependencies other than Java, it is easily pluggable into your software solution to provide distributed data structures and distributed computing utilities.
Hazelcast is highly scalable and available. Distributed applications can use Hazelcast for distributed caching, synchronization, clustering, processing, pub/sub messaging, etc. Hazelcast is implemented in Java and has clients for Java, C/C++, .NET as well as REST. Hazelcast can also speak memcache protocol. It also plugs in to Hibernate and can easily be used with any existing database system.
If you are looking for In-Memory speed, elastic scalability and the developer friendliness of NoSQL, Hazelcast is a great choice for you.
Hazelcast is simple
Hazelcast is written in Java with no other dependencies. It exposes the same API from the familiar Java util package. Just add hazelcast.jar
to your classpath, enjoy JVMs clustering in less than a minute and start building scalable applications.
Hazelcast is Peer-to-Peer
Unlike many NoSQL solutions, Hazelcast is peer-to-peer. There is no master and slave; there is no single point of failure. All nodes store equal amount of data and do equal amount of processing. Hazelcast can be embedded to your existing application or used in client and server mode where your application is client to the Hazelcast nodes.
Hazelcast is scalable
Hazelcast is designed to scale up to hundreds and thousands of nodes. Simply add new nodes and they will automatically discover the cluster and will linearly increase both memory and processing capacity. The nodes maintain a TCP connection between each other and all communication is performed through this layer.
Hazelcast is fast
Hazelcast stores everything in-memory. It is designed to perform very fast reads and updates.
Hazelcast is redundant
Hazelcast keeps the backup of each data entry on multiple nodes. On a node failure, the data is restored from the backup and cluster will continue to operate without a downtime.
Sharding in Hazelcast
Hazelcast shards are called Partitions. By default, Hazelcast has 271 partitions. Given a key; we serialize, hash and mode it with the number of partitions to find the partition it belongs to. The partitions themselves are distributed equally among the members of the cluster. Hazelcast also creates the backups of partitions and also distributes them among nodes for redundancy.
Partitions in a 1 node Hazelcast cluster.
Partitions in a 2 node cluster.
The blacks are primary partitions and reds are backups. In the above illustration, first node has 135 primary partitions (black) and each of these partitions are backed up in the second node (red). At the same time, first node has the backup partitions of second node's primary partitions.
As you add more nodes, Hazelcast will move one by one some of the primary and backup partitions to new nodes to make all nodes equal and redundant. Only minimum amount of partitions will be moved to scale out Hazelcast.
Hazelcast Topology
If you have an application whose main focal point is asynchronous or high performance computing and lots of task executions, then embedded deployment is the most useful. In this type, nodes include both the application and data, see the below illustration.
You can have a cluster of server nodes that can be independently created and scaled. Your clients communicate with these server nodes to reach to the data on them. Hazelcast provides native clients (Java, .NET and C++), Memcache clients and REST clients. See the below illustration.
A Glance at Traditional Data Persistence
Data is at the core of software systems and in conventional architectures, relational database persists and provides access to data. Basically, applications are talking directly with a database which has its backup as another machine. To increase the performance, tuning or a faster machine is required. This leads to a large amount of money or effort.
Then, there is the idea of keeping copies of data next to the database. This is performed using technologies like external key-value stores or second level caching. This helps to offload the database. However, when the database is saturated or if the applications perform mostly "put" operations (writes), this approach is of no use, since it insulates the database only from the "get" loads (reads). Even if the applications is read intensive, then there can be consistency problems: when data changes, what happens to the cache, and how are the changes handled? This is when concepts like time-to-live (TTL) or write-through come in.
However, in the case of TTL, if the access is less frequent then the TTL, the result will always be a cache miss. On the other hand, in the case of write-through caches; if there are more than one of these caches in a cluster, then we again have consistency issues. This can be avoided by having the nodes communicating with each other so that entry invalidations can be propagated.
We can conclude that an ideal cache would combine TTL and write-through features. And, there are several cache servers and in-memory database solutions in this field. However, those are stand-alone single instances with a distribution mechanism to an extent provided by other technologies. This brings us back to square one: we would experience saturation or capacity issues if the product is a single instance or if consistency is not provided by the distribution.
And, there is Hazelcast
Hazelcast, a brand new approach to data, is designed around the concept of distribution. Data is shared around the cluster for flexibility and performance. It is an in-memory data grid for clustering and highly scalable data distribution.
One of the main features of Hazelcast is not having a master node. Each node in the cluster is configured to be the same in terms of functionality. The oldest node manages the cluster members, i.e. automatically performs the data assignment to nodes.
Another main feature is the data being held entirely in-memory. This is fast. In the case of a failure, such as a node crash, no data will be lost since Hazelcast distributes copies of data across all the nodes of cluster.
As it can be seen in the feature list given in Hazelcast Overview section, Hazelcast supports a number of distributed data structures and distributed computing utilities. This provides powerful ways of accessing distributed clustered memory, but also CPUs for true distributed computing.
Hazelcast's Distinctive Strengths
Finally, Hazelcast has a vibrant open source community enabling it to be continuously developed.
Hazelcast is a fit when you need:
It is more than simple to start enjoying Hazelcast:
Download hazelcast-<
version>.zip
from www.hazelcast.org.
Unzip hazelcast-<
version>.zip
file.
Add hazelcast-<
version>.jar
file into your classpath.
That is all.
Alternatively, Hazelcast can be found in standard Maven repositories. So, if your project uses Maven, you do not need to add additional repositories to your pom.xml
. Just add the following lines to pom.xml
:
<dependencies>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast</artifactId>
<version>3.3</version>
</dependency>
</dependencies>
There are two Maven repositories defined for Hazelcast Enterprise:
<repository>
<id>Hazelcast Private Snapshot Repository</id>
<url>https://repository-hazelcast-l337.forge.cloudbees.com/snapshot/</url>
</repository>
<repository>
<id>Hazelcast Private Release Repository</id>
<url>https://repository-hazelcast-l337.forge.cloudbees.com/release/</url>
</repository>
Hazelcast Enterprise customers may also define dependencies, a sample of which is shown below.
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-enterprise-tomcat6</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-enterprise-tomcat7</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-enterprise</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-enterprise-all</artifactId>
<version>${project.version}</version>
</dependency>
To be able to use Hazelcast Enterprise, you need to set license the key in configuration.
<hazelcast>
...
<license-key>HAZELCAST_ENTERPRISE_LICENSE_KEY</license-key>
...
</hazelcast>
Config config = new Config();
config.setLicenseKey( "HAZELCAST_ENTERPRISE_LICENSE_KEY" );
<hz:config>
...
<hz:license-key>HAZELCAST_ENTERPRISE_LICENSE_KEY</hz:license-key>
...
</hz:config>
-Dhazelcast.enterprise.license.key=HAZELCAST_ENTERPRISE_LICENSE_KEY
Having hazelcast-
<version>
.jar
added to your classpath, it is time to get started.
In this short tutorial, we will:
Let's begin.
customers
map and queue.import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import java.util.Map;
import java.util.Queue;
public class GettingStarted {
public static void main( String[] args ) {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Map<Integer, String> customers = hazelcastInstance.getMap( "customers" );
customers.put( 1, "Joe" );
customers.put( 2, "Ali" );
customers.put( 3, "Avi" );
System.out.println( "Customer with key 1: " + customers.get(1) );
System.out.println( "Map Size:" + hazelcastInstance.size() );
Queue<String> queueCustomers = hazelcastInstance.getQueue( "customers" );
queueCustomers.offer( "Tom" );
queueCustomers.offer( "Mary" );
queueCustomers.offer( "Jane" );
System.out.println( "First customer: " + queueCustomers.poll() );
System.out.println( "Second customer: "+ queueCustomers.peek() );
System.out.println( "Queue size: " + queueCustomers.size() );
}
}
Members [2] {
Member [<IP_ADDRESS>:5701]
Member [<IP_ADDRESS>:5702] this
}
Now, add hazelcast-client-
<version>
.jar
to your classpath, too. This is required to be able to use a Hazelcast client.
Following code will start a Hazelcast Client, connect to our two node cluster and print the size of our customers
map.
package com.hazelcast.test;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.client.HazelcastClient;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
public class GettingStartedClient {
public static void main( String[] args ) {
ClientConfig clientConfig = new ClientConfig();
HazelcastInstance client = HazelcastClient.newHazelcastClient( clientConfig );
IMap map = client.getMap( "customers" );
System.out.println( "Map Size:" + map.size() );
}
}
Hazelcast also offers a tool, Management Center, that enables monitoring your cluster. To be able to use it, deploy the mancenter-
<version>
.war
included in the ZIP file to your web server. You can use it to monitor your maps, queues, other distributed data structures and nodes. Please see Management Center for usage explanations.
By default Hazelcast uses Multicast to discover other nodes to form a cluster. If you are working with other Hazelcast developers on the same network, you may find yourself joining their clusters using the default settings. Hazelcast provides a way to segregate clusters within the same network when using Multicast. Please see How do I create separate clusters for more information. Alternatively, if you do not wish to use the default Multicast mechanism, you can provide a fixed list of IP addresses that are allowed to join. Please see the section Configuring TCP/IP Cluster for more information.
RELATED INFORMATION
You can also check the video tutorials here.
While Hazelcast is starting up, it checks for the configuration as follows:
First, it looks for hazelcast.config
system property. If it is set, its value is used as the path. It is useful if you want to be able to change your Hazelcast configuration. This is possible because it is not embedded within the application. The config
option can be set by the below command:
- Dhazelcast.config=
<path to the hazelcast.xml>
.
The path can be a normal one or a classpath reference with the prefix CLASSPATH
.
hazelcast.xml
file in the working directory.hazelcast.xml
exists on the classpath.hazelcast-default.xml
that comes with hazelcast.jar
.When you download and unzip hazelcast-<
version>.zip
you will see the hazelcast.xml
in /bin
folder. This is the configuration XML file for Hazelcast, a part of which is shown below.
For most of the users, default configuration should be fine. If not, you can tailor this XML file according to your needs by adding/removing/modifying properties (Declarative Configuration). Please refer to Configuration Properties for details.
Besides declarative configuration, you can configure your cluster programmatically (Programmatic Configuration). Just instantiate a Config
object and add/remove/modify properties.
You can also use wildcards while configuring Hazelcast. Please refer to the section Using Wildcard for details.
RELATED INFORMATION
Please refer to Configuration chapter for more information.
Some example usages are listed below. Hazelcast can be used:
As a real time streamer for performance detection,
As a storage for session data in web applications (enables horizontal scalability of the web application).
As mentioned in the Overview section, Hazelcast offers distributed implementations of Java interfaces. Below is the full list with links to their sections in this manual.
Standard utility collections:
java.util.Map
. Operations like reading and writing from/to a Hazelcast map can be performed with the well known methods like get and put.java.util.concurrent.BlockingQueue
. You can add an item in one machine and remove it from another one.java.util.Set
. It does not allow duplicate elements and does not preserve their order.java.util.concurrent.locks.Lock
. When you lock using Hazelcast Lock, the critical section that it guards is guaranteed to be executed by only one thread in entire cluster.java.util.concurrent.Semaphore
. When performing concurrent activities semaphores offer permits to control the thread counts.java.util.concurrent.atomic.AtomicLong
. Most of AtomicLong's operations are offered but these operations involve remote calls and hence their performances differ from AtomicLong, due to being distributed.java.util.concurrent.atomic.AtomicReference
.AtomicLong.incrementAndGet()
.java.util.concurrent.CountDownLatch
. It is a gate keeper for concurrent activities, enabling the threads to wait for other threads to complete their operations.Common Features of all Hazelcast Data Structures:
Here is how you can retrieve existing data structure instances (map, queue, set, lock, topic, etc.) and how you can listen for instance events to get notified when an instance is created or destroyed.
import java.util.Collection;
import com.hazelcast.config.Config;
import com.hazelcast.core.*;
public class Sample implements DistributedObjectListener {
public static void main(String[] args) {
Sample sample = new Sample();
Config config = new Config();
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance(config);
hazelcastInstance.addDistributedObjectListener(sample);
Collection<DistributedObject> distributedObjects = hazelcastInstance.getDistributedObjects();
for (DistributedObject distributedObject : distributedObjects) {
System.out.println(distributedObject.getName() + "," + distributedObject.getId());
}
}
@Override
public void distributedObjectCreated(DistributedObjectEvent event) {
DistributedObject instance = event.getDistributedObject();
System.out.println("Created " + instance.getName() + "," + instance.getId());
}
@Override
public void distributedObjectDestroyed(DistributedObjectEvent event) {
DistributedObject instance = event.getDistributedObject();
System.out.println("Destroyed " + instance.getName() + "," + instance.getId());
}
}
Hazelcast Map (IMap
) extends the interface java.util.concurrent.ConcurrentMap
and hence java.util.Map
. In simple terms, it is the distributed implementation of Java map. And operations like reading and writing from/to a Hazelcast map can be performed with the well known methods like get and put.
Hazelcast will partition your map entries and almost evenly distribute onto all Hazelcast members. Each member carries approximately "(1/n *
total-data) + backups", n being the number of nodes in the cluster.
Just for exemplary purposes, let's create a Hazelcast instance (node) and fill a map named Capitals
with key-value pairs using the below code.
public class FillMapMember {
public static void main( String[] args ) {
HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance();
Map<String, String> capitalcities = hzInstance.getMap( "capitals" );
capitalcities.put( "1", "Tokyo" );
capitalcities.put( "2", "Paris” );
capitalcities.put( "3", "Washington" );
capitalcities.put( "4", "Ankara" );
capitalcities.put( "5", "Brussels" );
capitalcities.put( "6", "Amsterdam" );
capitalcities.put( "7", "New Delhi" );
capitalcities.put( "8", "London" );
capitalcities.put( "9", "Berlin" );
capitalcities.put( "10", "Oslo" );
capitalcities.put( "11", "Moscow" );
...
...
capitalcities.put( "120", "Stockholm" )
}
}
When you run this code, a node is created with a map whose entries are distributed across the node's partitions. See the below illustration. This is a single node cluster for now.
NOTE: Please note that some of the partitions will not contain any data entries since we only have 120 objects and the partition count is 271 by default. This count is configurable and can be changed using the system property hazelcast.partition.count
. Please see Advanced Configuration Properties.
Now, let's create a second node which will result in a cluster with 2 nodes. This is where backups of entries are created, too. Please remember the backup partitions mentioned in Hazelcast Overview section. So, run the above code again to create the second node. Below illustration shows two nodes and how the data and its backup is distributed.
As you see, when a new member joins the cluster, it takes ownership (responsibility) and load of -some- of the entire data in the cluster. Eventually, it will carry almost "(1/n *
total-data) + backups" and reduces the load on others.
HazelcastInstance::getMap
actually returns an instance of com.hazelcast.core.IMap
which extends java.util.concurrent.ConcurrentMap
interface. So methods like ConcurrentMap.putIfAbsent(key,value)
and ConcurrentMap.replace(key,value)
can be used on distributed map as shown in the example below.
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import java.util.concurrent.ConcurrentMap;
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Customer getCustomer( String id ) {
ConcurrentMap<String, Customer> customers = hazelcastInstance.getMap( "customers" );
Customer customer = customers.get( id );
if (customer == null) {
customer = new Customer( id );
customer = customers.putIfAbsent( id, customer );
}
return customer;
}
public boolean updateCustomer( Customer customer ) {
ConcurrentMap<String, Customer> customers = hazelcastInstance.getMap( "customers" );
return ( customers.replace( customer.getId(), customer ) != null );
}
public boolean removeCustomer( Customer customer ) {
ConcurrentMap<String, Customer> customers = hazelcastInstance.getMap( "customers" );
return customers.remove( customer.getId(), customer );
}
All ConcurrentMap
operations such as put
and remove
might wait if the key is locked by another thread in the local or remote JVM. But, they will eventually return with success. ConcurrentMap
operations never throw java.util.ConcurrentModificationException
.
Also see:
Hazelcast will distribute map entries onto multiple JVMs (cluster members). Each JVM holds some portion of the data but you do not want to lose data when a JVM crashes.
Distributed maps have 1 backup by default so that if a member goes down, you do not lose data. Backup operations are synchronous, so when a map.put(key, value)
returns, it is guaranteed that the entry is replicated to one other node. For the reads, it is also guaranteed that map.get(key)
returns the latest value of the entry. Consistency is strictly enforced.
To provide data safety, Hazelcast allows you to specify the number of backup copies you want to have. That way, data on a JVM will be copied onto other JVM(s). It is configured using the backup-count
property.
<hazelcast>
<map name="default">
<backup-count>1</backup-count>
</map>
</hazelcast>
When this count is 1, it means that a map entry will have its backup on another node in the cluster. If it is set as 2, then it will have its backup on two other nodes. It can be set as 0, if you do not want your entries to be backed up, e.g. if performance is more important than backing up. Maximum value for this property is 6.
Hazelcast supports both synchronous and asynchronous backups. By default, backup operations are synchronous (configured with backup-count
). In this case, backup operations block operations until backups are successfully copied to backups nodes (or deleted from backup nodes in case of remove) and acknowledgements are received. Therefore, for example, you can be sure that backups are updated before a put
operation is completed. Of course, sync backup operations have a blocking cost which may lead to latency issues.
Asynchronous backups, on the other hand, do not block operations. They are fire & forget and do not require acknowledgements (backup operations are performed at some point in time). Async backup is configured using the async-backup-count
property.
<hazelcast>
<map name="default">
<backup-count>0</backup-count>
<async-backup-count>1</async-backup-count>
</map>
</hazelcast>
NOTE: Backups increase memory usage since they are also kept in memory. So for every backup, you double the original memory consumption.
NOTE: A map can have both sync and aysnc backups at the same time.
By default, Hazelcast will have one sync backup copy. If backup count is more than 1, then each member will carry both owned entries and backup copies of other member(s). So for the map.get(key)
call, it is possible that calling member has backup copy of that key but by default, map.get(key)
will always read the value from the actual owner of the key for consistency.
It is possible to enable backup reads (read local backup entries) by setting the value of read-backup-data
property to true. Its default value is false for strong consistency. Enabling backup reads can improve the performance.
<hazelcast>
<map name="default">
<backup-count>0</backup-count>
<async-backup-count>1</async-backup-count>
<read-backup-data>true</read-backup-data>
</map>
</hazelcast>
This feature is available when there is at least 1 sync or async backup.
Unless you delete the map entries manually or use an eviction policy, they will remain in the map. Hazelcast supports policy based eviction for distributed maps. Currently supported policies are LRU (Least Recently Used) and LFU (Least Frequently Used). There are also other properties as shown in the below sample declarative configuration.
<hazelcast>
<map name="default">
...
<time-to-live-seconds>0</time-to-live-seconds>
<max-idle-seconds>0</max-idle-seconds>
<eviction-policy>LRU</eviction-policy>
<max-size policy="PER_NODE">5000</max-size>
<eviction-percentage>25</eviction-percentage>
...
</map>
</hazelcast>
Let's describe each property.
time-to-live
: Maximum time in seconds for each entry to stay in the map. If it is not 0, entries that are older than and not updated for this time are evicted automatically. Valid values are integers between 0 and Integer.MAX VALUE
. Default value is 0 and it means infinite. Moreover, if it is not 0, entries are evicted regardless of the set eviction-policy
. max-idle-seconds
: Maximum time in seconds for each entry to stay idle in the map. Entries that are idle for more than this time are evicted automatically. An entry is idle if no get
, put
or containsKey
is called. Valid values are integers between 0 and Integer.MAX VALUE
. Default value is 0 and it means infinite.eviction-policy
: Valid values are described below.
max-size
will be ignored. Of course, you still can combine it with time-to-live-seconds
and max-idle-seconds
.max-size
: Maximum size of the map. When maximum size is reached, map is evicted based on the policy defined. Valid values are integers between 0 and Integer.MAX VALUE
. Default value is 0. If you want max-size
to work, eviction-policy
property must be set to a value other than NONE. Its attributes are described below.
PER_NODE
: Maximum number of map entries in each JVM. This is the default policy.
<max-size policy="PER_NODE">5000</max-size>
PER_PARTITION
: Maximum number of map entries within each partition. Storage size depends on the partition count in a JVM. So, this attribute may not be used often. If the cluster is small it will be hosting more partitions and therefore map entries, than that of a larger cluster.
<max-size policy="PER_PARTITION">27100</max-size>
USED_HEAP_SIZE
: Maximum used heap size in megabytes for each JVM.
<max-size policy="USED_HEAP_SIZE">4096</max-size>
USED_HEAP_PERCENTAGE
: Maximum used heap size percentage for each JVM. If, for example, JVM is configured to have 1000 MB and this value is 10, then the map entries will be evicted when used heap size exceeds 100 MB.
<max-size policy="USED_HEAP_PERCENTAGE">10</max-size>
eviction-percentage
: When max-size
is reached, specified percentage of the map will be evicted. If 25 is set for example, 25% of the entries will be evicted. Setting this property to a smaller value will cause eviction of small number of map entries. So, if map entries are inserted frequently, smaller percentage values may lead to overheads. Valid values are integers between 0 and 100. Default value is 25.
<map name="documents">
<max-size policy="PER_NODE">10000</max-size>
<eviction -policy>LRU</eviction -policy>
<max-idle-seconds>60</max-idle-seconds>
</map>
In the above sample, documents
map starts to evict its entries from a member when the map size exceeds 10000 in that member. Then, the entries least recently used will be evicted. And, the entries not used for more than 60 seconds will be evicted as well.
Above explained eviction policies and configurations apply to all the entries of a map. The entries that meet the specified eviction conditions are evicted.
But, you may particularly want to evict some specific map entries. In this case, you can use the ttl
and timeunit
parameters of the method map.put()
. A sample code line is given below.
myMap.put( "1", "John", 50, TimeUnit.SECONDS )
So, the map entry with the key "1" will be evicted in 50 seconds after it is put into myMap
.
The method evictAll()
is developed for evicting all keys from the map except the locked ones. If a MapStore is defined for the map, deleteAll
is not called by evictAll
. If you want to call the method deleteAll
, use clear()
.
A sample is given below.
public class EvictAll {
public static void main(String[] args) {
final int numberOfKeysToLock = 4;
final int numberOfEntriesToAdd = 1000;
HazelcastInstance node1 = Hazelcast.newHazelcastInstance();
HazelcastInstance node2 = Hazelcast.newHazelcastInstance();
IMap<Integer, Integer> map = node1.getMap(EvictAll.class.getCanonicalName());
for (int i = 0; i < numberOfEntriesToAdd; i++) {
map.put(i, i);
}
for (int i = 0; i < numberOfKeysToLock; i++) {
map.lock(i);
}
// should keep locked keys and evict all others.
map.evictAll();
System.out.printf("# After calling evictAll...\n");
System.out.printf("# Expected map size\t: %d\n", numberOfKeysToLock);
System.out.printf("# Actual map size\t: %d\n", map.size());
}
}
NOTE: Only EVICT_ALL event is fired for any registered listeners.
IMap has in-memory-format
configuration option. By default, Hazelcast stores data into memory in binary (serialized) format. But sometimes, it can be efficient to store the entries in their object form, especially in cases of local processing like entry processor and queries. Setting in-memory-format
in map's configuration, you can decide how the data will be stored in memory. There are below options.
BINARY
(default): This is the default option. The data will be stored in serialized binary format. You can use this option if you mostly perform regular map operations like put and get.
OBJECT
: The data will be stored in deserialized form. This configuration is good for maps where entry processing and queries form the majority of all operations and the objects are complex ones, so serialization cost is respectively high. By storing objects, entry processing will not contain the deserialization cost.
Regular operations like get
rely on the object instance. When OBJECT format is used and, for example, when a get
is performed, the map does not return the stored instance, but creates a clone. So, this whole get
operation includes a serialization first (on the node owning the instance) and then deserialization (on the node calling the instance). But, when BINARY format is used, only a deserialization is required and this is faster.
Similarly, put
operation is faster when BINARY format is used. If it was OBJECT, map would create a clone of the instance. So, there would first a serialization and then deserialization. Again, when BINARY is used, only a deserialization is needed.
NOTE: If a value is stored in OBJECT
format, a change on a returned value does not effect the stored instance. In this case, the returned instance is not the actual one but a clone. Therefore, changes made on an object after it is returned will not reflect on the actual stored data. Similarly, when a value is written to a map and the value is stored in OBJECT
format, it will be a copy of the put value. So changes made on the object after it is stored, will not reflect on the actual stored data.
Hazelcast allows you to load and store the distributed map entries from/to a persistent data store such as a relational database. For these, you can use Hazelcast's MapStore
and MapLoader
interfaces.
When you provide a MapLoader
implementation and request an entry (IMap.get()
) that does not exist in the memory, MapLoader
's load
or loadAll
methods will load that entry from the data store. This loaded entry is placed into the map and will stay there until it is removed or evicted.
When a MapStore
implementation is provided, an entry is put also into a user defined data store.
NOTE: Data store needs to be a centralized system that is accessible from all Hazelcast Nodes. Persisting to local file system is not supported.
Please see the below example.
public class PersonMapStore implements MapStore<Long, Person> {
private final Connection con;
public PersonMapStore() {
try {
con = DriverManager.getConnection("jdbc:hsqldb:mydatabase", "SA", "");
con.createStatement().executeUpdate(
"create table if not exists person (id bigint, name varchar(45))");
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
public synchronized void delete(Long key) {
System.out.println("Delete:" + key);
try {
con.createStatement().executeUpdate(
format("delete from person where id = %s", key));
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
public synchronized void store(Long key, Person value) {
try {
con.createStatement().executeUpdate(
format("insert into person values(%s,'%s')", key, value.name));
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
public synchronized void storeAll(Map<Long, Person> map) {
for (Map.Entry<Long, Person> entry : map.entrySet())
store(entry.getKey(), entry.getValue());
}
public synchronized void deleteAll(Collection<Long> keys) {
for (Long key : keys) delete(key);
}
public synchronized Person load(Long key) {
try {
ResultSet resultSet = con.createStatement().executeQuery(
format("select name from person where id =%s", key));
try {
if (!resultSet.next()) return null;
String name = resultSet.getString(1);
return new Person(name);
} finally {
resultSet.close();
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
public synchronized Map<Long, Person> loadAll(Collection<Long> keys) {
Map<Long, Person> result = new HashMap<Long, Person>();
for (Long key : keys) result.put(key, load(key));
return result;
}
public Set<Long> loadAllKeys() {
return null;
}
}
NOTE: Loading process is performed on a thread different than the partition threads using ExecutorService.
RELATED INFORMATION
For more MapStore/MapLoader code samples please see here.
Hazelcast supports read-through, write-through and write-behind persistence modes which are explained in below subsections.
If an entry does not exist in the memory when an application asks, Hazelcast asks your loader implementation to load that entry from the data store. If the entry exists there, the loader implementation gets it, hands it to Hazelcast, and Hazelcast puts it into the memory. This is read-through persistence mode.
MapStore
can be configured as write-through by setting the write-delay-seconds
property to 0. This means the entries will be put to the data store synchronously.
In this mode, when the map.put(key,value)
call returns, you can be sure that
MapStore.store(key,value)
is successfully called so the entry is persisted.
In-Memory entry is updated
In-Memory backup copies are successfully created on other JVMs (if backup-count
is greater than 0)
Same behavior goes for the map.remove(key)
, only difference is that MapStore.delete(key)
is called when it will be deleted.
If MapStore
throws an exception, then the exception will be propagated back to the original put
or remove
call in the form of RuntimeException
.
MapStore
can be configured as write-behind by setting the write-delay-seconds
property to a value bigger than 0. This means the modified entries will be put to the data store asynchronously after a configured delay.
NOTE: In write-behind mode, Hazelcast coalesces updates on a specific key, i.e. applies only the last update on it.
In this mode, when the map.put(key,value)
call returns, you can be sure that
In-Memory entry is updated
In-Memory backup copies are successfully created on other JVMs (if backup-count
is greater than 0)
The entry is marked as dirty so that after write-delay-seconds
, it can be persisted with MapStore.store(key,value)
call.
Same behavior goes for the map.remove(key)
, only difference is that MapStore.delete(key)
is called when it will be deleted.
If MapStore
throws an exception, then Hazelcast retries to store the entry. If it still cannot be stored, a log message is printed and the entry is re-queued.
For batch write operations, which are only allowed in write-behind mode, Hazelcast will call MapStore.storeAll(map)
, and MapStore.deleteAll(collection)
to do all writes in a single call.
NOTE: If a map entry is marked as dirty, i.e. it is waiting to be persisted to the MapStore
in a write-behind scenario, the eviction process forces the entry to be stored. By this way, you will have control on the number of entries waiting to be stored, so that a possible OutOfMemory exception can be prevented.
NOTE: MapStore or MapLoader implementations should not use Hazelcast Map/Queue/MultiMap/List/Set operations. Your implementation should only work with your data store. Otherwise, you may get into deadlock situations.
Here is a sample configuration:
<hazelcast>
...
<map name="default">
...
<map-store enabled="true">
<!--
Name of the class implementing MapLoader and/or MapStore.
The class should implement at least of these interfaces and
contain no-argument constructor. Note that the inner classes are not supported.
-->
<class-name>com.hazelcast.examples.DummyStore</class-name>
<!--
Number of seconds to delay to call the MapStore.store(key, value).
If the value is zero then it is write-through so MapStore.store(key, value)
will be called as soon as the entry is updated.
Otherwise it is write-behind so updates will be stored after write-delay-seconds
value by calling Hazelcast.storeAll(map). Default value is 0.
-->
<write-delay-seconds>60</write-delay-seconds>
<!--
Used to create batch chunks when writing map store.
In default mode all entries will be tried to persist in one go.
To create batch chunks, minimum meaningful value for write-batch-size
is 2. For values smaller than 2, it works as in default mode.
-->
<write-batch-size>1000</write-batch-size>
</map-store>
</map>
</hazelcast>
As you know, a configuration can be applied to more than one map using wildcards (Please see Using Wildcard), meaning the configuration is shared among the maps. But, MapStore
does not know which entries to be stored when there is one configuration applied to multiple maps. To overcome this, Hazelcast provides MapStoreFactory
interface.
Using this factory, MapStore
s for each map can be created, when a wildcard configuration is used. A sample code is given below.
Config config = new Config();
MapConfig mapConfig = config.getMapConfig( "*" );
MapStoreConfig mapStoreConfig = mapConfig.getMapStoreConfig();
mapStoreConfig.setFactoryImplementation( new MapStoreFactory<Object, Object>() {
@Override
public MapLoader<Object, Object> newMapStore( String mapName, Properties properties ) {
return null;
}
});
Moreover, if the configuration implements MapLoaderLifecycleSupport
interface, then the user will have the control to initialize the MapLoader
implementation with the given map name, configuration properties and the Hazelcast instance. See the below code portion.
public interface MapLoaderLifecycleSupport {
/**
* Initializes this MapLoader implementation. Hazelcast will call
* this method when the map is first used on the
* HazelcastInstance. Implementation can
* initialize required resources for the implementing
* mapLoader such as reading a config file and/or creating
* database connection.
*/
void init( HazelcastInstance hazelcastInstance, Properties properties, String mapName );
/**
* Hazelcast will call this method before shutting down.
* This method can be overridden to cleanup the resources
* held by this map loader implementation, such as closing the
* database connections etc.
*/
void destroy();
}
MapLoader.loadAllKeys
API is used for pre-populating the in-memory map when the map is first touched/used. If MapLoader.loadAllKeys
returns NULL then nothing will be loaded. Your MapLoader.loadAllKeys
implementation can return all or some of the keys. You may select and return only the hot
keys, for instance. Also note that this is the fastest way of pre-populating the map as Hazelcast will optimize the loading process by having each node loading owned portion of the entries.
Moreover, there is InitialLoadMode configuration parameter in the class MapStoreConfig
class. This parameter has two values: LAZY and EAGER. If InitialLoadMode is set as LAZY, data is not loaded during the map creation. If it is set as EAGER, whole data is loaded while the map is being created and everything becomes ready to use. Also, if you add indices to your map by MapIndexConfig
class or addIndex
method, then InitialLoadMode is overridden and MapStoreConfig behaves as if EAGER mode is on.
Here is MapLoader initialization flow:
getMap()
is first called from any node, initialization will start depending on the the value of InitialLoadMode. If it is set as EAGER, initialization starts. If it is set as LAZY, initialization actually does not start but data is loaded at each time a partition loading is completed.MapLoader.loadAllKeys()
to get all your keys on each nodeMapLoader.loadAll(keys)
IMap.putTransient(key,value)
NOTE: If the load mode is LAZY and when clear()
method is called (which triggers MapStore.deleteAll()
), Hazelcast will remove ONLY the loaded entries from your map and datastore. Since the whole data is not loaded for this case (LAZY mode), please note that there may be still entries in your datastore.
The method loadAll
is developed to load some or all keys into a data store in order to optimize the multiple load operations. The method has two signatures (i.e. same method can take two different parameter lists). One loads the given keys and the other loads all keys. Please see the sample code below.
public class LoadAll {
public static void main(String[] args) {
final int numberOfEntriesToAdd = 1000;
final String mapName = LoadAll.class.getCanonicalName();
final Config config = createNewConfig(mapName);
final HazelcastInstance node = Hazelcast.newHazelcastInstance(config);
final IMap<Integer, Integer> map = node.getMap(mapName);
populateMap(map, numberOfEntriesToAdd);
System.out.printf("# Map store has %d elements\n", numberOfEntriesToAdd);
map.evictAll();
System.out.printf("# After evictAll map size\t: %d\n", map.size());
map.loadAll(true);
System.out.printf("# After loadAll map size\t: %d\n", map.size());
}
}
In some scenarios, you may need to modify the object after storing it into the map store.
For example, you can get ID or version auto generated by your database and you need to modify your object stored in distributed map, not to break the sync between database and data grid. You can do that by implementing PostProcessingMapStore
interface;
so the modified object will be put to the distributed map. That will cause an extra step of Serialization
, so use it just when needed (This explanation is only valid when using write-through
map store configuration).
Here is an example of post processing map store:
class ProcessingStore extends MapStore<Integer, Employee> implements PostProcessingMapStore {
@Override
public void store( Integer key, Employee employee ) {
EmployeeId id = saveEmployee();
employee.setId( id.getId() );
}
}
Map entries in Hazelcast are partitioned across the cluster. Imagine that you are reading key k
so many times and k
is owned by another member in your cluster. Each map.get(k)
will be a remote operation, meaning lots of network trips. If you have a map that is read-mostly, then you should consider creating a Near Cache for the map so that reads can be much faster and consume less network traffic. All these benefits do not come free. When using near cache, you should consider the following issues:
JVM will have to hold extra cached data so it will increase the memory consumption.
If invalidation is turned on and entries are updated frequently, then invalidations will be costly.
Near cache breaks the strong consistency guarantees; you might be reading stale data.
Near cache is highly recommended for the maps that are read-mostly. Here is a near cache configuration for a map:
<hazelcast>
...
<map name="my-read-mostly-map">
...
<near-cache>
<!--
Maximum size of the near cache. When max size is reached,
cache is evicted based on the policy defined.
Any integer between 0 and Integer.MAX_VALUE. 0 means
Integer.MAX_VALUE. Default is 0.
-->
<max-size>5000</max-size>
<!--
Maximum number of seconds for each entry to stay in the near cache. Entries that are
older than <time-to-live-seconds> will get automatically evicted from the near cache.
Any integer between 0 and Integer.MAX_VALUE. 0 means infinite. Default is 0.
-->
<time-to-live-seconds>0</time-to-live-seconds>
<!--
Maximum number of seconds each entry can stay in the near cache as untouched (not-read).
Entries that are not read (touched) more than <max-idle-seconds> value will get removed
from the near cache.
Any integer between 0 and Integer.MAX_VALUE. 0 means
Integer.MAX_VALUE. Default is 0.
-->
<max-idle-seconds>60</max-idle-seconds>
<!--
Valid values are:
NONE (no extra eviction, <time-to-live-seconds> may still apply),
LRU (Least Recently Used),
LFU (Least Frequently Used).
NONE is the default.
Regardless of the eviction policy used, <time-to-live-seconds> will still apply.
-->
<eviction-policy>LRU</eviction-policy>
<!--
Should the cached entries get evicted if the entries are changed (updated or removed).
true of false. Default is true.
-->
<invalidate-on-change>true</invalidate-on-change>
<!--
You may want also local entries to be cached.
This is useful when in memory format for near cache is different than the map's one.
By default it is disabled.
-->
<cache-local-entries>false</cache-local-entries>
</near-cache>
</map>
</hazelcast>
NOTE: Programmatically, near cache configuration is done by using the class NearCacheConfig. And this class is used both in nodes and clients. To create a near cache in a client (native Java client), use the method addNearCacheConfig
in the class ClientConfig
(please see Java Client section). Please note that near cache configuration is specific to the node or client itself, a map in a node may not have near cache configured while the same map in a client may have.
Hazelcast Distributed Map (IMap) is thread-safe and meets your thread safety requirements. When these requirements increase or you want to have more control on the concurrency, below features and solutions provided by Hazelcast can be considered.
Let's work on a sample case as shown below.
public class RacyUpdateMember {
public static void main( String[] args ) throws Exception {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IMap<String, Value> map = hz.getMap( "map" );
String key = "1";
map.put( key, new Value() );
System.out.println( "Starting" );
for ( int k = 0; k < 1000; k++ ) {
if ( k % 100 == 0 ) System.out.println( "At: " + k );
Value value = map.get( key );
Thread.sleep( 10 );
value.amount++;
map.put( key, value );
}
System.out.println( "Finished! Result = " + map.get(key).amount );
}
static class Value implements Serializable {
public int amount;
}
}
If the above code is run by more than one cluster members simultaneously, there will be likely a race condition.
One usual way to solve this race issue is using the lock mechanism provided by Hazelcast distributed map, i.e. map.lock
and map.unlock
methods. You simply lock the entry until you finished with it. See the below sample code.
public class PessimisticUpdateMember {
public static void main( String[] args ) throws Exception {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IMap<String, Value> map = hz.getMap( "map" );
String key = "1";
map.put( key, new Value() );
System.out.println( "Starting" );
for ( int k = 0; k < 1000; k++ ) {
map.lock( key );
try {
Value value = map.get( key );
Thread.sleep( 10 );
value.amount++;
map.put( key, value );
} finally {
map.unlock( key );
}
}
System.out.println( "Finished! Result = " + map.get( key ).amount );
}
static class Value implements Serializable {
public int amount;
}
}
The IMap lock will automatically be collected by the garbage collector when the map entry is removed.
The IMap lock is reentrant, but it does not support fairness.
Another way can be acquiring a predictable Lock
object from Hazelcast. By this way, every value in the map can be given a lock or you can create a stripe of locks.
Hazelcast way of optimistic locking is to use map.replace
method. See the below sample code.
public class OptimisticMember {
public static void main( String[] args ) throws Exception {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IMap<String, Value> map = hz.getMap( "map" );
String key = "1";
map.put( key, new Value() );
System.out.println( "Starting" );
for ( int k = 0; k < 1000; k++ ) {
if ( k % 10 == 0 ) System.out.println( "At: " + k );
for (; ; ) {
Value oldValue = map.get( key );
Value newValue = new Value( oldValue );
Thread.sleep( 10 );
newValue.amount++;
if ( map.replace( key, oldValue, newValue ) )
break;
}
}
System.out.println( "Finished! Result = " + map.get( key ).amount );
}
static class Value implements Serializable {
public int amount;
public Value() {
}
public Value( Value that ) {
this.amount = that.amount;
}
public boolean equals( Object o ) {
if ( o == this ) return true;
if ( !( o instanceof Value ) ) return false;
Value that = ( Value ) o;
return that.amount == this.amount;
}
}
}
NOTE: Above sample code is intentionally broken.
Depending on the locking requirements, one locking strategy can be picked.
Optimistic locking is better for mostly read only systems and it brings a performance boost over pessimistic locking.
Pessimistic locking is good if there are lots of updates on the same key and it is more robust than optimistic one from the perspective of data consistency.
In Hazelcast, use IExecutorService
for submitting a task to a key owner or to a member, or members. This is the recommended way of task executions which uses pessimistic or optimistic locking techniques. By following this manner, there will be less network hops and less data over wire and also tasks will be executed very near to data. Please refer to Data Affinity.
ABA problem occurs in environments when a shared resource is open to change by multiple threads. So, even one thread sees the same value for a particular key in consecutive reads, it does not mean nothing has changed between the reads. Because one another thread may come and change the value, do another work and change the value back, but the first thread can think that nothing has changed.
To prevent these kind of problems, one possible solution is to use a version number and to check it before any write to be sure that nothing has changed between consecutive reads. Although all the other fields will be equal, the version field will prevent objects from being seen as equal. This is called the optimistic locking strategy and it is used in environments which do not expect intensive concurrent changes on a specific key.
In Hazelcast, you can apply optimistic locking strategy by using replace
method of map. This method compares values in object or data forms depending on the in memory format configuration. If the values are equal, it replaces the old value with the new one. If you want to use your defined equals
method, in memory format should be Object
. Otherwise, Hazelcast serializes objects to binary forms and compares them.
Hazelcast keeps extra information about each map entry such as creation time, last update time, last access time, number of hits, version, and this information is exposed to the developer via IMap.getEntryView(key)
call. Here is an example:
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.EntryView;
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
EntryView entry = hz.getMap( "quotes" ).getEntryView( "1" );
System.out.println ( "size in memory : " + entry.getCost() );
System.out.println ( "creationTime : " + entry.getCreationTime() );
System.out.println ( "expirationTime : " + entry.getExpirationTime() );
System.out.println ( "number of hits : " + entry.getHits() );
System.out.println ( "lastAccessedTime: " + entry.getLastAccessTime() );
System.out.println ( "lastUpdateTime : " + entry.getLastUpdateTime() );
System.out.println ( "version : " + entry.getVersion() );
System.out.println ( "key : " + entry.getKey() );
System.out.println ( "value : " + entry.getValue() );
Map entry events can be listened. Hazelcast distributed map offers the method addEntryListener
to add an entry listener to the map and listen to the entry events.
Let's take a look at the below sample code.
public class Listen {
public static void main( String[] args ) {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IMap<String, String> map = hz.getMap( "somemap" );
map.addEntryListener( new MyEntryListener(), true );
System.out.println( "EntryListener registered" );
}
static class MyEntryListener implements EntryListener<String, String> {
@Override
public void entryAdded( EntryEvent<String, String> event ) {
System.out.println( "Entry Added:" + event );
}
@Override
public void entryRemoved( EntryEvent<String, String> event ) {
System.out.println( "Entry Removed:" + event );
}
@Override
public void entryUpdated( EntryEvent<String, String> event ) {
System.out.println( "Entry Updated:" + event );
}
@Override
public void entryEvicted( EntryEvent<String, String> event ) {
System.out.println( "Entry Evicted:" + event );
}
@Override
public void mapEvicted( MapEvent event ) {
System.out.println( "Map Evicted:" + event );
}
@Override
public void mapCleared( MapEvent event ) {
System.out.println( "Map Cleared:" + event );
}
}
}
And, now let's perform some modifications on the map entries using the below sample code.
public class Modify {
public static void main( String[] args ) {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IMap<String, String> map = hz.getMap( "somemap");
String key = "" + System.nanoTime();
String value = "1";
map.put( key, value );
map.put( key, "2" );
map.delete( key );
}
}
Now, let's first execute the class Listen
and then execute Modify
. Check out the below output produced by Listen
.
entryAdded:EntryEvent {Address[<IP_ADDRESS>]:5702} key=251359212222282,
oldValue=null, value=1, event=ADDED, by Member [<IP_ADDRESS>]:5702
entryUpdated:EntryEvent {Address[<IP_ADDRESS>]:5702} key=251359212222282,
oldValue=1, value=2, event=UPDATED, by Member [<IP_ADDRESS>]:5702
entryRemoved:EntryEvent {Address[<IP_ADDRESS>]:5702} key=251359212222282,
oldValue=2, value=2, event=REMOVED, by Member [<IP_ADDRESS>]:5702
Entry Listener runs on event threads which are also used by other listeners (e.g. collection listeners, pub/sub message listeners, etc.). This means entry listeners can access to other partitions. So, consideration should be given when running long tasks since listening to those tasks may cause other event listeners to starve.
You can add intercept operations and then execute your own business logic synchronously blocking the operations. You can change the returned value from a get
operation, change the value to be put
or cancel
operations by throwing an exception.
Interceptors are different from listeners. With listeners, you take an action after the operation has been completed. Interceptor actions are synchronous and you can alter the behavior of operation, change the values, or totally cancel it.
Map interceptors are chained, so adding the same interceptor multiple times to the same map can result in duplicate effects. This can easily happen when the interceptor is added to the map at node initialization, so that each node adds the same interceptor. When adding the interceptor in this way, be sure that the hashCode()
method is implemented to return the same value for every instance of the interceptor. It is not strictly necessary, but it is a good idea to also implement equals()
as this will ensure that the map interceptor can be removed reliably.
IMap API has two methods for adding and removing an interceptor to the map,addInterceptor
and removeInterceptor
:
/**
* Adds an interceptor for this map. Added interceptor will intercept operations
* and execute user defined methods and will cancel operations if user defined method throw exception.
*
*
* @param interceptor map interceptor
* @return id of registered interceptor
*/
String addInterceptor( MapInterceptor interceptor );
/**
* Removes the given interceptor for this map. So it will not intercept operations anymore.
*
*
* @param id registration id of map interceptor
*/
void removeInterceptor( String id );
Here is the MapInterceptor
interface:
public interface MapInterceptor extends Serializable {
/**
* Intercept the get operation before it returns a value.
* Return another object to change the return value of get(..)
* Returning null will cause the get(..) operation to return the original value,
* namely return null if you do not want to change anything.
*
*
* @param value the original value to be returned as the result of get(..) operation
* @return the new value that will be returned by get(..) operation
*/
Object interceptGet( Object value );
/**
* Called after get(..) operation is completed.
*
*
* @param value the value returned as the result of get(..) operation
*/
void afterGet( Object value );
/**
* Intercept put operation before modifying map data.
* Return the object to be put into the map.
* Returning null will cause the put(..) operation to operate as expected,
* namely no interception. Throwing an exception will cancel the put operation.
*
*
* @param oldValue the value currently in map
* @param newValue the new value to be put
* @return new value after intercept operation
*/
Object interceptPut( Object oldValue, Object newValue );
/**
* Called after put(..) operation is completed.
*
*
* @param value the value returned as the result of put(..) operation
*/
void afterPut( Object value );
/**
* Intercept remove operation before removing the data.
* Return the object to be returned as the result of remove operation.
* Throwing an exception will cancel the remove operation.
*
*
* @param removedValue the existing value to be removed
* @return the value to be returned as the result of remove operation
*/
Object interceptRemove( Object removedValue );
/**
* Called after remove(..) operation is completed.
*
*
* @param value the value returned as the result of remove(..) operation
*/
void afterRemove( Object value );
}
Example Usage:
public class InterceptorTest {
@Test
public void testMapInterceptor() throws InterruptedException {
HazelcastInstance hazelcastInstance1 = Hazelcast.newHazelcastInstance();
HazelcastInstance hazelcastInstance2 = Hazelcast.newHazelcastInstance();
IMap<Object, Object> map = hazelcastInstance1.getMap( "testMapInterceptor" );
SimpleInterceptor interceptor = new SimpleInterceptor();
map.addInterceptor( interceptor );
map.put( 1, "New York" );
map.put( 2, "Istanbul" );
map.put( 3, "Tokyo" );
map.put( 4, "London" );
map.put( 5, "Paris" );
map.put( 6, "Cairo" );
map.put( 7, "Hong Kong" );
try {
map.remove( 1 );
} catch ( Exception ignore ) {
}
try {
map.remove( 2 );
} catch ( Exception ignore ) {
}
assertEquals( map.size(), 6) ;
assertEquals( map.get( 1 ), null );
assertEquals( map.get( 2 ), "ISTANBUL:" );
assertEquals( map.get( 3 ), "TOKYO:" );
assertEquals( map.get( 4 ), "LONDON:" );
assertEquals( map.get( 5 ), "PARIS:" );
assertEquals( map.get( 6 ), "CAIRO:" );
assertEquals( map.get( 7 ), "HONG KONG:" );
map.removeInterceptor( interceptor );
map.put( 8, "Moscow" );
assertEquals( map.get( 8 ), "Moscow" );
assertEquals( map.get( 1 ), null );
assertEquals( map.get( 2 ), "ISTANBUL" );
assertEquals( map.get( 3 ), "TOKYO" );
assertEquals( map.get( 4 ), "LONDON" );
assertEquals( map.get( 5 ), "PARIS" );
assertEquals( map.get( 6 ), "CAIRO" );
assertEquals( map.get( 7 ), "HONG KONG" );
}
static class SimpleInterceptor implements MapInterceptor, Serializable {
@Override
public Object interceptGet( Object value ) {
if (value == null)
return null;
return value + ":";
}
@Override
public void afterGet( Object value ) {
}
@Override
public Object interceptPut( Object oldValue, Object newValue ) {
return newValue.toString().toUpperCase();
}
@Override
public void afterPut( Object value ) {
}
@Override
public Object interceptRemove( Object removedValue ) {
if(removedValue.equals( "ISTANBUL" ))
throw new RuntimeException( "you can not remove this" );
return removedValue;
}
@Override
public void afterRemove( Object value ) {
// do something
}
}
}
Hazelcast distributed queue is an implementation of java.util.concurrent.BlockingQueue
. Being distributed, it enables all cluster members to interact with it. Meaning that, you can add an item in one machine and remove it from another one.
import com.hazelcast.core.Hazelcast;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
BlockingQueue<MyTask> queue = hazelcastInstance.getQueue( "tasks" );
queue.put( new MyTask() );
MyTask task = queue.take();
boolean offered = queue.offer( new MyTask(), 10, TimeUnit.SECONDS );
task = queue.poll( 5, TimeUnit.SECONDS );
if ( task != null ) {
//process task
}
FIFO ordering will apply to all queue operations cluster wide. User objects (such as MyTask
in the example above), that are (en/de)queued have to be Serializable
.
There is no batching while iterating over Queue. All items will be copied to local and iteration will occur locally.
Below sample codes illustrate a producer and consumer connected by a distributed queue.
Let's put one integer at each second on a queue, 100 integers in total.
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
public class ProducerMember {
public static void main( String[] args ) throws Exception {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IQueue<Integer> queue = hz.getQueue( "queue" );
for ( int k = 1; k < 100; k++ ) {
queue.put( k );
System.out.println( "Producing: " + k );
Thread.sleep(1000);
}
queue.put( -1 );
System.out.println( "Producer Finished!" );
}
}
Producer
puts a -1 on the queue to show that put
's are finished. Now, let's create a Consumer
class that take a message from this queue, as shown below.
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
public class ConsumerMember {
public static void main( String[] args ) throws Exception {
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
IQueue<Integer> queue = hz.getQueue( "queue" );
while ( true ) {
int item = queue.take();
System.out.println( "Consumed: " + item );
if ( item == -1 ) {
queue.put( -1 );
break;
}
Thread.sleep( 5000 );
}
System.out.println( "Consumer Finished!" );
}
}
As seen in the above sample code, Consumer
waits 5 seconds before it consumes the next message. It stops once it receives -1. Also note that, Consumer
puts -1 back on the queue before the loop is ended.
When you start first the Producer
and then the Consumer
, items produced on the queue will be consumed from that same queue.
From the above codes, you can see that item production is done at every second, and the consumption is performed at every 5 seconds. So, it can be realized that the consumer keeps growing. To balance the produce/consume operation, let's start another consumer. By this way, consumption is distributed to these two consumers, see the sample outputs below.
Once the second consumer is started after a while, first consumer output:
...
Consumed 13
Consumed 15
Consumer 17
...
Second consumer output:
...
Consumed 14
Consumed 16
Consumer 18
...
In the case of a lot of producers and consumers for the queue, using a list of queues may solve the queue bottlenecks. Of course, in this case, you should be aware that ordering of messages being sent to different queues is not guaranteed. But, since in most cases strict ordering is not important, list of queues would be a good solution.
NOTE: The items are taken from the queue in the same order they were put. However, if there are more than one consumers, this ordering is not guaranteed.
A bounded queue is a queue with a limited capacity. When the bounded queue is full, no more items can be put into the queue until some items are taken out.
A Hazelcast distributed queue can be turned into a bounded queue by setting the capacity limit using the max-size
property.
Queue capacity can be set using the max-size
property in the configuration, as shown below. max-size
specifies the maximum size of the queue. Once the queue size reaches this value, put
operations will be blocked until the queue size goes below max-size
, that happens when a consumer removes items from the queue.
Let's set 10 as the maximum size of our sample queue in the Sample Queue Code.
<hazelcast>
...
<queue name="queue">
<max-size>10</max-size>
</queue>
...
</hazelcast>
When the producer is started, 10 items are put into the queue and then the queue will not allow more put
operations. When the consumer is started, it will remove items from the queue. This means that the producer can put
more items into the queue until there are 10 items in the queue again, at which point put
operation again become blocked.
But in this sample code, the producer is 5 times faster than the consumer. It will effectively always be waiting for the consumer to remove items before it can put more on the queue. For this sample code, if maximum throughput was the goal, it would be a good option to start multiple consumers to prevent the queue from filling up.
Hazelcast allows you to load and store the distributed queue entries from/to a persistent datastore using the interface QueueStore
. If queue store is enabled, each entry added to queue will also be stored at the configured queue store. When the number of items in queue exceeds the memory limit, items will only persisted to queue store, they will not be stored in queue memory.
QueueStore
interface enables you store, load and delete entries with its methods like store
, storeAll
, load
and delete
. Below sample class includes all the QueueStore
methods.
public class TheQueueStore implements QueueStore<Item> {
@Override
public void delete(Long key) {
System.out.println("delete");
}
@Override
public void store(Long key, Item value) {
System.out.println("store");
}
@Override
public void storeAll(Map<Long, Item> map) {
System.out.println("store all");
}
@Override
public void deleteAll(Collection<Long> keys) {
System.out.println("deleteAll");
}
@Override
public Item load(Long key) {
System.out.println("load");
return null;
}
@Override
public Map<Long, Item> loadAll(Collection<Long> keys) {
System.out.println("loadAll");
return null;
}
@Override
public Set<Long> loadAllKeys() {
System.out.println("loadAllKeys");
return null;
}
As you can guess, Item
must be serializable. And below is a sample queue store configuration.
<queue-store>
<class-name>com.hazelcast.QueueStoreImpl</class-name>
<properties>
<property name="binary">false</property>
<property name="memory-limit">10000</property>
<property name="bulk-load">500</property>
</properties>
</queue-store>
Let's explain the properties.
Binary: By default, Hazelcast stores queue items in serialized form in memory and before inserting into datastore, deserializes them. But if you will not reach the queue store from an external application, you can prefer the items to be inserted in binary form. So you get rid of de-serialization step which is a performance optimization. Binary feature is disabled by default.
Memory Limit:
This is the number of items after which Hazelcast will just store items to datastore. For example, if memory limit is 1000, then 1001st item will be just put into datastore. This feature is useful when you want to avoid out-of-memory conditions. Default number for memory limit is 1000. If you want to always use memory, you can set it to Integer.MAX_VALUE
.
Bulk Load: At initialization of queue, items are loaded from QueueStore in bulks. Bulk load is the size of these bulks. By default it is 250.
MultiMap
is a specialized map where you can store multiple values under a single key. Just like any other distributed data structure implementation in Hazelcast, MultiMap
is distributed and thread-safe.
MultiMap
is not an implementation of java.util.Map
due to the difference in method signatures. It supports most features of IMap
except for indexing, predicates and MapLoader/MapStore. Yet, like map, entries are almost evenly distributed onto all cluster members and when a new member joins to the cluster, the same ownership logic used in distributed map applies.
Let's write a code that puts data into a MultiMap.
public class PutMember {
public static void main( String[] args ) {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
MultiMap <String , String > map = hazelcastInstance.getMultiMap( "map" );
map.put( "a", "1" );
map.put( "a", "2" );
map.put( "b", "3" );
System.out.println( "PutMember:Done" );
}
}
And, now let's print the entries in this MultiMap.
public class PrintMember {
public static void main( String[] args ) {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
MultiMap <String, String > map = hazelcastInstance.getMultiMap( "map" );
for ( String key : map.keySet() ){
Collection <String > values = map.get( key );
System.out.println( "%s -> %s\n",key, values );
}
}
}
After you run the first code, run the PrintMember
. You will see the key a
has two values, as shown below.
b -> [3]
a -> [2, 1]
When using MultiMap, the collection type of values can be either a Set or a List. This is configured with valueCollectionType
parameter. If you choose Set
, duplicate and null values are not allowed in your collection and ordering is irrelevant. If you choose List
, ordering is relevant and your collection can include duplicate and null values.
You can also enable statistics for your MultiMap using the statisticsEnabled
parameter. If enabled, statistics can be retrieved with getLocalMultiMapStats()
method.
RELATED INFORMATION
Please refer to MultiMapConfig.java for more information on configuration options.
Hazelcast Set is distributed and concurrent implementation of java.util.Set
.
java.util.HashSet
.import com.hazelcast.core.Hazelcast;
import java.util.Set;
import java.util.Iterator;
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Set<Price> set = hazelcastInstance.getSet( "IBM-Quote-History" );
set.add( new Price( 10, time1 ) );
set.add( new Price( 11, time2 ) );
set.add( new Price( 12, time3 ) );
set.add( new Price( 11, time4 ) );
//....
Iterator<Price> iterator = set.iterator();
while ( iterator.hasNext() ) {
Price price = iterator.next();
//analyze
}
Hazelcast Set uses ItemListener to listen to events which occur when items are added and removed.
import java.util.Queue;
import java.util.Map;
import java.util.Set;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.ItemListener;
import com.hazelcast.core.EntryListener;
import com.hazelcast.core.EntryEvent;
public class Sample implements ItemListener {
public static void main( String[] args ) {
Sample sample = new Sample();
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ISet<Price> set = hazelcastInstance.getSet( "default" );
set.addItemListener( sample, true );
Price price = new Price( 10, time1 )
set.add( price );
set.remove( price );
}
public void itemAdded( Object item ) {
System.out.println( "Item added = " + item );
}
public void itemRemoved( Object item ) {
System.out.println( "Item removed = " + item );
}
}
NOTE: To learn more about the configuration of listeners please refer to Listener Configurations.
Hazelcast List is very similar to Hazelcast Set but it allows duplicate elements.
import com.hazelcast.core.Hazelcast;
import java.util.List;
import java.util.Iterator;
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
List<Price> list = hz.getList( "IBM-Quote-Frequency" );
list.add( new Price( 10 ) );
list.add( new Price( 11 ) );
list.add( new Price( 12 ) );
list.add( new Price( 11 ) );
list.add( new Price( 12 ) );
//....
Iterator<Price> iterator = list.iterator();
while ( iterator.hasNext() ) {
Price price = iterator.next();
//analyze
}
Hazelcast List uses ItemListener
to listen to events which occur when items are added and removed.
import java.util.Queue;
import java.util.Map;
import java.util.Set;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.ItemListener;
import com.hazelcast.core.EntryListener;
import com.hazelcast.core.EntryEvent;
public class Sample implements ItemListener{
public static void main( String[] args ) {
Sample sample = new Sample();
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IList<Price> list = hazelcastInstance.getList( "default" );
list.addItemListener( sample, true );
Price price = new Price( 10, time1 )
list.add( price );
list.remove( price );
}
public void itemAdded( Object item ) {
System.out.println( "Item added = " + item );
}
public void itemRemoved( Object item ) {
System.out.println( "Item removed = " + item );
}
}
NOTE: To learn more about the configuration of listeners please refer to Listener Configurations.
Hazelcast provides distribution mechanism for publishing messages that are delivered to multiple subscribers which is also known as publish/subscribe (pub/sub) messaging model. Publishing and subscribing operations are cluster wide. When a member subscribes for a topic, it is actually registering for messages published by any member in the cluster, including the new members joined after you add the listener.
NOTE: Publish operation is async. It does not wait for operations to run in remote nodes, it works as fire and forget.
import com.hazelcast.core.Topic;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.MessageListener;
public class Sample implements MessageListener<MyEvent> {
public static void main( String[] args ) {
Sample sample = new Sample();
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ITopic topic = hazelcastInstance.getTopic( "default" );
topic.addMessageListener( sample );
topic.publish( new MyEvent() );
}
public void onMessage( Message<MyEvent> message ) {
MyEvent myEvent = message.getMessageObject();
System.out.println( "Message received = " + myEvent.toString() );
if ( myEvent.isHeavyweight() ) {
messageExecutor.execute( new Runnable() {
public void run() {
doHeavyweightStuff( myEvent );
}
} );
}
}
// ...
private final Executor messageExecutor = Executors.newSingleThreadExecutor();
}
Topic has two statistic variables that can be queried. These values are incremental and local to the member.
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ITopic<Object> myTopic = hazelcastInstance.getTopic( "myTopicName" );
myTopic.getLocalTopicStats().getPublishOperationCount();
myTopic.getLocalTopicStats().getReceiveOperationCount();
getPublishOperationCount
and getReceiveOperationCount
returns total number of publishes and received messages since the start of this node, respectively. Please note that, these values are not backed up and if the node goes down, they will be lost.
This feature can be disabled with topic configuration. Please see Topic Configuration.
NOTE: These statistics values can be also viewed in Management Center. Please see Topics.
Each node has the list of all registrations in the cluster. When a new node is registered for a topic, it will send a registration message to all members in the cluster. Also, when a new node joins the cluster, it will receive all registrations made so far in the cluster.
The behavior of topic varies depending on the value of configuration parameter globalOrderEnabled
.
globalOrderEnabled
is disabled:Messages are ordered, i.e. listeners (subscribers) will process the messages in the order they are actually published. If cluster member M publishes messages m1, m2, m3,...,mn to a topic T, then Hazelcast makes sure that all of the subscribers of topic T will receive and process m1, m2, m3,...,mn in the given order.
Here is how it works. Let's say that we have three nodes (node1, node2 and node3) and that node1 and node2 are registered to a topic named news
. Notice that, all three nodes know that node1 and node2 registered to news
.
In this example, node1 publishes two messages: a1
and a2
. And, node3 publishes two messages: c1
and c2
. When node1 and node3 publishes a message, they will check their local list for registered nodes. They discover that node1 and node2 are in the list. Then, it fires messages to those nodes. One of the possible order of messages received can be following.
Node1 -> c1, b1, a2, c2
Node2 -> c1, c2, a1, a2
globalOrderEnabled
is enabled:When enabled, it guarantees that all nodes listening the same topic will get messages in the same order.
Here is how it works. Let's say that again we have three nodes (node1, node2 and node3) and that node1 and node2 are registered to a topic named news
. Notice that all three nodes know that node1 and node2 registered to news
.
In this example, node1 publishes two messages: a1
and a2
. And, node3 publishes two messages: c1
and c2
. When a node publishes messages over topic news
, it first calculates which partition news
ID corresponds to. Then, send an operation to owner of the partition for that node to publish messages. Let's assume that news
corresponds to a partition that node2 owns. Then, node1 and node3 first sends all messages to node2. Assume that the messages are published in the following order.
Node1 -> a1, c1, a2, c2
Then, node2 publishes these messages by looking at registrations in its local list. It sends these messages to node1 and node2 (it will make a local dispatch for itself).
Node1 -> a1, c1, a2, c2
Node2 -> a1, c1, a2, c2
This way we guarantee that all nodes will see the events in same order.
In both cases, there is a StripedExecutor
in EventService responsible for dispatching the received message. For all events in Hazelcast, the order that events are generated and the order they are published to the user are guaranteed to be the same via this StripedExecutor
.
In StripedExecutor
, there are as much threads specified in the property hazelcast.event.thread.count
(default is 5). For a specific event source (for topic, for a particular topic name), hash of that source's name % 5 gives the ID of responsible thread. Note that, there can be another event source (entry listener of a map, item listener of a collection, etc.) corresponding to the same thread. In order not to make other messages to block, heavy process should not be done in this thread. If there is a time consuming work needs to be done, the work should be handed over to another thread. Please see Sample Topic Code.
Declarative Configuration:
<hazelcast>
...
<topic name="yourTopicName">
<global-ordering-enabled>true</global-ordering-enabled>
<statistics-enabled>true</statistics-enabled>
<message-listeners>
<message-listener>MessageListenerImpl</message-listener>
</message-listeners>
</topic>
...
</hazelcast>
Programmatic Configuration:
TopicConfig topicConfig = new TopicConfig();
topicConfig.setGlobalOrderingEnabled( true );
topicConfig.setStatisticsEnabled( true );
topicConfig.setName( "yourTopicName" );
MessageListener<String> implementation = new MessageListener<String>() {
@Override
public void onMessage( Message<String> message ) {
// process the message
}
};
topicConfig.addMessageListenerConfig( new ListenerConfig( implementation ) );
HazelcastInstance instance = Hazelcast.newHazelcastInstance()
Default values are
Global ordering is false, meaning there is no global order guarantee by default.
Statistics are true, meaning statistics are calculated by default.
Topic related but not topic specific configuration parameters
hazelcast.event.queue.capacity
: default value is 1,000,000hazelcast.event.queue.timeout.millis
: default value is 250hazelcast.event.thread.count
: default value is 5
RELATED INFORMATION
For description of these parameters, please see Global Event Configuration
ILock is the distributed implementation of java.util.concurrent.locks.Lock
. Meaning if you lock on an ILock, the critical
section that it guards is guaranteed to be executed by only one thread in entire cluster. Even though locks are great for synchronization, they can lead to problems if not used properly. And also please note that Hazelcast Lock does not support fairness.
A few warnings when using locks:
import com.hazelcast.core.Hazelcast;
import java.util.concurrent.locks.Lock;
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Lock lock = hazelcastInstance.getLock( "myLock" );
lock.lock();
try {
// do something here
} finally {
lock.unlock();
}
tryLock
with a timeout value can be used. One can
set a high value (normally should not take that long) for tryLock
. Return value of tryLock
can be checked as follows:if ( lock.tryLock ( 10, TimeUnit.SECONDS ) ) {
try {
// do some stuff here..
} finally {
lock.unlock();
}
} else {
// warning
}
Another method to avoid ending up with indefinitely waiting threads is using lock with lease time. This will cause
lock to be released in the given time. Lock can be unlocked before time expires safely. Note that the unlock operation can
throw IllegalMonitorStateException
if lock is released because of lease time expiration. If it is the case, it means
that critical section guarantee is broken.
Please see the below example.
lock.lock( 5, TimeUnit.SECONDS )
try {
// do some stuff here..
} finally {
try {
lock.unlock();
} catch ( IllegalMonitorStateException ex ){
// WARNING Critical section guarantee can be broken
}
}
Locks are re-entrant, meaning same thread can lock multiple times on the same lock. Note that for other threads to be able to require this lock, owner of the lock should call unlock as many times as it called lock.
In split-brain scenario, cluster behaves as if there are two different clusters. Since two separate clusters are not aware of each other, two nodes from different clusters can acquire the same lock. For more information on places where split brain can be handled, please see Split Brain.
Locks are not automatically removed. If a lock is not used anymore, Hazelcast will not automatically garbage collect the lock and this can lead to an OutOfMemoryError. So if you create locks on the fly, make sure they are destroyed.
Hazelcast IMap also provides a locking support on the entry level using the method IMap.lock(key)
. Although the same infrastructure
is being used, IMap.lock(key)
is not an ILock and it is not possible to expose it directly.
ICondition is the distributed implementation of notify
, notifyAll
and wait
operations on Java object . It can be used to synchronize
threads across the cluster. More specifically, it is used when a thread's work depends on another thread's output. A good example
can be producer/consumer methodology.
Please see the below code snippets for a sample producer/consumer implementation.
Producer thread:
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Lock lock = hazelcastInstance.getLock( "myLockId" );
ICondition condition = lock.newCondition( "myConditionId" );
lock.lock();
try {
while ( !shouldProduce() ) {
condition.await(); // frees the lock and waits for signal
// when it wakes up it re-acquires the lock
// if available or waits for it to become
// available
}
produce();
condition.signalAll();
} finally {
lock.unlock();
}
Consumer thread:
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Lock lock = hazelcastInstance.getLock( "myLockId" );
ICondition condition = lock.newCondition( "myConditionId" );
lock.lock();
try {
while ( !canConsume() ) {
condition.await(); // frees the lock and waits for signal
// when it wakes up it re-acquires the lock if
// available or waits for it to become
// available
}
consume();
condition.signalAll();
} finally {
lock.unlock();
}
Hazelcast IAtomicLong is the distributed implementation of java.util.concurrent.atomic.AtomicLong
. It offers most of AtomicLong's operations such as get
, set
, getAndSet
, compareAndSet
and incrementAndGet
. Since IAtomicLong is a distributed implementation, these operations involve remote calls and hence their performances differ from AtomicLong.
Below sample code creates an instance, increments it by a million and prints the count.
public class Member {
public static void main( String[] args ) {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IAtomicLong counter = hazelcastInstance.getAtomicLong( "counter" );
for ( int k = 0; k < 1000 * 1000; k++ ) {
if ( k % 500000 == 0 ) {
System.out.println( "At: " + k );
}
counter.incrementAndGet();
}
System.out.printf( "Count is %s\n", counter.get() );
}
}
When you start other instances with the code above, you will see the count as member count times a million.
You can send functions to an IAtomicLong. Function
is a Hazelcast owned, single method interface. Below sample Function
implementation doubles the original value.
private static class Add2Function implements Function <Long, Long> {
@Override
public Long apply( Long input ) {
return input + 2;
}
}
Below methods can be used to execute functions on IAtomicLong.
apply
: It applies the function to the value in IAtomicLong without changing the actual value and returning the result.alter
: It alters the value stored in the IAtomicLong by applying the function. It will not send back a result.alterAndGet
: It alters the value stored in the IAtomicLong by applying the function, storing the result in the IAtomicLong and returning the result.getAndAlter
: It alters the value stored in the IAtomicLong by applying the function and returning the original value.Below sample code includes these methods.
public class Member {
public static void main( String[] args ) {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IAtomicLong atomicLong = hazelcastInstance.getAtomicLong( "counter" );
atomicLong.set( 1 );
long result = atomicLong.apply( new Add2Function() );
System.out.println( "apply.result: " + result);
System.out.println( "apply.value: " + atomicLong.get() );
atomicLong.set( 1 );
atomicLong.alter( new Add2Function() );
System.out.println( "alter.value: " + atomicLong.get() );
atomicLong.set( 1 );
result = atomicLong.alterAndGet( new Add2Function() );
System.out.println( "alterAndGet.result: " + result );
System.out.println( "alterAndGet.value: " + atomicLong.get() );
atomicLong.set( 1 );
result = atomicLong.getAndAlter( new Add2Function() );
System.out.println( "getAndAlter.result: " + result );
System.out.println( "getAndAlter.value: " + atomicLong.get() );
}
}
The reason for using a function instead of a simple code line like atomicLong.set(atomicLong.get() + 2));
is that, read and write operations of IAtomicLong are not atomic. Since it is a distributed implementation, those operations can be remote ones, which may lead to race problems. By using functions, the data is not pulled into the code, but the code is sent to the data. And this makes it more scalable.
NOTE: IAtomicLong has 1 synchronous backup and no asynchronous backups. Its backup count is not configurable.
Hazelcast ISemaphore is the distributed implementation of java.util.concurrent.Semaphore
. As you may know, semaphores offer permits to control the thread counts in the case of performing concurrent activities. To execute a concurrent activity, a thread grants a permit or waits until a permit becomes available. When the execution is completed, permit is released.
NOTE: Semaphore with a single permit may be considered as a lock. But, unlike the locks, when semaphores are used any thread can release the permit and also semaphores can have multiple permits.
NOTE: Hazelcast Semaphore does not support fairness.
When a permit is acquired on ISemaphore:
InstanceDestroyedException
is thrown.Below sample code uses an IAtomicLong resource for 1000 times, increments the resource when a thread starts to use it and decrements it when the thread completes.
public class SemaphoreMember {
public static void main( String[] args ) throws Exception{
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ISemaphore semaphore = hazelcastInstance.getSemaphore( "semaphore" );
IAtomicLong resource = hazelcastInstance.getAtomicLong( "resource" );
for ( int k = 0 ; k < 1000 ; k++ ) {
System.out.println( "At iteration: " + k + ", Active Threads: " + resource.get() );
semaphore.acquire();
try {
resource.incrementAndGet();
Thread.sleep( 1000 );
resource.decrementAndGet();
} finally {
semaphore.release();
}
}
System.out.println("Finished");
}
}
Let's limit the concurrent access to this resource by allowing at most 3 threads. This can be configured declaratively by setting the initial-permits
property, as shown below.
<semaphore name="semaphore">
<initial-permits>3</initial-permits>
</semaphore>
NOTE: If there is a shortage of permits while the semaphore is being created, value of this property can be set to a negative number.
If you execute the above SemaphoreMember
class 5 times, output will be similar to the following:
At iteration: 0, Active Threads: 1
At iteration: 1, Active Threads: 2
At iteration: 2, Active Threads: 3
At iteration: 3, Active Threads: 3
At iteration: 4, Active Threads: 3
As can be seen, maximum count of concurrent threads is equal or smaller than 3. If you remove the semaphore acquire/release statements in SemaphoreMember
, you will see that there is no limitation on the number of concurrent usages.
Hazelcast also provides backup support for ISemaphore. When a member goes down, another member can take over the semaphore with the permit information (permits are automatically released when a member goes down). To enable this, synchronous or asynchronous backup should be configured with the properties backup-count
and async-backup-count
(by default, synchronous backup is already enabled).
A sample configuration is shown below.
<semaphore name="semaphore">
<initial-permits>3</initial-permits>
<backup-count>1</backup-count>
</semaphore>
NOTE: If high performance is more important (than not losing the permit information), you can disable the backups by setting backup-count
to 0.
The IAtomicLong is very useful if you need to deal with a long, but in some cases you need to deal with a reference. That is why Hazelcast also supports the IAtomicReference which is the distributed version of the java.util.concurrent.atomic.AtomicReference
.
So lets see the IAtomicReference in action:
public class Member {
public static void main(String[] args) {
Config config = new Config();
HazelcastInstance hz = Hazelcast.newHazelcastInstance(config);
IAtomicReference<String> ref = hz.getAtomicReference("reference");
ref.set("foo");
System.out.println(ref.get());
System.exit(0);
}
}
When you execute the above sample, you will see the below output.
foo
Just like the IAtomicLong, the IAtomicReference has a bunch of methods that accept a 'function' as argument like alter
, alterAndGet
, getAndAlter
and apply
. There are two big advantages of using these methods:
Of course there are some issues you need to be aware of as described below.
compareAndSet
method, it is important not to change to original value because its serialized content will then be different.
Also important to know is that, if you rely on Java serialization, sometimes (especially with hashmaps) the same object can result in different binary content.binary
. So the receiving side does not need to have the class definition available, unless it needs to be deserialized on the other side (e.g. because a method like 'alter' is executed. This deserialization is done for every call that needs to have the object instead of the binary content, so be careful with expensive object graphs that need to be deserialized.apply
method for that. This way the whole object does not need to be sent over the line, only the information that is relevant.Hazelcast ICountDownLatch is the distributed implementation of java.util.concurrent.CountDownLatch
. As you may know, CountDownLatch is considered to be a gate keeper for concurrent activities. It enables the threads to wait for other threads to complete their operations.
Below sample codes describe the mechanism of ICountDownLatch. Assume that there is a leader process and there are follower ones that will wait until the leader completes. Here is the leader:
public class Leader {
public static void main( String[] args ) throws Exception {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ICountDownLatch latch = hazelcastInstance.getCountDownLatch( "countDownLatch" );
System.out.println( "Starting" );
latch.trySetCount( 1 );
Thread.sleep( 30000 );
latch.countDown();
System.out.println( "Leader finished" );
latch.destroy();
}
}
Since only a single step is needed to be completed as a sample, above code initializes the latch with 1. Then, sleeps for a while to simulate a process and starts the countdown. Finally, it clears up the latch. And now, let's write a follower:
public class Follower {
public static void main( String[] args ) throws Exception {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ICountDownLatch latch = hazelcastInstance.getCountDownLatch( "countDownLatch" );
System.out.println( "Waiting" );
boolean success = latch.await( 10, TimeUnit.SECONDS );
System.out.println( "Complete: " + success );
}
}
The follower class above first retrieves ICountDownLatch and then calls the await
method to enable the thread to listen for the latch. The method await
has a timeout value as a parameter. This is useful when countDown
method fails. To see ICountDownLatch on action, start the leader first and then start one or more followers. You will see that followers will wait until the leader completes.
In a distributed environment, it is possible that the counting down cluster member may go down. In this case, all listeners are notified immediately and automatically by Hazelcast. Of course, state of the current process just before the failure should be verified and 'how to continue now' should be decided (e.g. restart all process operations, continue with the first failed process operation, throw an exception, etc.).
Although the ICountDownLatch is a very useful synchronization aid, it probably is not one you will use on a daily basis. Unlike Java’s implementation, Hazelcast’s ICountDownLatch count can be re-set after a countdown has finished but not during an active count.
NOTE: ICountDownLatch has 1 synchronous backup and no asynchronous backups. Its backup count is not configurable. Also, the count cannot be re-set during an active count, it should be re-set after the countdown is finished.
Hazelcast IdGenerator is used to generate cluster-wide unique identifiers. Generated identifiers are long type primitive values between 0 and Long.MAX_VALUE
.
ID generation occurs almost at the speed of AtomicLong.incrementAndGet()
. A group of 1 million identifiers is allocated for each cluster member. In the background, this allocation takes place with an IAtomicLong incremented by 1 million. Once cluster member claims to generate IDs (allocation is done), IdGenerator is able to increment a local counter. If a cluster member uses all IDs in the group, it will have another 1 million IDs. By this way, only one time of network traffic is needed, meaning 999.999 identifiers are generated in memory. And this is fast.
Let's write a sample identifier generator.
public class IdGeneratorExample {
public static void main( String[] args ) throws Exception {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IdGenerator idGen = hazelcastInstance.getIdGenerator( "newId" );
while (true) {
Long id = idGen.newId();
System.err.println( "Id: " + id );
Thread.sleep( 1000 );
}
}
}
And let's run the above code two times. Output will be similar to the below.
Members [1] {
Member [<IP_ADDRESS>]:5701 this
}
Id: 1
Id: 2
Id: 3
Members [2] {
Member [<IP_ADDRESS>]:5701
Member [<IP_ADDRESS>]:5702 this
}
Id: 1000001
Id: 1000002
Id: 1000003
You can see that the generated IDs are unique and counting upwards. If you see duplicated identifiers, it means your instances could not form a cluster.
NOTE: Generated IDs are unique during the life cycle of the cluster. If the entire cluster is restarted, IDs start from 0 again or you can initialize to a value using the init()
method of IdGenerator.
NOTE: IdGenerator has 1 synchronous backup and no asynchronous backups. Its backup count is not configurable.
A replicated map is a weakly consistent, distributed key-value data structure provided by Hazelcast.
In difference to all other data structures which are partitioned in design, a replicated map does not partition data (it does not spread data to different cluster members) but replicates the data to all nodes.
This leads to higher memory consumption but faster read and write access since data are available on all nodes and writes take place on local nodes, eventually being replicated to all other nodes.
Weak consistency compared to eventually consistency means that replication is done on a best efforts basis. Lost or missing updates are neither tracked nor resent. This kind of data structures is suitable for immutable objects, catalogue data or idempotent calculable data (like HTML pages).
It nearly fully implements the java.util.Map
interface but lacks the methods from java.util.concurrent.ConcurrentMap
since
there are no atomic guarantees to writes or reads.
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import java.util.Collection;
import java.util.Map;
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
Map<String, Customer> customers = hazelcastInstance.getReplicatedMap("customers");
customers.put( "1", new Customer( "Joe", "Smith" ) );
customers.put( "2", new Customer( "Ali", "Selam" ) );
customers.put( "3", new Customer( "Avi", "Noyan" ) );
Collection<Customer> colCustomers = customers.values();
for ( Customer customer : colCustomers ) {
// process customer
}
HazelcastInstance::getReplicatedMap
actually returns com.hazelcast.core.ReplicatedMap
which, as stated above, extends
java.util.Map
interface.
The com.hazelcast.core.ReplicatedMap
interface has some additional methods for registering entry listeners or retrieving
values in an expected order.
A replicated map does not support ordered writes! In case of a conflict by two nodes simultaneously written to the same key, a vector clock algorithm is used to resolve and decide on one of the values.
Due to the weakly consistent nature and the previously mentioned behaviors of replicated map, there is a chance of reading staled data at any time. There is no read guarantee like repeatable reads.
ReplicatedMap does offer a distributed java.util.Map::clear
implementation, but due to the asynchronous nature and the
weakly consistency of it, there is no point in time where you can say the map is empty. Every node on itself
applies it to its local dataset in "a near point in time".
If you need a definite point in time to empty the map, you may want to consider using a lock around the clear
operation.
There are ways to simulate this method by locking your user codebase and executing a remote operation that will
utilize DistributedObject::destroy
to destroy the node's own proxy and storage of the ReplicatedMap. A new proxy instance
and storage will be created on next retrieval of the ReplicatedMap using HazelcastInstance::getReplicatedMap
.
That means, you have to reallocate the ReplicatedMap in your code. Afterwards, just release the lock when finished.
There are several technical design decisions that worth mentioning. We strongly suggest reading the following section to be aware of the configurable behavior.
Initial provisioning
If a new member joins, there are two ways of handling the initial provisioning that is executed to replicate all existing values to the new member.
First of all, you can have an async fill up which does not block reads while fill up operation is underway. That way, you have immediate access on the new member but it will take time until all values are eventually accessible. Not yet replicated values are returned as non existing (null). Write operations to already existing keys during this async phase can be lost since the vector clock for an entry might not be initialized by another member yet and might be seen as an old update by other members.
The second way is to have a synchronous initial fill up which blocks every read or write access to the map until the fill up operation is finished. This way should be used with caution since it might block your application from operation.
Replication delay
By default, the replication of values is delayed by 100 milliseconds when no current waiting replication is found. This is used
to collect multiple updates and to minimize the operations overhead on replication. A hard limit of 1000 replications
is built into the system to prevent OutOfMemory
situations where you put lots of data into the replicated map in a really
short time.
The delay is configurable and a value of "0" means immediate replication. That way, you can configure the trade off between
replication overhead and time for the value to be replicated.
Concurrency Level
The concurrency level configuration is used to define the number of mutexes and segments inside the replicated map storage.
A mutex/segment is chosen by calculating the hashCode
of the key and using module by the concurrency level. If multiple
keys fall into the same mutex ,they will have to wait for other mutex holders on the same mutex to finish their operation.
For high amount of values or high contention on the mutexes, this value might be interesting to be changed.
Currently two in-memory-format
values are usable with the ReplicatedMap.
OBJECT
(default): The data will be stored in deserialized form. This configuration is the default choice since
data replication is mostly used for high speed access. Please be aware that, changing values without a Map::put
is
not reflected on other nodes but is visible on the changing nodes for later value accesses.
BINARY
: The data will be stored in serialized binary format and have to be deserialized on every request. This
option offers higher encapsulation since changes to values are always discarded as long as the newly changed object is
not explicitly Map::put
into the map again.
In general a com.hazelcast.core.EntryListener
used on a ReplicatedMap serves the same purpose as it would on other
data structures in Hazelcast. You can use it to react on add, update, remove operations whereas eviction is not yet
supported by replicated maps.
The fundamental difference in behavior, compared to the other data structures, is that an EntryListener only reflects changes on local data. Since replication is asynchronous, all listener events are fired only when an operation is finished on a local node. With that in mind, events can fire at different times on different nodes.
import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.EntryListener;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ReplicatedMap;
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
ReplicatedMap<String, Customer> customers = hazelcastInstance.getReplicatedMap("customers");
customers.addEntryListener( new EntryListener<String, Customer>() {
@Override
public void entryAdded( EntryEvent<String, Customer> event ) {
log( "Entry added: " + event );
}
@Override
public void entryUpdated( EntryEvent<String, Customer> event ) {
log( "Entry updated: " + event );
}
@Override
public void entryRemoved( EntryEvent<String, Customer> event ) {
log( "Entry removed: " + event );
}
@Override
public void entryEvicted( EntryEvent<String, Customer> event ) {
// Currently not supported, will never fire
}
});
customers.put( "1", new Customer( "Joe", "Smith" ) ); // add event
customers.put( "1", new Customer( "Ali", "Selam" ) ); // update event
customers.remove( "1" ); // remove event
Hazelcast allows you to register for entry events to get notified when events occurred. Event Listeners are cluster-wide so when a listener is registered in one member of cluster, it is actually registering for events originated at any member in the cluster. When a new member joins, events originated at the new member will also be delivered.
An Event is created only if there is a listener registered. If there is no listener registered, then no event will be created. If a predicate provided while registering the listener, predicate should be passed before sending the event to the listener (node/client).
As a rule of thumb, event listener should not implement heavy processes in its event methods which block the thread for long time. If needed, ExecutorService
can be used to transfer long running processes to another thread and offload current listener thread.
hazelcast.event.queue.capacity
: default value is 1000000hazelcast.event.queue.timeout.millis
: default value is 250hazelcast.event.thread.count
: default value is 5There is a striped executor in each node to control and dispatch received events to user. This striped executor also guarantees the order. For all events in Hazelcast, the order that events are generated and the order they are published to the user are guaranteed for given keys. For map and multimap, order is preserved for the operations on same key of the entry. For list, set, topic and queue, order is preserved for events on that instance of the distributed data structure.
Order guarantee is achieved by making only one thread responsible for a particular set of events (entry events of a key in a map, item events of a collection, etc.) in StripedExecutor
.
If event queue reaches the capacity (hazelcast.event.queue.capacity
) and last item cannot be put to the event queue for the period specified in hazelcast.event.queue.timeout.millis
, these events will be dropped with a warning message like "EventQueue overloaded".
If listeners are doing a computation that requires a long time, this can cause event queue to reach its maximum capacity and lost of events. For map and multimap, hazelcast.event.thread.count
can be configured to a higher value so that less collision occurs for keys, therefore worker threads will not block each other in StripedExecutor
. For list, set, topic and queue, heavy work should be offloaded to another thread. Notice that, in order to preserve order guarantee, the user should implement similar logic with StripedExecutor
in offloaded thread pool.
RELATED INFORMATION
Please refer to Listener Configurations section on how to configure each listener.
One of the coolest features of Java 1.5 is the Executor framework, which allows you to asynchronously execute your tasks, logical units of works, such as database query, complex calculation, image rendering, etc.
The default implementation of this framework (ThreadPoolExecutor) is designed to run within a single JVM. In distributed systems, this implementation is not desired since you may want a task submitted in a JVM and processed in another one. Hazelcast offers IExecutorService to be used in distributed environments that implements java.util.concurrent.ExecutorService
, to serve the applications requiring computational and data processing power.
With IExecutorService, you can execute tasks asynchronously and perform other useful things meanwhile. When ready, get the result and move on. If execution of the task takes longer than expected, you may consider canceling the task execution. In Java Executor framework, tasks are implemented as java.util.concurrent.Callable
and java.util.Runnable
. If you need to return a value and submit to Executor, Callable is used. Otherwise, Runnable is used (if you do not need to return a value). Naturally, tasks should be Serializable
since they will be distributed.
Below is a sample Callable.
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.HazelcastInstanceAware;
import com.hazelcast.core.IMap;
import java.io.Serializable;
import java.util.concurrent.Callable;
public class SumTask
implements Callable<Integer>, Serializable, HazelcastInstanceAware {
private transient HazelcastInstance hazelcastInstance;
public void setHazelcastInstance( HazelcastInstance hazelcastInstance ) {
this.hazelcastInstance = hazelcastInstance;
}
public Integer call() throws Exception {
IMap<String, Integer> map = hazelcastInstance.getMap( "map" );
int result = 0;
for ( String key : map.localKeySet() ) {
System.out.println( "Calculating for key: " + key );
result += map.get( key );
}
System.out.println( "Local Result: " + result );
return result;
}
}
Executing a task by using executor framework is very straight forward. Simply obtain an ExecutorService
instance, generally via Executors
and submit the task which returns a Future
. After executing task, you do not have to wait for execution to complete, you can process other things and when ready use the future
object to retrieve the result as shown in code below.
ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<String> future = executorService.submit( new Echo( "myinput" ) );
//while it is executing, do some useful stuff
//when ready, get the result of your execution
String result = future.get();
Please note that Echo callable in the very above sample also implements Serializable interface, since it may be sent to another JVM to be processed.
NOTE: When a task is deserialized, HazelcastInstance needs to be accessed. To do this, the task should implement HazelcastInstanceAware
interface. Please see HazelcastInstanceAware Interface for more information.
Let's now go with a sample that is Runnable. Below is a task that waits for some time and echoes a message.
public class EchoTask implements Runnable, Serializable {
private final String msg;
public EchoTask( String msg ) {
this.msg = msg;
}
@Override
public void run() {
try {
Thread.sleep( 5000 );
} catch ( InterruptedException e ) {
}
System.out.println( "echo:" + msg );
}
}
Then let's write a class that submits and executes echo messages:
public class MasterMember {
public static void main( String[] args ) throws Exception {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IExecutorService executor = hazelcastInstance.getExecutorService( "exec" );
for ( int k = 1; k <= 1000; k++ ) {
Thread.sleep( 1000 );
System.out.println( "Producing echo task: " + k );
executor.execute( new EchoTask( String.valueOf( k ) ) );
}
System.out.println( "EchoTaskMain finished!" );
}
}
Above, Executor is retrieved from HazelcastInstance and 1000 echo tasks are submitted.
By default, Executor is configured to have 8 threads in the pool. It can be changed through the pool-size
property using declarative configuration (hazelcast.xml
). A sample is shown below (using above Executor).
<executor-service name="exec">
<pool-size>1</pool-size>
</executor-service>
Executor service can be scaled both vertically (scale up) and horizontally (scale out).
To scale up, processing capacity of the JVM should be improved. This can be done by increasing the pool-size
property mentioned in the previous subsection (i.e., increasing the thread count). However, please be aware of your JVM's capacity. If you think it cannot handle such additional load caused by increasing the thread count, you may want to consider improving the JVM's resources (CPU, memory, etc.). As an example, please set the pool-size
to 5 and run the above MasterMember
. You will see that EchoTask
is run as soon as it is produced.
To scale out, more JVMs should be added instead of increasing only one JVM's capacity. In reality, you may want to expand your cluster by adding more physical or virtual machines. For the EchoTask sample above, you can simply create another Hazelcast instance. It will automatically get involved in the executions started in MasterMember
and start processing.
Distributed executor service is a distributed implementation of java.util.concurrent.ExecutorService
. It allows you to execute your code in the cluster. In this section, all the code samples are based on the Echo class above. Please note that Echo class is Serializable
. You can ask Hazelcast to execute your code (Runnable, Callable
);
import com.hazelcast.core.Member;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.IExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.Set;
public void echoOnTheMember( String input, Member member ) throws Exception {
Callable<String> task = new Echo( input );
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IExecutorService executorService =
hazelcastInstance.getExecutorService( "default" );
Future<String> future = executorService.submitToMember( task, member );
String echoResult = future.get();
}
public void echoOnTheMemberOwningTheKey( String input, Object key ) throws Exception {
Callable<String> task = new Echo( input );
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IExecutorService executorService =
hazelcastInstance.getExecutorService( "default" );
Future<String> future = executorService.submitToKeyOwner( task, key );
String echoResult = future.get();
}
public void echoOnSomewhere( String input ) throws Exception {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IExecutorService executorService =
hazelcastInstance.getExecutorService( "default" );
Future<String> future = executorService.submit( new Echo( input ) );
String echoResult = future.get();
}
public void echoOnMembers( String input, Set<Member> members ) throws Exception {
HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();
IExecutorService executorService =
hazelcastInstance.getExecutorService( "default" );
Map<Member, Future<String>> futures = executorService
.submitToMembers( new Echo( input ), members );
for ( Future<String> future : futures.values() ) {
String echoResult = future.get();
// ...
}
}
NOTE: You can obtain the set of cluster members via HazelcastInstance#getCluster().getMembers()
call.