kafka指定partition生產消費

在實際的業務中，特別是涉及到指定任務是否結束，任務對應訊息是否消費完畢時，單純指定topic消費，由kafka自動分配partition已經無法滿足我們的實際需求了，這時我們就需要指定partition進行生產與消費。閒話少說，下面我們通過程式碼來詳細描述生產者與消費者的配置。
producer程式碼
注意：producer程式碼中我們需要兩個類，一個時指定partitioner的類，一個為真正的producer類.
指定partitioner的類

package com.wshare.common;

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

import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created by flyfy on 2017/11/15.
 */
public class KafkaPartitionerProducerUtil {
    private final Producer<String, String> producer;
    private
 
 final String topic = "test";
    private final String kafkaQueueHost = "1.1.1.1";
    private final int kafkaQueuePort = 9092;
    private Properties props = new Properties();

    private BlockingQueue<KeyedMessage<String, String>> blockingQueue = new LinkedBlockingDeque<KeyedMessage<String, String>>();

    public
 
 KafkaPartitionerProducerUtil() {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", kafkaQueueHost + ":" + kafkaQueuePort);
        props.put("partitioner.class","com.wshare.common.CidPartitioner");
        this.producer = new Producer<String, String>(new ProducerConfig(props));
    }

    public void send(String key,String msg) {
        producer.send(new KeyedMessage<String, String>(topic,key,msg));
    }
}

producer類

package com.wshare.common;

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

import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created by flyfy on 2017/11/15.
 */
public class KafkaPartitionerProducerUtil {
    private final Producer<String, String> producer;
    private final String topic = "test";
    private final String kafkaQueueHost = "1.1.1.1";
    private final int kafkaQueuePort = 9092;
    private Properties props = new Properties();

    private BlockingQueue<KeyedMessage<String, String>> blockingQueue = new LinkedBlockingDeque<KeyedMessage<String, String>>();

    public KafkaPartitionerProducerUtil() {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", kafkaQueueHost + ":" + kafkaQueuePort);
        props.put("partitioner.class","com.wshare.common.CidPartitioner");
        this.producer = new Producer<String, String>(new ProducerConfig(props));
    }

    public void send(String key,String msg) {
        producer.send(new KeyedMessage<String, String>(topic,key,msg));
    }
}

注意props.put(“partitioner.class”,”com.wshare.common.CidPartitioner”)即指定我們定義好的partitioner類

consumer程式碼

package com.lyf.scandatatofile.Utils;

import kafka.cluster.Partition;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.io.File;
import java.io.FileWriter;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.BlockingQueue;

/**
 * Created by flyfy on 2017/10/7.
 */
public class KafkaConsumerUtil {
    final String topic = "test";
    final Properties props = new Properties();
    public void execute(String groupid, BlockingQueue<String> bq) {
        props.put("bootstrap.servers", "1.1.1.1:9092");
        props.put("group.id", "test");
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //consumer.subscribe(Arrays.asList("IP_REGISTER_yangkai"));
        consumer.assign(Arrays.asList(new TopicPartition("kafka.wshare.match_meta_data.topic",0)));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                 System.out.println("offset"+record.offset()+"key"
                        +record.key()+"value"+record.value());
        }
    }
}

消費者有一處需要特別注意consumer.subscribe(Arrays.asList(“IP_REGISTER_yangkai”))與consumer.assign(Arrays.asList(new TopicPartition(“kafka.wshare.match_meta_data.topic”,0)))對kafka而言是互斥的，二者只能指定其一，前者指定消費模式為指定topic,由kafka指定partition分配策略，後者指定消費模式為指定partition,由使用者自定義partition分配策略。