【Kafka六】Kafka Producer和Consumer多Broker、多Partition场景

0.Kafka服务器配置

3个broker

1个topic，6个partition，副本因子是2

2个consumer，每个consumer三个线程并发读取

 

1. Producer

package kafka.examples.multibrokers.producers;

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

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

public class MultiBrokerProducer {
    private static Producer<String, String> producer;
    private static Properties props = new Properties();

    static {
        props.put("metadata.broker.list", "192.168.26.140:9092,192.168.26.140:9093,192.168.26.140:9094");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "kafka.examples.multibrokers.partitioner.TopicPartitioner");
        props.put("request.required.acks", "1");
        ProducerConfig config = new ProducerConfig(props);
        producer = new Producer<String, String>(config);
    }

    public static void main(String[] args) {
        Random rnd = new Random();
        String topic = "learn.topic.p8.r2";
        for (long i = 0; i < 10000; i++) {
            String key = "" + rnd.nextInt(255);
            String msg = "The " + i + " message is for key - " + key;
            KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, key, msg);
            producer.send(data);
           System.out.println(i);
        }
        producer.close();
    }
}

 

 

2. Partitioner

package kafka.examples.multibrokers.partitioner;


import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

import java.util.Random;

public class TopicPartitioner implements Partitioner {

    public TopicPartitioner(VerifiableProperties props) {

    }

    @Override
    public int partition(Object key, int numPartitions) {
        int hashCode;
        if (key == null) {
            hashCode = new Random().nextInt(255);
        } else {
            hashCode = key.hashCode();
        }
        if (numPartitions <= 0) {
            return 0;
        }
        return hashCode % numPartitions;
    }
}

 

3. Consumer

package kafka.examples.multibrokers.consumers;


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

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

public class MultiThreadHLConsumer {
    private ExecutorService executor;
    private final ConsumerConnector consumer;
    private final String topic;

    public MultiThreadHLConsumer(String zookeeper, String groupId,
                                 String topic) {
        Properties props = new Properties();
        props.put("zookeeper.connect", zookeeper);
        props.put("group.id", groupId);
        props.put("zookeeper.session.timeout.ms", "500");
        props.put("zookeeper.sync.time.ms", "250");
        props.put("auto.commit.interval.ms", "1000");
        consumer = Consumer.createJavaConsumerConnector(new
                ConsumerConfig(props));
        this.topic = topic;
    }

    public void doConsume(int threadCount) {
        Map<String, Integer> topicCount = new HashMap<String, Integer>();
        // Define thread count for each topic
        topicCount.put(topic, new Integer(threadCount));
        // Here we have used a single topic but we can also add multiple topics to topicCount MAP
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer.createMessageStreams(topicCount);
        List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);
        System.out.println("streams length: " + streams.size());
        // Launching the thread pool
        executor = Executors.newFixedThreadPool(threadCount);
        //Creating an object messages consumption
        final CountDownLatch latch = new CountDownLatch(3);
        for (final KafkaStream stream : streams) {
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    ConsumerIterator<byte[], byte[]> consumerIte = stream.iterator();
                    while (consumerIte.hasNext()) {
                        System.out.println("Message from thread :: " + Thread.currentThread().getName() + " -- " + new String(consumerIte.next().message()));
                    }
                    latch.countDown();
                }
            });
        }

        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

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

        if (executor != null)
            executor.shutdown();
    }

    public static void main(String[] args) {
        String topic = "learn.topic.p8.r2";
        int threadCount = 3;
        MultiThreadHLConsumer simpleHLConsumer = new MultiThreadHLConsumer("192.168.26.140:2181", "learn.topic.p8.r2.consumers.group", topic);
        simpleHLConsumer.doConsume(threadCount);
    }
}

 

 

4. 注意的问题