参考

https://docs.jboss.org/author/display/ISPN/Infinispan+Distributed+Execution+Framework#InfinispanDistributedExecutionFramework-MapReducemodel

首先下载我们的infinispan5的运行环境

https://github.com/infinispan/infinispan-quickstart.git

修改clustered-cache-quickstart项目

中org.infinispan.quickstart.clusteredcache.distribution.Node2类为

/* * JBoss, Home of Professional Open Source * Copyright 2011 Red Hat Inc. and/or its affiliates and other * contributors as indicated by the @author tags. All rights reserved. * See the copyright.txt in the distribution for a full listing of * individual contributors. * * This is free software; you can redistribute it and/or modify it * under the terms of the GNU Lesser General Public License as * published by the Free Software Foundation; either version 2.1 of * the License, or (at your option) any later version. * * This software is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General Public * License along with this software; if not, write to the Free * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA * 02110-1301 USA, or see the FSF site: http://www.fsf.org. */ package org.infinispan.quickstart.clusteredcache.distribution; import java.util.Iterator; import java.util.Map; import java.util.StringTokenizer; import java.util.Map.Entry; import org.infinispan.Cache; import org.infinispan.distexec.mapreduce.Collator; import org.infinispan.distexec.mapreduce.Collector; import org.infinispan.distexec.mapreduce.MapReduceTask; import org.infinispan.distexec.mapreduce.Mapper; import org.infinispan.distexec.mapreduce.Reducer; public class Node2 extends AbstractNode { /** * * @author fansxnet * */ static class WordCountMapper implements Mapper<String, String, String, Integer> { private static final long serialVersionUID = 1L; @Override public void map(String key, String value, Collector<String, Integer> c) { StringTokenizer tokens = new StringTokenizer(value); while (tokens.hasMoreElements()) { String s = (String) tokens.nextElement(); System.out.println("入口"+s); c.emit(s, 1); } } } static class WordCountReducer implements Reducer<String, Integer> { /** The serialVersionUID */ private static final long serialVersionUID = 1901016598354633256L; @Override public Integer reduce(String key, Iterator<Integer> iter) { int sum = 0; while (iter.hasNext()) { Integer i = (Integer) iter.next(); sum += i; } System.out.print("统计"+sum); return sum; } } public static void main(String[] args) throws Exception { new Node2().run(); } public void run() { Cache<String, String> cache = getCacheManager().getCache("Demo"); for (int i = 0; i < 1000; i++) { cache.put("key" + i, "" + i); } cache.put("key" + 10, "" + 4); MapReduceTask<String, String, String, Integer> task = new MapReduceTask<String, String, String, Integer>( cache); task.mappedWith(new WordCountMapper()).reducedWith( new WordCountReducer()); String mostFrequentWord = task .execute(new Collator<String, Integer, String>() { @Override public String collate(Map<String, Integer> reducedResults) { String mostFrequent = ""; int maxCount = 0; for (Entry<String, Integer> e : reducedResults .entrySet()) { Integer count = e.getValue(); if (count > maxCount) { maxCount = count; mostFrequent = e.getKey(); } System.out.println(e.getKey()+"======"+maxCount); } return mostFrequent; } }); System.out.println("The most frequent word is " + mostFrequentWord); } @Override protected int getNodeId() { return 2; } }

分别运行

org.infinispan.quickstart.clusteredcache.distribution.Node0

org.infinispan.quickstart.clusteredcache.distribution.Node1

org.infinispan.quickstart.clusteredcache.distribution.Node2