我的Kafka自定義分區程序類錯誤

Question

我正在使用Kafka自定義分區程序類。這裏我試圖將數據推送到單獨的分區。 我卡夫卡製作類：

import java.util.Date; 
import java.util.Properties; 
import java.util.Random; 

import kafka.javaapi.producer.Producer; 
import kafka.producer.KeyedMessage; 
import kafka.producer.ProducerConfig; 

public class KafkaCustomPartitioner { 
    public static void main(String[] args) { 
     long events = Long.parseLong(args[0]); 
     int blocks = Integer.parseInt(args[1]); 
     Random rnd = new Random(); 

     Properties props = new Properties(); 
     props.put("metadata.broker.list", "localhost:9092"); 
     props.put("serializer.class","kafka.serializer.StringEncoder"); 
     props.put("key.serializer.class", "kafka.serializer.StringEncoder"); 
     props.put("partitioner.class","com.kafka.partdecider.CustomPartitioner"); 
     props.put("producer.type", "sync"); 
     props.put("request.required.acks","1"); 

     ProducerConfig config = new ProducerConfig(props); 
     Producer producer = new Producer(config); 

     for(int nBlocks=0; nBlocks<blocks; nBlocks++) { 
      for(long nEvents=0; nEvents<events; nEvents++) { 
       long runTime = new Date().getTime(); 
       String msg = runTime + ": " + (50+nBlocks) + ": " + nEvents + ": " + rnd; 
       KeyedMessage<String, String> data = new KeyedMessage<String, String>("CustPartTopic",String.valueOf(nBlocks),msg); 
       producer.send(data); 
      } 
     } 
     producer.close(); 
    } 
} 

客戶分區程序類：

import kafka.producer.Partitioner; 

public class CustomPartitioner implements Partitioner { 

    public int partition(Object key, int arg1) { 
     String receivingkey = (String) key; 
     long id = Long.parseLong(receivingkey); 
     return (int) (id%arg1); 
    } 
} 

該項目的論證部分的值有：3 2 我得到「ArrayOutOfBoundsException」在這一行，如果我運行的類：

Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 0 
    at com.kafka.custompartitioner.KafkaCustomPartitioner.main(KafkaCustomPartitioner.java:13) 

顯示在該行的錯誤：long events = Long.parseLong(args[0]); 但我不明白爲什麼該行給出錯誤。 任何人都可以讓我知道我該如何解決這個問題？

Answer 1

這工作對我來說，API有很大的不同：

package mypackage.io; 

import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.ProducerConfig; 
import org.apache.kafka.clients.producer.ProducerRecord; 

import java.util.Date; 
import java.util.Properties; 
import java.util.Random; 
import java.util.concurrent.ExecutionException; 

public class KafkaCustomPartitioner { 

public static void main(String[] args) throws InterruptedException, ExecutionException { 

    long events = Long.parseLong(args[0]); 
    int blocks = Integer.parseInt(args[1]); 
    Random rnd = new Random(); 

    Properties props = new Properties(); 
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); 
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "mypackage.io.CustomPartitioner"); 
    props.put(ProducerConfig.ACKS_CONFIG, "1"); 

    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props); 

    for(int nBlocks=0; nBlocks<blocks; nBlocks++) { 

     for(long nEvents=0; nEvents<events; nEvents++) { 

      long runTime = new Date().getTime(); 
      String msg = runTime + ": " + (50+nBlocks) + ": " + nEvents + ": " + rnd; 
      producer.send(new ProducerRecord<String, String>("CustPartTopic", String.valueOf(nBlocks), msg)).get(); 
     } 
    } 
    producer.close(); 
    } 
} 

然後將自定義分區

package mypackage.io; 

import org.apache.kafka.clients.producer.Partitioner; 
import org.apache.kafka.common.Cluster; 

import java.util.Map; 

public class CustomPartitioner implements Partitioner { 

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 

    String receivingkey = (String) key; 
    long id = Long.parseLong(receivingkey); 
    int numPartitions = cluster.availablePartitionsForTopic(topic).size(); 
    return (int) (id % numPartitions); 
} 

public void close() { 

} 

public void configure(Map<String, ?> map) { 

} 
}