我正在尝试将 AspectJ 与 MapReduce 示例一起使用,尽管我不明白一件事。但首先,让我给你我拥有的代码。
[1]字数统计示例
package org.apache.hadoop.mapred.examples;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;
/**
* Common Wordcount example
*/
public class WordCount {
public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
while (tokenizer.hasMoreTokens()) {
word.set(tokenizer.nextToken());
output.collect(word, one);
}
}
}
public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
}
}
public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(WordCount.class);
conf.setJobName("wordcount");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);
conf.setNumReduceTasks(2);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
JobClient.runJob(conf);
}
}
[2]我的mapreduce方面
package org.apache.hadoop.mapred.aspects;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
@Aspect
public class MapReduceAspects {
@Before("execution(* map(..))")
public void mymap(JoinPoint joinPoint) {
System.out.println("My Map Execution: " + joinPoint.getArgs() + ":" + joinPoint.getTarget());
Object[] obj = joinPoint.getArgs();
for (Object o : obj){
System.out.println(o.toString());
}
}
@Before("execution(* reduce(..))")
public void myreduce() { System.out.println("My Reduce Execution"); }
@Before("execution(* collect(..))")
public void updatehash(JoinPoint joinPoint) {
System.out.println("Output collect: Args: " + joinPoint.getArgs());
}
}
```
[3]bean-aspects.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
xmlns:jee="http://www.springframework.org/schema/jee" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.2.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.2.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd">
<aop:aspectj-autoproxy proxy-target-class="true">
<aop:include name="mapreduceAspect"/>
</aop:aspectj-autoproxy>
<bean id="mapreduceAspect" class="org.apache.hadoop.mapred.aspects.MapReduceAspects"/></beans>
[4]OutputCollector接口(interface)
package org.apache.hadoop.mapred;
import java.io.IOException;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Stable;
@Public
@Stable
public interface OutputCollector<K, V> {
void collect(K var1, V var2) throws IOException;
}
在 [1] 中,我有一个带有 map
和 reduce
函数的字数统计示例。当我在 MapReduce 框架中启动应用程序时,该框架将创建一个作业来执行 map
和 reduce
函数。 map
函数接受输入目录,reduce
函数输出结果。
我可以使用AspectJ拦截map
和reduce
函数调用,但无法拦截指令中的
位于 collect
调用>output.collect(word, one)map
函数中。为什么会出现这种情况?是不是因为接口(interface)中的collect
方法没有注解[4]?或者我没有正确配置方面?
如果有人能向我解释为什么 AspectJ 会这样,我会很高兴。
谢谢
最佳答案
答案很简单:
map
和reduce
方法位于您自己的代码中,即它们受到execution()
切入点的方面编织的影响。<collect
方法位于第三方库中,在正常情况下不受切面编织的影响。因此,您不能使用execution()
切入点拦截它,只能使用call()
切入点拦截它。- 尝试类似
call(*collect(..))
的方法,它会起作用。 - 警告:Spring AOP 不支持
call()
切入点,您必须使用成熟的 AspectJ 才能使用它。请参阅 Spring 手册,章节 10.8 Using AspectJ with Spring applications了解有关如何激活 AspectJ LTW(加载时编织)的更多信息。 - 如果您通过 LTW 或二进制 CTW(编译时编织)使用完整的 AspectJ,还可以编织第 3 方代码并使用
execution()
切入点。您只需要确保编织代理在任何目标第三方代码之前加载,通常就是这种情况,因为这就是发明 Java 代理的目的。
关于java - 无法使用方面拦截对方付费调用,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32476730/