import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/**
 * Helper method to help assign ComputeJobs to respective ClusterNodes
 * 
 * @param clusterNode
 * @param keys
 * @param map
 * @return Map<ComputeJob, ClusterNode>
 */

private Map<ComputeJob, ClusterNode> setupCrossOver(ClusterNode clusterNode, List<Long> keys,
    Map<ComputeJob, ClusterNode> map) {
    // Calculate number of Jobs = keys / 2
    // as we desire pairs of Chromosomes to be swapped
    int numberOfJobs = keys.size() / 2;
    int k = 0;
    for (int i = 0; i < numberOfJobs; i++) {
        Long key1 = keys.get(k);
        Long key2 = keys.get(k + 1);

        CrossOverJob job = new CrossOverJob(key1, key2, this.config.getCrossOverRate());
        map.put(job, clusterNode);
        k = k + 2;
    }
    return map;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void invalidate(Iterable<Metadata> subPath, final IgniteBiInClosure<String, Set<Object>> action) {
    Map<String, List<Metadata>> names = getSnapshotsByCache(subPath);
    if (!names.isEmpty()) {
        ignite.compute().execute(new ComputeTaskSplitAdapter<Map<String, List<Metadata>>, Void>() {
            /** {@inheritDoc} */
            @Override protected Collection<? extends ComputeJob> split(int gridSize,
                Map<String, List<Metadata>> byCache) throws IgniteException {
                List<ComputeJob> result = new ArrayList<>();
                for (Map.Entry<String, List<Metadata>> entry : byCache.entrySet()) {
                    String cacheName = entry.getKey();
                    for (Metadata metadata : entry.getValue()) {
                        result.add(new ProcessAllKeysJob(cacheName, metadata, action));
                    }
                }
                return result;
            }

            /** {@inheritDoc} */
            @Override public Void reduce(List<ComputeJobResult> results) throws IgniteException {
                return null;
            }
        }, names);
    }
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/**
 * Performs restore operaton on given resource names.
 *
 * @param source base path to existing backup.
 * @param names of resources included in this backup.
 */
private void restoreData(final URI source, Iterable<String> names) {
    failOnExistingTransactions();
    ignite.compute().execute(new ComputeTaskSplitAdapter<Iterable<String>, Object>() {
        /** {@inheritDoc} */
        @Override protected Collection<? extends ComputeJob> split(int gridSize,
            Iterable<String> arg) throws IgniteException {
            List<ComputeJob> result = new ArrayList<>();
            for (String name : arg) {
                result.add(new RestoreJob(source, name));
            }
            return result;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Object reduce(List<ComputeJobResult> results) throws IgniteException {
            return null;
        }
    }, names);
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
@Override
protected Collection<? extends ComputeJob> split(int gridSize, final Integer arg) throws IgniteException {
    Set<ComputeJob> answer = new HashSet<>();
    for (int i = 0; i < arg; i++) {
        final int c = i;
        answer.add(new ComputeJob() {
            private static final long serialVersionUID = 3365213549618276779L;

            @Override
            public Object execute() throws IgniteException {
                return "a" + c;
            }

            @Override
            public void cancel() {
                // nothing
            }
        });
    }
    return answer;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) throws IgniteException {
    List<ComputeJob> jobs = new ArrayList<>(gridSize * 2);

    for (int i = 0; i < gridSize * 2; ++i) {
        jobs.add(new ComputeJobAdapter() {
            @Override public Object execute() throws IgniteException {
                assertTrue(Thread.currentThread().getName().contains(EXEC_NAME0));

                return null;
            }
        });
    }

    return jobs;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable ClusterNode arg) {
    for (ClusterNode node : subgrid) {
        if (node.id().equals(arg.id()))
            return Collections.singletonMap(new ComputeJobAdapter() {
                @TaskSessionResource
                private ComputeTaskSession ses;

                @Nullable @Override public Object execute() {
                    ses.saveCheckpoint("checkpoint-key", "checkpoint-value");

                    return null;
                }
            }, node);
    }

    assert false : "Expected node wasn't found in grid";

    // Never accessible.
    return null;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected Collection<? extends ComputeJob> split(int gridSize, List<Object> args) {
    Collection<ComputeJobAdapter> jobs = new ArrayList<>(args.size());

    for (final Object arg : args) {
        jobs.add(new ComputeJobAdapter() {
            @SuppressWarnings("OverlyStrongTypeCast")
            @Override public Object execute() {
                try {
                    return ((String)arg).length();
                }
                catch (ClassCastException ignored) {
                    assert arg instanceof Integer;

                    return arg;
                }
            }
        });
    }

    return jobs;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected Collection<? extends ComputeJob> split(int gridSize, Void arg) {
    return F.asSet(new ComputeJobAdapter() {
        @TaskSessionResource
        private ComputeTaskSession ses;

        @Override public Object execute() {
            CNT.incrementAndGet();

            if (fail)
                throw new ComputeExecutionRejectedException("Expected error.");

            return ses.getTaskName();
        }
    });
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected Collection<? extends ComputeJob> split(int gridSize, String arg) {
    Collection<ComputeJobAdapter> jobs = new ArrayList<>();

    if (arg != null)
        for (final Object val : arg.split(""))
            jobs.add(new ComputeJobAdapter() {
                @Override public Object execute() {
                    try {
                        Thread.sleep(5);
                    }
                    catch (InterruptedException ignored) {
                        Thread.currentThread().interrupt();
                    }

                    return val == null ? 0 : val.toString().length();
                }
            });

    return jobs;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Collection<? extends ComputeJob> split(int gridSize, String arg) {
    return Collections.singleton(new ComputeJobAdapter(arg) {
        @Override public Object execute() {
            try {
                Thread.sleep(10000);

                String val = argument(0);

                return val == null ? 0 : val.length();
            }
            catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    });
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/**
 * Splits the received string to words, creates a child job for each word, and sends
 * these jobs to other nodes for processing. Each such job simply prints out the received word.
 *
 * @param clusterSize Number of available cluster nodes. Note that returned number of
 *      jobs can be less, equal or greater than this cluster size.
 * @param arg Task execution argument. Can be {@code null}.
 * @return The list of child jobs.
 */
@Override protected Collection<? extends ComputeJob> split(int clusterSize, String arg) {
    Collection<ComputeJob> jobs = new LinkedList<>();

    for (final String word : arg.split(" ")) {
        jobs.add(new ComputeJobAdapter() {
            @Nullable @Override public Object execute() {
                System.out.println();
                System.out.println(">>> Printing '" + word + "' on this node from ignite job.");

                // Return number of letters in the word.
                return word.length();
            }
        });
    }

    return jobs;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, String arg) {
    assert ignite != null;

    UUID locNodeId = ignite.configuration().getNodeId();

    assert locNodeId != null;

    ClusterNode remoteNode = null;

    for (ClusterNode node : subgrid) {
        if (!node.id().equals(locNodeId))
            remoteNode = node;
    }

    return Collections.singletonMap(new ComputeJobAdapter(arg) {
        @Override public Serializable execute() {
            throw new IgniteException("Job exception.");
        }
    }, remoteNode);
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected Collection<? extends ComputeJob> split(int clusterSize, String arg) {
    Collection<ComputeJob> jobs = new ArrayList<>(clusterSize);

    for (int i = 0; i < clusterSize; i++) {
        jobs.add(new ComputeJobAdapter() {
            @Nullable @Override public Serializable execute() {
                System.out.println(">>> Executing deployment example job on this node.");

                // This job does not return any result.
                return null;
            }
        });
    }

    return jobs;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
    try {
        JobMapper mapper = new JobMapper(args.size());

        for (T jobArg : args) {
            ComputeJob job = job(this.job, jobArg);

            ClusterNode node = lb.getBalancedNode(job, null);

            mapper.map(job, node);
        }

        return mapper.map();
    }
    catch (IgniteCheckedException e) {
        throw new IgniteException(e);
    }
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) {
    return Collections.singletonList(new ComputeJobAdapter() {
        @IgniteInstanceResource
        private Ignite ignite;

        @Override public Object execute() throws IgniteException {
            MyService svc = ignite.services().service("my-service");

            while (!isCancelled()) {
                try {
                    Thread.sleep(1000);

                    svc.hello();
                }
                catch (InterruptedException ignored) {
                    // No-op.
                }
            }

            assertTrue(isCancelled());

            return svc.hello();
        }
    });
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, GridifyArgument arg) {
    assert !subgrid.isEmpty() : "Subgrid should not be empty: " + subgrid;

    assert ignite != null : "Grid instance could not be injected";
    assert balancer != null : "Load balancer could not be injected";

    ComputeJob job = new GridifyJobAdapter(arg);

    ClusterNode node = balancer.getBalancedNode(job, Collections.<ClusterNode>singletonList(ignite.cluster().localNode()));

    if (node != null) {
        // Give preference to remote nodes.
        return Collections.singletonMap(job, node);
    }

    return Collections.singletonMap(job, balancer.getBalancedNode(job, null));
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public ClusterNode getBalancedNode(ComputeTaskSession ses, List<ClusterNode> top, ComputeJob job) {
    A.notNull(ses, "ses");
    A.notNull(top, "top");
    A.notNull(job, "job");

    IgniteBiTuple<Boolean, WeightedTopology> weightedTop = taskTops.get(ses.getId());

    // Create new cached topology if there is no one. Do not
    // use cached topology after task has been mapped.
    if (weightedTop == null)
        // Called from ComputeTask#map(). Put new topology and false as not mapped yet.
        taskTops.put(ses.getId(), weightedTop = F.t(false, new WeightedTopology(top)));
    // We have topology - check if task has been mapped.
    else if (weightedTop.get1())
        // Do not use cache after ComputeTask#map().
        return new WeightedTopology(top).pickWeightedNode();

    return weightedTop.get2().pickWeightedNode();
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, UUID arg) {
    assert subgrid.size() == 2;
    assert arg != null;

    attrVal = UUID.randomUUID();

    for (ClusterNode node : subgrid) {
        if (node.id().equals(arg))
            return Collections.singletonMap(new SessionTestJob(attrVal), node);
    }

    assert false;

    return null;
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/** {@inheritDoc} */
@Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
    @Nullable Object arg) {
    taskSubjId = ((GridTaskSessionInternal)ses).subjectId();

    ClusterNode node = null;

    for (ClusterNode subgridNode : subgrid) {
        if (F.eq(targetNodeId, subgridNode.id())) {
            node = subgridNode;

            break;
        }
    }

    assert node != null;

    return Collections.singletonMap(new Job(), node);
}
 

import org.apache.ignite.compute.ComputeJob; //导入依赖的package包/类
/**
 * @param subgrid Subgrid.
 * @param args Args.
 * @param first First.
 * @param cache Cache.
 */
@Nullable private Map<? extends ComputeJob, ClusterNode> mapToNode(
    List<ClusterNode> subgrid,
    Map<String, Object> args,
    boolean first,
    String cache
) {
    GridDiscoveryManager discoMgr = ((IgniteKernal)ignite).context().discovery();

    for (ClusterNode n : subgrid) {
        if (discoMgr.cacheAffinityNode(n, cache)) {
            args.put("cache", cache);

            return F.asMap(new JdbcDriverJob(args, first), n);
        }
    }

    return null;
}