Kafka : Cant 복수 스트림 소비자 생성

Question

방금 ​​Kafka 0.8 베타 1을 실행했습니다. 저는 정말 간단한 예제를 작성하여 실행하고 있습니다. 문제는 하나의 메시지 소비자를 여러 명이 아니라 하나씩 만 얻을 수 있다는 것입니다. 즉, runSingleWorker() 메소드가 작동합니다. 실행() 메소드가 작동하지 않습니다

import kafka.consumer.ConsumerIterator; 
import kafka.consumer.KafkaStream; 
import kafka.consumer.ConsumerConfig; 
import kafka.javaapi.consumer.ConsumerConnector; 

import java.util.Map; 
import java.util.List; 
import java.util.HashMap; 
import java.util.concurrent.Executors; 
import java.util.concurrent.ExecutorService; 

import org.springframework.context.ApplicationContext; 
import org.springframework.context.annotation.AnnotationConfigApplicationContext; 

import com.truecar.inventory.worker.core.application.config.AppConfig; 

public class ConsumerThreadPool { 

    private final ConsumerConnector consumer; 
    private final String topic; 

    private ExecutorService executor; 
    private static ApplicationContext context = new AnnotationConfigApplicationContext(AppConfig.class); 

    public ConsumerThreadPool(String topic) { 
     consumer = kafka.consumer.Consumer.createJavaConsumerConnector((ConsumerConfig)context.getBean("consumerConfig")); 
     this.topic = topic; 
    } 

    public void shutdown() { 
     if (consumer != null) consumer.shutdown(); 
     if (executor != null) executor.shutdown(); 
    } 

    public void run(Integer numThreads) { 
     Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); 

     topicCountMap.put(topic, numThreads); 
     Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); 
     List<KafkaStream<byte[], byte[]>> topicListeners = consumerMap.get(topic); 

     executor = Executors.newFixedThreadPool(numThreads); 

     for(Integer i = 0; i < numThreads; i++){ 
      KafkaStream<byte[], byte[]> stream = topicListeners.get(i); 
      executor.submit(new Consumer(stream, i)); 
     } 
    } 


    public void runSingleWorker(Integer numThreads) { 
     Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); 

     topicCountMap.put(topic, new Integer(1)); 

     Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); 

     KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0); 
     ConsumerIterator<byte[], byte[]> it = stream.iterator(); 
     while(true) { 
      try { 
       Thread.sleep(1000); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
      while(it.hasNext()){ 
       System.out.println(new String(it.next().message())); 

      } 
     } 
    } 
} 

그리고 내 장난감 소비자 내부 :

Created iterator empty iterator thread number 3 
Created iterator empty iterator thread number 6 
Created iterator empty iterator thread number 9 
Created iterator empty iterator thread number 7 
Created iterator empty iterator thread number 0 
Created iterator empty iterator thread number 0 
Created iterator empty iterator thread number 8 
Created iterator empty iterator thread number 3 
etc... 

:

이 문제가

import kafka.consumer.KafkaStream; 
import kafka.consumer.ConsumerIterator; 

public class Consumer implements Runnable { 

    private KafkaStream kafkaStream; 
    private Integer threadNumber; 

    public Consumer(KafkaStream kafkaStream, Integer threadNumber) { 
     this.threadNumber = threadNumber; 
     this.kafkaStream = kafkaStream; 
    } 

    public void run() { 
     ConsumerIterator<byte[], byte[]> it = kafkaStream.iterator(); 
     System.out.println("Created iterator " + it.toString() + " thread number " + threadNumber); 
     while(true) { 

      try { 
       Thread.sleep(1000); 
      } catch (InterruptedException e) { 
       break; 
      } 

      while(it.hasNext()) { 
       System.out.println("Thread " + threadNumber + ": " + new String(it.next().message())); 
      } 
     } 
     System.out.println("Shutting down Thread: " + threadNumber); 
    } 
} 

, 노동자의 풀 메시지를 선택하지 않습니다 생산 명령 줄을 통해 메시지를 추가하면 메시지는 단일 스레드 작업자 버전 아래에 인쇄되지만 다중 스트림 상황에서는 메시지가 인쇄되지 않습니다. 여기서 뭐하는거야? 이 문제를 어떻게 해결할 수 있습니까?

