KafkaAPI實現生產者與消費者自定義Partition，奇數、偶數資料分在不同的Partition.

思路

建立三個類，包括Consumer、Producer、Partition在Producer端產生訊息，Consumer端接收訊息，Partition實現分割槽規則。Producer根據隨機函式隨機產生十個資料，其中包括奇數和偶數。因為是將奇數和偶數劃分到不同的partition當中去，所以在實現的分割槽的時候使用的是取模的方法。

• 設定producer的配置資訊

  public Producer(String topic) {
        Properties props = new Properties();
 

        props.put("bootstrap.servers", "172.17.11.182:9092,172.17.11.183:9092");
        props.put("client.id", "DemoProducer");
        props.put("batch.size", 16384);//16M
        props.put("linger.ms", 1000);
        props.put("buffer.memory", 33554432);//32M
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"
 
);
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("partitioner.class", "com.horizon.kafka.v010.SimplePartitioner");
        producer = new KafkaProducer<>(props);
        this.topic = topic;
    }

• 傳送訊息佇列，使用迴圈的方法，後面的number和runtime是鍵值對的形式

public void producerMsg() throws InterruptedException{
        Random rnd = new Random();
        int events = 10;
        for (long nEvents = 0; nEvents < events; nEvents++) {
            long  runtime = new Date().getTime();
            String number = String.valueOf(rnd.nextInt(255));
            try {
                producer.send(new ProducerRecord<>(topic, number, String.valueOf(runtime)));
                System.out.println("Sent message: (" + number + ", " +runtime+ ")");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        Thread.sleep(10000);
    }

• Consumer讀取資訊，同樣使用迴圈將key和value以及分割槽全部打印出來

public void consumerMsg(){
        try {
            consumer.subscribe(Collections.singletonList(this.topic));
            while(true){
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at partition "+record.partition()+" offset " + record.offset());
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

• 分割槽規則的建立

public class SimplePartitioner implements Partitioner {

      public SimplePartitioner() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        String k = (String) key;
        if(Integer.parseInt(k) % 2 == 0)
            return 0;
        else
            return 1;
      // return Integer.parseInt(k)%2;

    }

    @Override
    public void close() {
    }


}

在建立分割槽的時候，如果不建立分割槽0，系統也會自動創建出0，在下面的程式中建立的分割槽是1和2，但是也會自動建立0分割槽，即一共會有三個分割槽。

• 結果
  producer端產生的訊息
  
  兩個consumer分別接收到的訊息
  偶數
  
  奇數
  

遇到的問題

1) 在使用單機模式，即Kafka的伺服器只有一個的時候，在Producer端產生訊息的速率特別慢，然後在consumer端接收不到訊息，放到前端去跑，會發現報錯。
kafka.admin.AdminOperationException: replication factor: 1 larger than available brokers: 0
解決辦法: 複製kafka/config路徑下的server.properties檔案為：server-1.properties和server-2.properties
並修改這兩個檔案的配置項：
server-1.properties
broker.id=1
port=9093
log.dir=/tmp/kafka-logs-1
host.name=localhost
server-2.properties
broker.id=2
port=9094
log.dir=/tmp/kafka-logs-2
host.name=localhost
然後重新啟動。就可以了
2) 在解決第一個問題之後，使用分割槽的時候還是出現問題，最後採用叢集的模式，但是在控制檯端執行的時候，先執行consumer，在執行producer發現consumer端並沒有分區劃分log日誌如下

解決方法：在執行的時候先執行producer、在同時執行兩個consumer、執行producer觀察開啟的consumer接收到的實時訊息。具體原因是為什麼也沒有找到，不過感覺可能是，如果先啟動consumer的話可能預先還沒有建立分割槽也就是沒有執行producer，所以找不到對應的分割槽資訊，所以先建立分割槽，第二次是用來傳遞訊息。
全部程式碼待我上傳到github上在發出來，如果現在需要直接私信我就好