Kafka使用java API创建分区

Question

我已经编写了一个代码来使用kafka获取twitter推文，它的工作正常，但它不适用于分区。我想创建一个话题3个分区..如何将值传递给分区类..任何地方我做错了

public class kafkaSpoutFetchingRealTweets { 


private String consumerKey; 
private String consumerSecret; 
private String accessToken; 
private String accessTokenSecret; 
private TwitterStream twitterStream; 

/** 
* @param contxt 
*/ 
void start(final Context context) { 

    /** Producer properties **/ 
    Properties props = new Properties(); 
    props.put("metadata.broker.list", 
      context.getString(Constant.BROKER_LIST)); 
    props.put("partitioner.class","SimplePartitioner"); 
    props.put("serializer.class", context.getString(Constant.SERIALIZER)); 
    props.put("request.required.acks", 
      context.getString(Constant.REQUIRED_ACKS)); 
    props.put("producer.type", "async"); 
    // props.put("partitioner.class", context.getClass()); 
    ProducerConfig config = new ProducerConfig(props); 

    final Producer<String, String> producer = new Producer<String, String>(
      config); 

    /** Twitter properties **/ 
    consumerKey = context.getString(Constant.CONSUMER_KEY_KEY); 
    consumerSecret = context.getString(Constant.CONSUMER_SECRET_KEY); 
    accessToken = context.getString(Constant.ACCESS_TOKEN_KEY); 
    accessTokenSecret = context.getString(Constant.ACCESS_TOKEN_SECRET_KEY); 

    ConfigurationBuilder cb = new ConfigurationBuilder(); 
    cb.setOAuthConsumerKey(consumerKey); 
    cb.setOAuthConsumerSecret(consumerSecret); 
    cb.setOAuthAccessToken(accessToken); 
    cb.setOAuthAccessTokenSecret(accessTokenSecret); 
    cb.setJSONStoreEnabled(true); 
    cb.setIncludeEntitiesEnabled(true); 

    twitterStream = new TwitterStreamFactory(cb.build()).getInstance(); 

    /** Twitter listener **/ 
    StatusListener listener = new StatusListener() { 
     // The onStatus method is executed every time a new tweet comes 
     // in. 
     public void onStatus(Status status) { 


      if(("en".equals(status.getLang())) && ("en".equals(status.getUser().getLang()))){ 

       KeyedMessage<String, String> data = new KeyedMessage<String, String>(
         context.getString(Constant.data), 
         DataObjectFactory.getRawJSON(status)); 
       producer.send(data); 
       System.out.println(DataObjectFactory.getRawJSON(status)); 

      } 
     } 
     } 


     public void onDeletionNotice(
       StatusDeletionNotice statusDeletionNotice) { 
     } 

     public void onTrackLimitationNotice(int numberOfLimitedStatuses) { 
     } 

     public void onScrubGeo(long userId, long upToStatusId) { 
     } 

     public void onException(Exception ex) { 
      ex.printStackTrace(); 
      logger.info("Shutting down Twitter sample stream..."); 
      twitterStream.shutdown(); 
     } 

     public void onStallWarning(StallWarning warning) { 
      System.out.println("stallWarning"); 
     } 
    }; 


    String[] lang = { "en" }; 
    fq.language(lang); 
    twitterStream.addListener(listener); 
    twitterStream.sample(); 

} 

public static void main(String[] args) { 
    try { 

     Context context = new Context(args[0]); 
     kafkaSpoutFetchingRealTweets tp = new kafkaSpoutFetchingRealTweets(); 
     tp.start(context); 

    } catch (Exception e) { 
     e.printStackTrace(); 
     logger.info(e.getMessage()); 
    } 

} 

}

Answer 1

建议，以便有几个问题。

• 你的问题和代码不匹配。你的问题是关于用3个分区创建一个主题。但是，您提供的代码和示例说明了如何确定将消息发送到哪个分区，因为您已经创建了包含3个分区的主题。
• 如果您实际上想要创建包含3个分区的主题，则需要使用命令行客户端。可以在这里找到一个示例，http://kafka.apache.org/documentation.html#quickstart

• 如果您确实想确定您需要发送数据的分区。您需要提供有关您遇到的实际问题的更多信息？他们都去了同一个分区吗？然后你需要看看你在你的config中指定的SimplePartitioner类中如何计算分区。什么是SimplePartitioner类？

  props.put("partitioner.class","SimplePartitioner");