Кафка Потребительская на Яве

Question

Поэтому я изучаю Кафку в настоящее время и пытался продублировать примеры, предоставленные здесь Apache. Это пример кода для потребителя, и я написал его в Java, как показано на рисунке. Однако, когда я пытаюсь выполнить файл, я сталкиваюсь с некоторыми проблемами. Я могу получить файл для компиляции, но он не будет работать должным образом.

Я выполняю программу со следующей строкой без кавычек: "java TestConsumer localhost:2181 group1 test 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
}

Answer 1

Попробуйте добавить BasicConfigurator.configure(); в основном методе, и он будет работать нормально.

Answer 2

У меня такая же проблема. Вам нужно добавитьlog4jjar в путь к классам. Также вам может потребоваться добавитьslf4j а также commons-logging.

java.lang.NoClassDefFoundErrorвозникает, когда JVM не может найти класс во время выполнения. (Но это было во время компиляции.) Случается, когда jar отсутствует во время выполнения, а также по многим другим причинам. Путь к классам во время компиляции и выполнения должен совпадать. Иногда у вас может быть одна и та же банка с разными версиями, поэтому во время выполнения JVM может найти другую версию, а не ту, которая используется при компиляции.