Kafka学习笔记-Java简单操作

Maven依赖包：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.8.2.1</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.8.2.1</version>
</dependency>

代码如下：

import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaProducerTest {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducerTest.class);

    private static Properties properties = null;

    static {
        properties = new Properties();
        properties.put("bootstrap.servers", "centos.master:9092,centos.slave1:9092,centos.slave2:9092");
        properties.put("producer.type", "sync");
        properties.put("request.required.acks", "1");
        properties.put("serializer.class", "kafka.serializer.DefaultEncoder");
        properties.put("partitioner.class", "kafka.producer.DefaultPartitioner");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
//      properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
//      properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }

    public void produce() {
        KafkaProducer<byte[], byte[]> kafkaProducer = new KafkaProducer<byte[],byte[]>(properties);
        ProducerRecord<byte[],byte[]> kafkaRecord = new ProducerRecord<byte[],byte[]>(
                "test", "kkk".getBytes(), "vvv".getBytes());
        kafkaProducer.send(kafkaRecord, new Callback() {
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if(null != e) {
                    LOG.info("the offset of the send record is {}", metadata.offset());
                    LOG.error(e.getMessage(), e);
                }
                LOG.info("complete!");
            }
        });
        kafkaProducer.close();
    }

    public static void main(String[] args) {
        KafkaProducerTest kafkaProducerTest = new KafkaProducerTest();
        for (int i = 0; i < 10; i++) {
            kafkaProducerTest.produce();
        }
    }
}

import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaConsumerTest {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerTest.class);

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "centos.master:9092,centos.slave1:9092,centos.slave2:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS, "1000");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY, "range");
//      properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY, "roundrobin");
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "10000");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.ByteArrayDeserializer");

        KafkaConsumer<byte[], byte[]> kafkaConsumer = new KafkaConsumer<byte[], byte[]>(properties);
        kafkaConsumer.subscribe("test");
//      kafkaConsumer.subscribe("*");
        boolean isRunning = true;
        while(isRunning) {
            Map<String, ConsumerRecords<byte[], byte[]>> results = kafkaConsumer.poll(100);
            if (null != results) {
                for (Map.Entry<String, ConsumerRecords<byte[], byte[]>> entry : results.entrySet()) {
                    LOG.info("topic {}", entry.getKey());
                    ConsumerRecords<byte[], byte[]> consumerRecords = entry.getValue();
                    List<ConsumerRecord<byte[], byte[]>> records = consumerRecords.records();
                    for (int i = 0, len = records.size(); i < len; i++) {
                        ConsumerRecord<byte[], byte[]> consumerRecord = records.get(i);
                        LOG.info("topic {} partition {}", consumerRecord.topic(), consumerRecord.partition());
                        try {
                            LOG.info("offset {} value {}", consumerRecord.offset(), new String(consumerRecord.value()));
                        } catch (Exception e) {
                            LOG.error(e.getMessage(), e);
                        }
                    }
                }
            }
        }

        kafkaConsumer.close();

    }

}

发现KafkaConsumer的poll方法未实现

@Override
public Map<String, ConsumerRecords<K,V>> poll(long timeout) {
     // TODO Auto-generated method stub
     return null;
}

后改为kafka.javaapi.consumer.SimpleConsumer实现，正常运行

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.cluster.Broker;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetRequest;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.TopicMetadataResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;

public class KafkaSimpleConsumerTest {

    private List<String> borkerList = new ArrayList<String>();

    public KafkaSimpleConsumerTest() {
        borkerList = new ArrayList<String>();
    }

    public static void main(String args[]) {
        KafkaSimpleConsumerTest kafkaSimpleConsumer = new KafkaSimpleConsumerTest();
        // 最大读取消息数量
        long maxReadNum = Long.parseLong("3");
        // 订阅的topic
        String topic = "test";
        // 查找的分区
        int partition = Integer.parseInt("0");
        // broker节点
        List<String> seeds = new ArrayList<String>();
        seeds.add("centos.master");
        seeds.add("centos.slave1");
        seeds.add("centos.slave2");
        // 端口
        int port = Integer.parseInt("9092");
        try {
            kafkaSimpleConsumer.run(maxReadNum, topic, partition, seeds, port);
        } catch (Exception e) {
            System.out.println("Oops:" + e);
            e.printStackTrace();
        }
    }