는, BTW 카프카 0.8에 대한 pom.xml 파일이 유효한 치어하지 않고 종속성을 취득하지 않을 것이다, 그래서 여기에 완전한 종속성에 치어입니다.

<?xml version="1.0" encoding="UTF-8"?> 
<project 
xmlns="http://maven.apache.org/POM/4.0.0" 
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
xsi:schemaLocation=" 
    http://maven.apache.org/POM/4.0.0 
    http://maven.apache.org/xsd/maven-4.0.0.xsd"> 
<modelVersion>4.0.0</modelVersion> 
<groupId>group1</groupId> 
<artifactId>artifact1</artifactId> 
<version>0.1.0</version> 
<packaging>jar</packaging> 
<properties> 
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> 
    <org.springframework.version>3.2.4.RELEASE</org.springframework.version> 
</properties> 
<dependencies> 
    <dependency> 
     <groupId>org.springframework</groupId> 
     <artifactId>spring-core</artifactId> 
     <version>3.2.4.RELEASE</version> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework</groupId> 
     <artifactId>spring-context</artifactId> 
     <version>3.2.4.RELEASE</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.kafka</groupId> 
     <artifactId>kafka_2.9.2</artifactId> 
     <version>0.8.0-beta1</version> 
    </dependency> 
    <dependency> 
     <groupId>javax.inject</groupId> 
     <artifactId>javax.inject</artifactId> 
     <version>1</version> 
    </dependency> 
    <dependency> 
     <groupId>org.scala-lang</groupId> 
     <artifactId>scala-library</artifactId> 
     <version>2.9.2</version> 
    </dependency> 
    <dependency> 
     <groupId>log4j</groupId> 
     <artifactId>log4j</artifactId> 
     <version>1.2.17</version> 
    </dependency> 
    <dependency> 
     <groupId>com.101tec</groupId> 
     <artifactId>zkclient</artifactId> 
     <version>0.3</version> 
    </dependency> 
    <dependency> 
     <groupId>com.yammer.metrics</groupId> 
     <artifactId>metrics-core</artifactId> 
     <version>2.2.0</version> 
    </dependency> 
</dependencies> 
<build> 
    <finalName>inventory-core</finalName> 
    <plugins> 
     <plugin> 
      <groupId>org.apache.maven.plugins</groupId> 
      <artifactId>maven-compiler-plugin</artifactId> 
      <version>3.0</version> 
      <configuration> 
       <source>1.7</source> 
       <target>1.7</target> 
      </configuration> 
     </plugin> 
     <plugin> 
      <groupId>org.apache.maven.plugins</groupId> 
      <artifactId>maven-jar-plugin</artifactId> 
      <configuration> 
       <archive> 
        <manifest> 
         <mainClass>com.truecar.inventory.worker.core.application.Starter</mainClass> 
        </manifest> 
       </archive> 
      </configuration> 
     </plugin> 
     <plugin> 
      <groupId>org.dstovall</groupId> 
      <artifactId>onejar-maven-plugin</artifactId> 
      <version>1.4.4</version> 
      <executions> 
       <execution> 
        <configuration> 
         <onejarVersion>0.97</onejarVersion> 
         <classifier>onejar</classifier> 
        </configuration> 
        <goals> 
         <goal>one-jar</goal> 
        </goals> 
       </execution> 
      </executions> 
     </plugin> 
    </plugins> 
</build> 
<pluginRepositories> 
    <pluginRepository> 
     <id>onejar-maven-plugin.googlecode.com</id> 
     <url>http://onejar-maven-plugin.googlecode.com/svn/mavenrepo</url> 
    </pluginRepository> 
</pluginRepositories> 
</project> 

Answer 1

질문자에게는 너무 늦을 지 모르지만 다른 개발자에게 유용 할 수 있습니다. 몇 명의 소비자를 위해 단 하나의 파티션 만 사용한 것처럼 보입니다. 그건 잘못되었습니다. Documentation에서 견적 : 많은 파티션이 있기 때문에

이 여전히 많은 소비자 인스턴스를 통해 부하 사이의 균형을 맞추 었습니다. 그러나 은 파티션보다 많은 소비자 인스턴스가 될 수 없습니다.

그래서 소비자에 대해 생각할 때 파티션별로 메시지를 나누는 방법을 고려해야합니다. 대부분의 경우 고급 그룹을 사용하거나 기본적으로 임의로 지정해야합니다.