import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
public Map<K, List<M>> putAll(@Nonnull Map<K, List<M>> values) {
    Map<K, ICompletableFuture<List<M>>> futureMap = new HashMap<>();
    values.forEach((key, message) -> futureMap.put(key, hazelcastMap.putAsync(key, message)));
    Map<K, List<M>> ret = new HashMap<>();
    futureMap.forEach((key, future) -> {
        try {
            List<M> value = future.get();
            if (value != null) {
                ret.put(key, value);
            }
        } catch (ExecutionException | InterruptedException e) {
            // TODO: Figure out if we timed out or were interrupted...
            throw new RuntimeException(e.getMessage(), e);
        }
    });
    return ret;
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
public Map<K, List<M>> removeAll(Collection<K> keys) {
    Map<K, ICompletableFuture<List<M>>> futureMap = new HashMap<>();
    keys.forEach(key -> futureMap.put(key, hazelcastMap.removeAsync(key)));
    Map<K, List<M>> ret = new HashMap<>();
    futureMap.forEach((key, future) -> {
        try {
            List<M> value = future.get();
            if (value != null) {
                ret.put(key, value);
            }
        } catch (ExecutionException | InterruptedException e) {
            // TODO: Figure out if we timed out or were interrupted...
            throw new RuntimeException(e.getMessage(), e);
        }

    });
    return ret;
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
@SuppressWarnings("unchecked")
public <B extends PMessageBuilder<Message, Field>> Map<Key, B> putAllBuilders(@Nonnull Map<Key, B> builders) {
    Map<Key, ICompletableFuture<Builder>> futureMap = new HashMap<>();
    builders.forEach((key, builder) -> futureMap.put(key, hazelcastMap.putAsync(key, (Builder) builder)));
    Map<Key, B> ret = new HashMap<>();
    futureMap.forEach((key, future) -> {
        try {
            Builder value = future.get();
            if (value != null) {
                ret.put(key, (B) value);
            }
        } catch (ExecutionException | InterruptedException e) {
            // TODO: Figure out if we timed out or were interrupted...
            throw new RuntimeException(e.getMessage(), e);
        }

    });
    return ret;
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
public Map<Key, Message> removeAll(Collection<Key> keys) {
    Map<Key, ICompletableFuture<Builder>> futureMap = new HashMap<>();
    keys.forEach(key -> futureMap.put(key, hazelcastMap.removeAsync(key)));
    Map<Key, Message> ret = new HashMap<>();
    futureMap.forEach((key, builder) -> {
        try {
            Builder value = builder.get();
            if (value != null) {
                ret.put(key, value.build());
            }
        } catch (ExecutionException | InterruptedException e) {
            // TODO: Figure out if we timed out or were interrupted...
            throw new RuntimeException(e.getMessage(), e);
        }
    });
    return ret;
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
public Map<Key, Message> putAll(@Nonnull Map<Key, Message> values) {
    Map<Key, ICompletableFuture<Message>> futureMap = new HashMap<>();
    values.forEach((key, message) -> futureMap.put(key, hazelcastMap.putAsync(key, message)));
    Map<Key, Message> ret = new HashMap<>();
    futureMap.forEach((key, future) -> {
        try {
            Message value = future.get();
            if (value != null) {
                ret.put(key, value);
            }
        } catch (ExecutionException | InterruptedException e) {
            // TODO: Figure out if we timed out or were interrupted...
            throw new RuntimeException(e.getMessage(), e);
        }
    });
    return ret;
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
public Map<Key, Message> removeAll(Collection<Key> keys) {
    Map<Key, ICompletableFuture<Message>> futureMap = new HashMap<>();
    keys.forEach(key -> futureMap.put(key, hazelcastMap.removeAsync(key)));
    Map<Key, Message> ret = new HashMap<>();
    futureMap.forEach((key, future) -> {
        try {
            Message value = future.get();
            if (value != null) {
                ret.put(key, value);
            }
        } catch (ExecutionException | InterruptedException e) {
            // TODO: Figure out if we timed out or were interrupted...
            throw new RuntimeException(e.getMessage(), e);
        }
    });
    return ret;
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    ICompletableFuture future = job.mapper(new SalaryMapper()) //
            .combiner(new SalaryCombinerFactory()) //
            .reducer(new SalaryReducerFactory()) //
            .submit();

    System.out.println(ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    // Collect all people by state
    ICompletableFuture future = job.mapper(new StateBasedCountMapper()).submit();

    // Count people by state
    // ICompletableFuture future = job.mapper(new StateBasedCountMapper()).reducer(new CountReducerFactory()).submit();

    // Same as above but with precalculation per node
    // ICompletableFuture future = job.mapper(new StateBasedCountMapper()).combiner(new CountCombinerFactory())
    //                                .reducer(new CountReducerFactory()).submit();

    System.out.println(ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
private <T> ScheduledFuture<T> submitToPartitionOwner(Callable<T> task, int partitionId, long delay, long period, boolean fixedRate) {
        if (task == null) {
            throw new NullPointerException("task can't be null");
        }
        if (isShutdown()) {
            throw new RejectedExecutionException(getRejectionMessage());
        }
        NodeEngine nodeEngine = getNodeEngine();
        Data taskData = nodeEngine.toData(task);
        String uuid = buildRandomUuidString();
        String name = getName();
        ScheduledCallableTaskOperation op = new ScheduledCallableTaskOperation(name, uuid, taskData, delay, period, fixedRate);
        ICompletableFuture future = invoke(partitionId, op);
        return new ScheduledDelegatingFuture<T>(future, nodeEngine.getSerializationService(), delay);
//        return new CancellableDelegatingFuture<T>(future, nodeEngine, uuid, partitionId);
    }
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
private static Map<String, Long> mapReduce(HazelcastInstance hazelcastInstance)
        throws Exception {

    // Retrieving the JobTracker by name
    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    // Creating the KeyValueSource for a Hazelcast IMap
    IMap<String, String> map = hazelcastInstance.getMap("articles");
    KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

    Job<String, String> job = jobTracker.newJob(source);

    // Creating a new Job
    ICompletableFuture<Map<String, Long>> future = job // returned future
            .mapper(new TokenizerMapper())             // adding a mapper
            .combiner(new WordCountCombinerFactory())  // adding a combiner through the factory
            .reducer(new WordCountReducerFactory())    // adding a reducer through the factory
            .submit();                                 // submit the task

    // Attach a callback listener
    future.andThen(buildCallback());

    // Wait and retrieve the result
    return future.get();
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
private static long mapReduceCollate(HazelcastInstance hazelcastInstance)
        throws Exception {

    // Retrieving the JobTracker by name
    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    // Creating the KeyValueSource for a Hazelcast IMap
    IMap<String, String> map = hazelcastInstance.getMap("articles");
    KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

    // Creating a new Job
    Job<String, String> job = jobTracker.newJob(source);

    ICompletableFuture<Long> future = job // returned future
            .mapper(new TokenizerMapper())             // adding a mapper
            .combiner(new WordCountCombinerFactory())  // adding a combiner through the factory
            .reducer(new WordCountReducerFactory())    // adding a reducer through the factory
            .submit(new WordCountCollator());          // submit the task and supply a collator

    // Wait and retrieve the result
    return future.get();
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
void add(ICompletableFuture<Long> future) {
    if (batchSize <= 0) {
        return;
    }

    batch.add(future);
    if (batch.size() == batchSize) {
        for (ICompletableFuture batchFuture : batch) {
            try {
                batchFuture.get();
            } catch (Exception e) {
                throw rethrow(e);
            }
        }
        batch.clear();
    }
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void pushEntry(K key, V value) {
    if (storedException.get() != null) {
        throw new RuntimeException("Aborting pushEntry; problems are detected. Please check the cause",
                storedException.get());
    }

    acquirePermit(1);
    try {
        ICompletableFuture<V> future = storeAsync(key, value);
        future.andThen(callback);
    } catch (Exception e) {
        releasePermit(1);

        throw rethrow(e);
    }
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Override
protected void init(@Nonnull Context context) throws Exception {
    ICompletableFuture<EventJournalInitialSubscriberState>[] futures = new ICompletableFuture[partitionIds.length];
    Arrays.setAll(futures, i -> eventJournalReader.subscribeToEventJournal(partitionIds[i]));
    for (int i = 0; i < futures.length; i++) {
        emitOffsets[i] = readOffsets[i] = getSequence(futures[i].get());
    }
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Nonnull
@Override
public ICompletableFuture<ClientMessage> invoke(@Nonnegative int partitionId, @Nonnull ClientMessage request) {
    try {
        ClientInvocation clientInvocation = new ClientInvocation(client, request, partitionId);
        return clientInvocation.invoke();

    } catch (Exception e) {
        throw ExceptionUtil.rethrow(e);
    }
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    ICompletableFuture<List<Map.Entry<String, Integer>>> future = //
            job.mapper(new SalaryMapper()) //
                    .combiner(new SalaryCombinerFactory()) //
                    .reducer(new SalaryReducerFactory()) //
                    .submit(new SalaryCollator());

    // Intermediate result
    List<Map.Entry<String, Integer>> orderedSalariesByState = future.get();
    Map.Entry<String, Integer> topSalary = orderedSalariesByState.get(0);

    IList<Crime> crimesList = hazelcastInstance.getList("crimes");
    KeyValueSource<String, Crime> crimeSource = KeyValueSource.fromList(crimesList);

    Job<String, Crime> crimeJob = jobTracker.newJob(crimeSource);

    ICompletableFuture<Map<CrimeCategory, Integer>> crimeFuture = //
            crimeJob.mapper(new CrimeMapper(topSalary.getKey())) //
                    .reducer(new CrimeReducerFactory()) //
                    .submit();

    System.out.println(ToStringPrettyfier.toString(crimeFuture.get()));
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
public static void main(String[] args)
        throws Exception {

    // Prepare Hazelcast cluster
    HazelcastInstance hazelcastInstance = buildCluster(3);

    try {

        // Read data
        fillMapWithData(hazelcastInstance);

        JobTracker tracker = hazelcastInstance.getJobTracker("default");

        IMap<String, String> map = hazelcastInstance.getMap(MAP_NAME);
        KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

        Job<String, String> job = tracker.newJob(source);
        ICompletableFuture<Map<String, Integer>> future = job
                .mapper(new TokenizerMapper())
                // Activate Combiner to add combining phase!
                // .combiner(new WordcountCombinerFactory())
                .reducer(new WordcountReducerFactory())
                .submit();



        System.out.println(ToStringPrettyfier.toString(future.get()));

    } finally {
        // Shutdown cluster
        Hazelcast.shutdownAll();
    }
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
private static void doMapReduce(
        final HazelcastInstance instance,
        final String filename) throws ExecutionException, InterruptedException, IOException
{
    final IMap<Integer, SpatialPoint> geomap = instance.getMap(GEOMAP);
    final JobTracker tracker = instance.getJobTracker("default");
    final Job<Integer, SpatialPoint> job = tracker.newJob(KeyValueSource.fromMap(geomap));
    final ICompletableFuture<Map<Integer, Integer>> future = job.
            mapper(new FeatureMapper(HEXMAP)).
            reducer(new FeatureReducerFactory()).
            submit();

    final FileOutputStream fileOutputStream = new FileOutputStream(filename);
    try
    {
        final PrintStream printStream = new PrintStream(fileOutputStream);
        printStream.format("ID,POPULATION%n");
        final Map<Integer, Integer> result = future.get();
        for (final Map.Entry<Integer, Integer> entry : result.entrySet())
        {
            printStream.format("%d,%d%n", entry.getKey(), entry.getValue());
        }
        printStream.flush();
    }
    finally
    {
        fileOutputStream.close();
    }
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
private <T> void executeAsync(ICompletableFuture<T> future,
                              Handler<AsyncResult<T>> resultHandler) {
    future.andThen(
            new HandlerCallBackAdapter(resultHandler),
            VertxExecutorAdapter.getOrCreate(vertx.getOrCreateContext())
    );
}
 

import com.hazelcast.core.ICompletableFuture; //导入依赖的package包/类
@Override
public void put(K k, V v, long ttl, Handler<AsyncResult<Void>> completionHandler) {
  K kk = convertParam(k);
  V vv = convertParam(v);
  executeAsyncVoid(
          (ICompletableFuture<Void>) map.putAsync(kk, vv, ttl, TimeUnit.MILLISECONDS),
          completionHandler
  );
}