因此，我目前正在学习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
}