    public void run(long maxReadNum, String topic, int partition, List<String> seedBrokers, int port) throws Exception {
        // 获取指定topic partition的元数据
        PartitionMetadata metadata = findLeader(seedBrokers, port, topic, partition);
        if (metadata == null) {
            System.out.println("can't find metadata for topic and partition. exit");
            return;
        }
        if (metadata.leader() == null) {
            System.out.println("can't find leader for topic and partition. exit");
            return;
        }
        String leadBroker = metadata.leader().host();
        String clientName = "client_" + topic + "_" + partition;

        SimpleConsumer consumer = new SimpleConsumer(leadBroker, port, 100000, 64 * 1024, clientName);
        long readOffset = getLastOffset(consumer, topic, partition, kafka.api.OffsetRequest.EarliestTime(), clientName);
        int numErrors = 0;
        while (maxReadNum > 0) {
            if (consumer == null) {
                consumer = new SimpleConsumer(leadBroker, port, 100000, 64 * 1024, clientName);
            }
            FetchRequest req = new FetchRequestBuilder().clientId(clientName).addFetch(topic, partition, readOffset, 100000).build();
            FetchResponse fetchResponse = consumer.fetch(req);

            if (fetchResponse.hasError()) {
                numErrors++;
                short code = fetchResponse.errorCode(topic, partition);
                System.out.println("error fetching data from the broker:" + leadBroker + " reason: " + code);
                if (numErrors > 5)
                    break;
                if (code == ErrorMapping.OffsetOutOfRangeCode()) {
                    readOffset = getLastOffset(consumer, topic, partition, kafka.api.OffsetRequest.LatestTime(), clientName);
                    continue;
                }
                consumer.close();
                consumer = null;
                leadBroker = findNewLeader(leadBroker, topic, partition, port);
                continue;
            }
            numErrors = 0;

            long numRead = 0;
            for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(topic, partition)) {
                long currentOffset = messageAndOffset.offset();
                if (currentOffset < readOffset) {
                    System.out.println("found an old offset: " + currentOffset + " expecting: " + readOffset);
                    continue;
                }

                readOffset = messageAndOffset.nextOffset();
                ByteBuffer payload = messageAndOffset.message().payload();

                byte[] bytes = new byte[payload.limit()];
                payload.get(bytes);
                System.out.println(String.valueOf(messageAndOffset.offset()) + ": " + new String(bytes, "UTF-8"));
                numRead++;
                maxReadNum--;
            }

            if (numRead == 0) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                }
            }
        }
        if (consumer != null)
            consumer.close();
    }

    /**
     * 从活跃的Broker列表中找出指定Topic、Partition中的Leader Broker
     * @param seedBrokers
     * @param port
     * @param topic
     * @param partition
     * @return
     */
    private PartitionMetadata findLeader(List<String> seedBrokers, int port, String topic, int partition) {
        PartitionMetadata partitionMetadata = null;
        loop: for (String seedBroker : seedBrokers) {
            SimpleConsumer consumer = null;
            try {
                consumer = new SimpleConsumer(seedBroker, port, 100000, 64 * 1024, "leaderLookup");
                List<String> topics = Collections.singletonList(topic);
                TopicMetadataRequest topicMetadataRequest = new TopicMetadataRequest(topics);
                TopicMetadataResponse topicMetadataResponse = consumer.send(topicMetadataRequest);

                List<TopicMetadata> topicMetadatas = topicMetadataResponse.topicsMetadata();
                for (TopicMetadata topicMetadata : topicMetadatas) {
                    for (PartitionMetadata pMetadata : topicMetadata.partitionsMetadata()) {
                        if (pMetadata.partitionId() == partition) {
                            partitionMetadata = pMetadata;
                            break loop;
                        }
                    }
                }
            } catch (Exception e) {
                System.out.println("error communicating with broker [" + seedBroker + "] to find leader for [" + topic + ", " + partition + "] reason: " + e);
            } finally {
                if (consumer != null)
                    consumer.close();
            }
        }
        if (partitionMetadata != null) {
            borkerList.clear();
            for (Broker replica : partitionMetadata.replicas()) {
                borkerList.add(replica.host());
            }
        }
        return partitionMetadata;
    }

    public static long getLastOffset(SimpleConsumer consumer, String topic, int partition, long whichTime, String clientName) {
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
        Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
        requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
        OffsetRequest request = new OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
        OffsetResponse response = consumer.getOffsetsBefore(request);
        if (response.hasError()) {
            System.out.println("error fetching data offset data the broker. reason: " + response.errorCode(topic, partition));
            return 0;
        }
        long[] offsets = response.offsets(topic, partition);
        return offsets[0];
    }

    private String findNewLeader(String oldLeader, String topic, int partition, int port) throws Exception {
        for (int i = 0; i < 3; i++) {
            boolean goToSleep = false;
            PartitionMetadata metadata = findLeader(borkerList, port, topic, partition);
            if (metadata == null) {
                goToSleep = true;
            } else if (metadata.leader() == null) {
                goToSleep = true;
            } else if (oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) {
                goToSleep = true;
            } else {
                return metadata.leader().host();
            }
            if (goToSleep) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                }
            }
        }
        System.out.println("unable to find new leader after broker failure. exit");
        throw new Exception("unable to find new leader after broker failure. exit");
    }

}