因此,我目前正在学习Kafka,并试图复制这里从apache提供的示例。这是消费者的示例代码,我用java编写了它,如图所示。当我试图执行文件,但我遇到了一些问题。我可以得到文件编译,但它不会正常运行。
我用下面的行执行这个程序,不带引号,“javatestconsumerlocalhost:2181 group1 测试4“这将传递示例代码中所需的4个参数。但是当我执行这个命令时,出现了以下错误。
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/log4j/Category
at kafka.utils.VerifiableProperties.<init>(Unknown Source)
at kafka.consumer.ConsumerConfig.<init>(Unknown Source)
at TestConsumer.ConsumerProps(TestConsumer.java:69)
at TestConsumer.<init>(TestConsumer.java:31)
at TestConsumer.main(TestConsumer.java:97)
Caused by: java.lang.ClassNotFoundException: org.apache.log4j.Category
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
... 5 more
我曾经尝试过用必要的值手动替换参数,并试图以这种方式执行,但我遇到了另一个问题。下面是错误消息以及我正在使用的代码,以防我从提供的示例中搞砸了一些东西。如果有人能帮我,我会非常感激,因为我正试图编写自己的消费者来测试解析给定的信息,等等。谢谢
log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
log4j:WARN Please initialize the log4j system properly.
Exception in thread "main" java.lang.NoClassDefFoundError: org/I0Itec/zkclient/IZkStateListener
at kafka.javaapi.consumer.ZookeeperConsumerConnector.<init>(Unknown Source)
at kafka.javaapi.consumer.ZookeeperConsumerConnector.<init>(Unknown Source)
at kafka.consumer.Consumer$.createJavaConsumerConnector(Unknown Source)
at kafka.consumer.Consumer.createJavaConsumerConnector(Unknown Source)
at TestConsumer.<init>(TestConsumer.java:31)
at TestConsumer.main(TestConsumer.java:97)
Caused by: java.lang.ClassNotFoundException: org.I0Itec.zkclient.IZkStateListener
at java.net.URLClassLoader.findClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
... 6 more
/*
* Test Consumer to gather input from
* a Producer. Attempt to perform functions
* from the produced data
* /
// Kafka API
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.Map;
import java.util.HashMap;
import java.util.Properties;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
public class TestConsumer{
private final ConsumerConnector consumer;
private final String topic;
private ExecutorService executor;
// CONSTRUCTOR
public TestConsumer(String zookeeper, String groupid, String aTopic){
consumer = kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerProps(zookeeper, groupid));
this.topic = aTopic;
}
// END CONSTRUCTOR
// RUN FUNCTION
public void run(int threads){
Map<String, Integer> topicMap = new HashMap<String, Integer>();
topicMap.put(topic, new Integer(threads));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicMap);
List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
executor = Executors.newFixedThreadPool(threads); // process threads
int numThread = 0; // thread counter for consumption
// consumer all messages
for(final KafkaStream stream : streams){
executor.submit(new TestConsumerRun(stream, numThread));
numThread ++;
}
}
// END RUN FUNCTION
// CREATE PROPERTIES FUNCTION
private static ConsumerConfig ConsumerProps(String zookeeper, String groupid){
Properties properties = new Properties(); // config properties file
properties.put("zookeeper.connect", zookeeper);
properties.put("group.id", groupid);
properties.put("zookeeper.session.timeout.ms", "400");
properties.put("zookeeper.sync.time.ms", "200");
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "smallest");
return new ConsumerConfig(properties);
}
// END CREATE PROPERTIES FUNCTION
// SHUTDOWN FUNCTION
public void shutdown(){
if (consumer != null) consumer.shutdown();
if (executor != null) executor.shutdown();
try{
if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)){
System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");
}
} catch (InterruptedException e){
System.out.println("Interrupted during shutdown, exiting uncleanly");
}
}
// END SHUTDOWN FUNCTION
// MAIN FUNCTION
public static void main(String[] args){
String zookeeper = args[0];
String groupid = args[1];
String topic = args[2];
int threads = Integer.parseInt(args[3]);
TestConsumer test = new TestConsumer(zookeeper, groupid, topic); // send information to constructor
test.run(threads); // pass threads for iteration
try{
Thread.sleep(10000);
} catch (InterruptedException ie){
}
test.shutdown(); // close program
}
// END MAIN FUNCTION
}
/*
* Test Consumer to gather input from
* a Producer. Attempt to perform functions
* from the produced data
* /
// Kafka API
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
public class TestConsumerRun implements Runnable{
private KafkaStream aStream;
private int aThread;
// CONSTRUCTOR
public TestConsumerRun(KafkaStream stream, int thread){
aStream = stream; // set stream from main read
aThread = thread; // set thread from main read
}
// END CONSTRUCTOR
// RUN FUNCTION
public void run(){
ConsumerIterator<byte[], byte[]> iterator = aStream.iterator(); // used to check throughout the list continiously
while(iterator.hasNext())
System.out.println("Thread " + aThread + ": " + new String(iterator.next().message()));
System.out.println("Shutting down Thread: " + aThread);
}
// END RUN FUNCTION
}
1条答案
按热度按时间lskq00tm1#
我也有同样的问题。你需要加上
log4j
jar到你的类路径。您可能还需要添加slf4j
以及commons-logging
.java.lang.NoClassDefFoundError
当jvm在运行时找不到类时发生(但它在编译时就存在了。)当jar在运行时丢失时会发生这种情况,还有许多其他原因。编译和运行时期间的类路径必须相同。有时您可能会使用不同版本的同一个jar,因此在运行时jvm可能会找到不同的版本,而不是编译时使用的版本。