infinispan5的MapReduce使用

参考

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

 

 

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

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

 

 

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

 

 

package org.infinispan.quickstart.clusteredcache.distribution;

import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.Map.Entry;

import org.infinispan.Cache;
import org.infinispan.config.Configuration;
import org.infinispan.config.GlobalConfiguration;
import org.infinispan.config.Configuration.CacheMode;
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;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;

/**
 * @author :xiaofancn
 * @version ：2011-11-30 下午10:15:44
 * 
 */

public class NodeClient {

	public static EmbeddedCacheManager createCacheManagerProgramatically() {

		return new DefaultCacheManager(GlobalConfiguration
				.getClusteredDefault().fluent().transport()
				.addProperty("configurationFile", "jgroups.xml").build(),
				new Configuration().fluent().mode(CacheMode.DIST_ASYNC).hash()
						.numOwners(2).build());
	}

	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()) {
				System.out.println(key + "=================" + value);
				String s = (String) tokens.nextElement();
				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.println(key+"："+sum);
			return sum;

		}
	}

	public static void main(String[] args) {

		Cache<String, String> cache = createCacheManagerProgramatically()
				.getCache("Demo");
		
		
		int count = 1000000000;
		for (int i = 0; i < count; i++) {
			cache.put("" + i, "" + i);
		}
		cache.put("" + count, "" + 4);

		System.out.println(new Date());
		Long start = System.currentTimeMillis();

		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();
							}
						}
						return mostFrequent;
					}

				});
		System.out.println(new Date());
		System.out.println(System.currentTimeMillis() - start);
		System.out.println("The most frequent word is " + mostFrequentWord);

	}

}

 

分别运行

org.infinispan.quickstart.clusteredcache.distribution.Node0

org.infinispan.quickstart.clusteredcache.distribution.Node1

org.infinispan.quickstart.clusteredcache.distribution.Node2

  org.infinispan.quickstart.clusteredcache.distribution.NodeClient