How do I make sure my apache Ignite 2.x distributed cache puts are asynchornous

Below I have a distributed cache example using apache ignite.

I want to make it so when I do a cache put operation: cache.put(i, new X12File("x12file" + i, LocalDateTime.now().toString())); that it is completely asynchronous. Meaning my put operation should be super fast, and the pushing to the rest of the cluster should happen in the background not inconveniencing the user. I need “fire and forget” functionality.

I am struggling to find an async option in the documentation. Can someone help me understand how to change this code to make my puts as “fire and forget” as possible? Thanks!

private static final String ORG_CACHE = IgniteCache.class.getSimpleName() + "Organizations";

public static void main(String[] args) throws Exception {
    DataRegionConfiguration dfltDataRegConf = new DataRegionConfiguration();
    dfltDataRegConf.setPersistenceEnabled(true);

    DataStorageConfiguration dsCfg = new DataStorageConfiguration();
    dsCfg.setDefaultDataRegionConfiguration(dfltDataRegConf);
    dsCfg.setStoragePath("/home/kazakov/tmp");

    IgniteConfiguration igniteConfiguration = new IgniteConfiguration();
    igniteConfiguration.setDataStorageConfiguration(dsCfg);

    TcpDiscoverySpi tcpDiscoverySpi = new TcpDiscoverySpi();
    TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder();
    ipFinder.setAddresses(Arrays.asList("127.0.0.1:47500..47509"));
    tcpDiscoverySpi.setIpFinder(ipFinder);

    igniteConfiguration.setDiscoverySpi(tcpDiscoverySpi);

    try(Ignite ignite = Ignition.start(igniteConfiguration)) {
        ignite.active(true);

        CacheConfiguration<Long, X12File> cacheCfg = new CacheConfiguration<>(ORG_CACHE);

        cacheCfg.setCacheMode(CacheMode.REPLICATED);
        cacheCfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
        cacheCfg.setBackups(1);
        cacheCfg.setOnheapCacheEnabled(true);

        IgniteCache<Long, X12File> cache = ignite.getOrCreateCache(cacheCfg).withExpiryPolicy(new CreatedExpiryPolicy(new Duration(TimeUnit.SECONDS, 1)));

        for (long i = 0; i < 4_000_000; i++) {
            if (i > 0 && i % 10_000 == 0)
                System.out.println("Done: " + i);

            cache.put(i, new X12File("x12file" + i, LocalDateTime.now().toString()));
        }

        Thread.sleep(5000);

        int matches = 0;
        for (long i = 0; i < 4_000_000; i++) {
            if (cache.get(i) != null)
                ++matches;
        }
        System.out.println("Matches: " + matches);
    }
}

JavaScript
​x
 
private static final String ORG_CACHE = IgniteCache.class.getSimpleName() + "Organizations";​public static void main(String[] args) throws Exception {    DataRegionConfiguration dfltDataRegConf = new DataRegionConfiguration();    dfltDataRegConf.setPersistenceEnabled(true);​    DataStorageConfiguration dsCfg = new DataStorageConfiguration();    dsCfg.setDefaultDataRegionConfiguration(dfltDataRegConf);    dsCfg.setStoragePath("/home/kazakov/tmp");​    IgniteConfiguration igniteConfiguration = new IgniteConfiguration();    igniteConfiguration.setDataStorageConfiguration(dsCfg);​    TcpDiscoverySpi tcpDiscoverySpi = new TcpDiscoverySpi();    TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder();    ipFinder.setAddresses(Arrays.asList("127.0.0.1:47500..47509"));    tcpDiscoverySpi.setIpFinder(ipFinder);​    igniteConfiguration.setDiscoverySpi(tcpDiscoverySpi);​    try(Ignite ignite = Ignition.start(igniteConfiguration)) {        ignite.active(true);​        CacheConfiguration<Long, X12File> cacheCfg = new CacheConfiguration<>(ORG_CACHE);​        cacheCfg.setCacheMode(CacheMode.REPLICATED);        cacheCfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);        cacheCfg.setBackups(1);        cacheCfg.setOnheapCacheEnabled(true);​        IgniteCache<Long, X12File> cache = ignite.getOrCreateCache(cacheCfg).withExpiryPolicy(new CreatedExpiryPolicy(new Duration(TimeUnit.SECONDS, 1)));​        for (long i = 0; i < 4_000_000; i++) {            if (i > 0 && i % 10_000 == 0)                System.out.println("Done: " + i);​            cache.put(i, new X12File("x12file" + i, LocalDateTime.now().toString()));        }​        Thread.sleep(5000);​        int matches = 0;        for (long i = 0; i < 4_000_000; i++) {            if (cache.get(i) != null)                ++matches;        }        System.out.println("Matches: " + matches);    }}​

Answer

Check out IgniteCache#putAsync(..)

see: https://ignite.apache.org/releases/latest/javadoc/org/apache/ignite/IgniteCache.html#putAsync-K-V-

Advertisement

Answer