注意：在实际处理过程中，MapActor的结果直接发送给了ReduceActor，而ReduceActor处理的结果直接发送给了AggregateActor，没有经过MasterActor

MasterActor：

点击(此处)折叠或打开

/**
* MasterActor管理各个Actor，以及作为消息的入口、进行消息的转发。
*
*/
public class MasterActor extends UntypedActor {
//通过ActorContext创建Actor
private ActorRef aggregateActor = getContext().actorOf(new Props(AggregateActor.class),"aggregate");
private ActorRef reduceActor = getContext().actorOf(new Props(new UntypedActorFactory(){public Actor create() {return new ReduceActor(aggregateActor);}}),"reduce");
private ActorRef mapActor = getContext().actorOf(new Props(new UntypedActorFactory(){public Actor create() {return new MapActor(reduceActor);}}),"map");
@Override
public void onReceive(Object message) throws Exception {
if(message instanceof String){
mapActor.tell(message);
}else if(message instanceof Result){
aggregateActor.tell(message); //返回最终结果
}else{
unhandled(message);
}
}
}

MapActor：

点击(此处)折叠或打开

/**接收MasterActor消息，以及将计算结果发送给ReduceActor
*
*/
public class MapActor extends UntypedActor {
private ActorRef reduceActor = null;
String[] STOP_WORDS = {"a","is"};
private List<String> STOP_WORD_LIST = Arrays.asList(STOP_WORDS);
public MapActor(ActorRef reduceActor) {
this.reduceActor = reduceActor;
}
@Override
public void onReceive(Object message) throws Exception {
if(message instanceof String){
String work = (String)message;
//map the words in the sentence
MapData data = evaluateExpression(work);
//send the result to ReduceActor
reduceActor.tell(data);
}else{
unhandled(message);
}
}
private MapData evaluateExpression(String line){
List<WordCount> dataList = new ArrayList<WordCount>();
StringTokenizer parser = new StringTokenizer(line);
while(parser.hasMoreTokens()){
String word = parser.nextToken().toLowerCase();
if(!STOP_WORD_LIST.contains(word)){
dataList.add(new WordCount(word, Integer.valueOf(1)));
}
}
return new MapData(dataList);
}
}

ReduceActor：

点击(此处)折叠或打开

/**接收MapActor消息，以及将计算结果发送给AggregateActor
*
*/
public class ReduceActor extends UntypedActor {
private ActorRef aggregateActor = null;
public ReduceActor(ActorRef aggregateActor) {
this.aggregateActor = aggregateActor;
}
@Override
public void onReceive(Object message) throws Exception {
if(message instanceof MapData){
MapData mapData = (MapData)message;
//reduce the incoming data
ReduceData reduceData = reduce(mapData.getDataList());
//forward the result to aggregate actor
aggregateActor.tell(reduceData);
}else{
unhandled(message);
}
}
private ReduceData reduce(List<WordCount> dataList){
HashMap<String,Integer> reduceMap = new HashMap<String, Integer>();
for(WordCount wordCount : dataList){
if(reduceMap.containsKey(wordCount.getWord())){
Integer value = (Integer) reduceMap.get(wordCount.getWord());
value++;
reduceMap.put(wordCount.getWord(), value);
}else{
reduceMap.put(wordCount.getWord(),Integer.valueOf(1));
}
}
return new ReduceData(reduceMap);
}
}

AggregateActor：

点击(此处)折叠或打开

/**接收ReduceActor消息，进行数据汇总；接收MasterActor消息，返回最终结果
*
*/
public class AggregateActor extends UntypedActor {
private Map<String, Integer> finalReducedMap = new ConcurrentHashMap<String, Integer>();
@Override
public void onReceive(Object message) throws Exception {
if(message instanceof ReduceData){
ReduceData reduceData = (ReduceData)message;
aggregateInMemoryReduce(reduceData.getReduceDataMap());
}else if(message instanceof Result){
System.out.println(finalReducedMap.toString());
}else{
unhandled(message);
}
}
private void aggregateInMemoryReduce(Map<String,Integer> reducedMap){
Integer count = null;
for(String key : reducedMap.keySet()){
if(finalReducedMap.containsKey(key)){
count = reducedMap.get(key) + finalReducedMap.get(key);
finalReducedMap.put(key, count);
}else{
finalReducedMap.put(key, reducedMap.get(key));
}
}
}
}

入口：

点击(此处)折叠或打开

/**入口
*
*/
public class HelloAkka {
public static void main(String[] args) throws Exception{
ActorSystem _system = ActorSystem.create("helloAkka");
//通过ActorSystem创建Actor
ActorRef master = _system.actorOf(new Props(MasterActor.class), "master");
//main所在的Actor给masterActor发消息，异步，非阻塞
master.tell("Hi! My name is Rocky. I'm so so so happy to be here.");
master.tell("Today, I'm going to read a news article for you.");
master.tell("I hope I hope you'll like it.");
Thread.sleep(500);
master.tell(new Result());
Thread.sleep(500);
_system.shutdown();
}
}

运行结果：
{rocky.=1, you.=1, hope=2, here.=1, my=1, article=1, be=1, so=3, i'm=2, hi!=1, happy=1, name=1, read=1, going=1, it.=1, like=1, you'll=1, i=2, for=1, today,=1, news=1, to=2}

Process finished with exit code 0

ITPub博客

Akka之WordCount

最新文章