Kafka Consumer应用与高级应用
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka Consumer应用与高级应用相关的知识,希望对你有一定的参考价值。
Kafka Consumer应用与高级应用
PS:本博客仅作学习、总结、交流使用,参考以下博客&资料
1.http://kafka.apache.org/intro.html
2.https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
3.http://www.cnblogs.com/luotianshuai/p/5206662.html
4.http://www.cnblogs.com/fxjwind/p/3794255.html
5.https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
一.Kafka使用背景
- 我们想分析下用户行为(pageviews),以便我们设计出更好的广告位
- 我想对用户的搜索关键词进行统计,分析出当前的流行趋势
- 有些数据,存储数据库浪费,直接存储硬盘效率又低
二.Kafka(科技术语)
2.1 特性
-
通过O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能
-
支持通过Kafka服务器和消费机集群来分区消息
-
支持Hadoop并行数据加载
2.2 Kafka相关术语介绍
-
Broker Kafka集群包含一个或多个服务器,这种服务器被称为broker
-
Topic 每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic。(物理上不同Topic的消息分开存储,逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处)
-
Partition Partition是物理上的概念,每个Topic包含一个或多个Partition.
-
Producer 负责发布消息到Kafka broker
-
Consumer 消息消费者,向Kafka broker读取消息的客户端。
-
Consumer Group 每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)
三、KafKa安装&配置 参考 http://www.cnblogs.com/denghongfu/p/6085685.html
四、KafKa Consumer接口
kafka的consumer接口,有两种版本:
A.high-level 比较简单不用关心offset, 会自动的读zookeeper中该Consumer group的last offset
B.就是官网上提供的 SimpleConsumer Example low-level
几点说明:
1. 如果consumer比partition多,是浪费,因为kafka的设计是在一个partition上是不允许并发的,所以consumer数不要大于partition数
2. 如果consumer比partition少,一个consumer会对应于多个partitions,这里主要合理分配consumer数和partition数,否则会导致partition里面的数据被取的不均匀
最好partiton数目是consumer数目的整数倍,所以partition数目很重要,比如取24,就很容易设定consumer数目
3. 如果consumer从多个partition读到数据,不保证数据间的顺序性,kafka只保证在一个partition上数据是有序的,但多个partition,根据你读的顺序会有不同
4. 增减consumer,broker,partition会导致rebalance,所以rebalance后consumer对应的partition会发生变化
5. High-level接口中获取不到数据的时候是会block的
1.High-level
如果测试流程是,先produce一些数据,然后再用consumer读的话,记得加上第一句设置因为初始的offset默认是非法的,然后这个设置的意思是,当offset非法时,如何修正offset,默认是largest,即最新,所以不加这个配置,你是读不到你之前produce的数据的,而且这个时候你再加上smallest配置也没用了,因为此时offset是合法的,不会再被修正了,需要手工或用工具改重置offset
package com.tydic.kafka.client;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
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;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringEncoder;
/**
* Description:
* Created by hadoop on 2016/11/9.
* Date 2016/11/9
*/
public class ConsumerKafka {
private ConsumerConfig config;
private String topic;
private int partitionsNum;
private MessageExecutor executor;
private ConsumerConnector connector;
private ExecutorService threadPool;
public ConsumerKafka(String topic,int partitionsNum,MessageExecutor executor) throws Exception{
Properties prop = new Properties();
prop.put("auto.offset.reset", "smallest"); //必须要加,如果要读旧数据
prop.put("zookeeper.connect", "cna3:2181,cna4:2181,cna5:2181");
prop.put("serializer.class", StringEncoder.class.getName());
prop.put("metadata.broker.list", "cna3:9092,cna4:9092,cna5:9092");
prop.put("group.id", "test-consumer-group");
config = new ConsumerConfig(prop);
this.topic = topic;
this.partitionsNum = partitionsNum;
this.executor = executor;
}
public void start() throws Exception{
connector = Consumer.createJavaConsumerConnector(config);
Map<String,Integer> topics = new HashMap<String,Integer>();
topics.put(topic, partitionsNum);
Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(topics);
List<KafkaStream<byte[], byte[]>> partitions = streams.get(topic);
threadPool = Executors.newFixedThreadPool(partitionsNum);
for(KafkaStream<byte[], byte[]> partition : partitions){
threadPool.execute(new MessageRunner(partition));
}
}
public void close(){
try{
threadPool.shutdownNow();
}catch(Exception e){
//
}finally{
connector.shutdown();
}
}
class MessageRunner implements Runnable{
private KafkaStream<byte[], byte[]> partition;
MessageRunner(KafkaStream<byte[], byte[]> partition) {
this.partition = partition;
}
public void run(){
ConsumerIterator<byte[], byte[]> it = partition.iterator();
while(it.hasNext()){
MessageAndMetadata<byte[],byte[]> item = it.next();
System.out.println("partiton:" + item.partition());
System.out.println("offset:" + item.offset());
executor.execute(new String(item.message()));//UTF-8
}
}
}
interface MessageExecutor {
public void execute(String message);
}
/**
* @param args
*/
public static void main(String[] args) {
ConsumerKafka consumer = null;
try{
MessageExecutor executor = new MessageExecutor() {
public void execute(String message) {
System.out.println(message);
}
};
consumer = new ConsumerKafka("topic1",3, executor);
consumer.start();
}catch(Exception e){
e.printStackTrace();
}finally{
if(consumer != null){
consumer.close();
}
}
}
}
在用high-level的consumer时,两个给力的工具,
1. bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv
可以看到当前group offset的状况,比如这里看pv的状况,3个partition
Group Topic Pid Offset logSize Lag Owner
pv page_visits 0 21 21 0 none
pv page_visits 1 19 19 0 none
pv page_visits 2 20 20 0 none
关键就是offset,logSize和Lag
这里以前读完了,所以offset=logSize,并且Lag=0
2. bin/kafka-run-class.sh kafka.tools.UpdateOffsetsInZK earliest config/consumer.properties page_visits
3个参数,
[earliest | latest],表示将offset置到哪里
consumer.properties ,这里是配置文件的路径
topic,topic名,这里是page_visits
我们对上面的pv group执行完这个操作后,再去check group offset状况,结果如下,
Group Topic Pid Offset logSize Lag Owner
pv page_visits 0 0 21 21 none
pv page_visits 1 0 19 19 none
pv page_visits 2 0 20 20 none
可以看到offset已经被清0,Lag=logSize
多线程consumer的完整代码
import kafka.consumer.ConsumerConfig; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class ConsumerGroupExample { private final ConsumerConnector consumer; private final String topic; private ExecutorService executor; public ConsumerGroupExample(String a_zookeeper, String a_groupId, String a_topic) { consumer = kafka.consumer.Consumer.createJavaConsumerConnector( // 创建Connector,注意下面对conf的配置 createConsumerConfig(a_zookeeper, a_groupId)); this.topic = a_topic; } public void shutdown() { if (consumer != null) consumer.shutdown(); if (executor != null) executor.shutdown(); } public void run(int a_numThreads) { // 创建并发的consumers Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); topicCountMap.put(topic, new Integer(a_numThreads)); // 描述读取哪个topic,需要几个线程读 Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); // 创建Streams List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic); // 每个线程对应于一个KafkaStream // now launch all the threads // executor = Executors.newFixedThreadPool(a_numThreads); // now create an object to consume the messages // int threadNumber = 0; for (final KafkaStream stream : streams) { executor.submit(new ConsumerTest(stream, threadNumber)); // 启动consumer thread threadNumber++; } } private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) { Properties props = new Properties(); props.put("zookeeper.connect", a_zookeeper); props.put("group.id", a_groupId); props.put("zookeeper.session.timeout.ms", "400"); props.put("zookeeper.sync.time.ms", "200"); props.put("auto.commit.interval.ms", "1000"); return new ConsumerConfig(props); } public static void main(String[] args) { String zooKeeper = args[0]; String groupId = args[1]; String topic = args[2]; int threads = Integer.parseInt(args[3]); ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic); example.run(threads); try { Thread.sleep(10000); } catch (InterruptedException ie) { } example.shutdown(); } }
2.low-level SimpleConsumer Example
package com.tydic.kafka.client; 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 com.alibaba.fastjson.JSON; import com.tydic.kafka.util.Utils; 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.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; /** * Description:kafka消费者实现,关注信息消费位置 * Created by hadoop on 2016/11/18. * Date 2016/11/18 */ public class KafkaSimpleConsumer { public static void main(String arg[]) { KafkaSimpleConsumer example = new KafkaSimpleConsumer(); List<String> seeds = new ArrayList<>(); int maxReads = 0; int partition = 0; int port = 0; String topic = ""; try { System.out.println(JSON.toJSONString(arg, true)); Map<String, Object> paramMap = Utils.parseParam(arg); System.out.println(JSON.toJSONString(paramMap, true)); if (null == paramMap || paramMap.size() == 0) { return; } maxReads = (int) paramMap.get("consumer.maxReads"); partition = (int) paramMap.get("consumer.partition"); List<String> seedStr = (List<String>) paramMap.get("consumer.seedBrokers"); for(String s:seedStr){ seeds.add(s); } port = (int) paramMap.get("consumer.port"); topic = (String) paramMap.get("consumer.topic"); System.out.print("maxReads=" + maxReads + "partition=" + partition + "seedStr=" + seedStr + "port=" + port + "topic=" + topic); example.run(maxReads, topic, partition, seeds, port); } catch (Exception e) { System.out.println("Oops:" + e); e.printStackTrace(); } } private List<String> m_replicaBrokers = new ArrayList<>(); public KafkaSimpleConsumer() { m_replicaBrokers = new ArrayList<>(); } /** * * @param a_maxReads Maximum number of messages to read (so we don’t loop forever) 最大读取消息数量 * @param a_topic Topic to read from 订阅的topic * @param a_partition Partition to read from 查找的分区 * @param a_seedBrokers One broker to use for Metadata lookup broker节点 * @param a_port Port the brokers listen on 端口 * @throws Exception */ public void run(long a_maxReads, String a_topic, int a_partition, List<String> a_seedBrokers, int a_port) throws Exception { // find the meta data about the topic and partition we are interested in // 获取指定topic partition的元数据 PartitionMetadata metadata = findLeader(a_seedBrokers, a_port, a_topic, a_partition); if (metadata == null) { System.out.println("Can‘t find metadata for Topic and Partition. Exiting"); return; } if (metadata.leader() == null) { System.out.println("Can‘t find Leader for Topic and Partition. Exiting"); return; } String leadBroker = metadata.leader().host(); String clientName = "Client_" + a_topic + "_" + a_partition; SimpleConsumer consumer = new SimpleConsumer(leadBroker, a_port, 100000, 64 * 1024, clientName); long readOffset = getLastOffset(consumer, a_topic, a_partition, kafka.api.OffsetRequest.LatestTime(), clientName); System.out.print("readOffset="+readOffset+kafka.api.OffsetRequest.LatestTime()); int numErrors = 0; while (a_maxReads > 0) { if (consumer == null) { consumer = new SimpleConsumer(leadBroker, a_port, 100000, 64 * 1024, clientName); } // Note: this fetchSize of 100000 might need to be increased if // large batches are written to Kafka FetchRequest req = new FetchRequestBuilder().clientId(clientName) .addFetch(a_topic, a_partition, readOffset, 100000).build(); FetchResponse fetchResponse = consumer.fetch(req); if (fetchResponse.hasError()) { System.out.println("error"); numErrors++; // Something went wrong! short code = fetchResponse.errorCode(a_topic, a_partition); System.out.println("Error fetching data from the Broker:" + leadBroker + " Reason: " + code); if (numErrors > 5) break; // 处理offset非法的问题,用最新的offset if (code == ErrorMapping.OffsetOutOfRangeCode()) { // We asked for an invalid offset. For simple case ask for // the last element to reset readOffset = getLastOffset(consumer, a_topic, a_partition, kafka.api.OffsetRequest.LatestTime(), clientName); continue; } consumer.close(); consumer = null; // 更新leader broker leadBroker = findNewLeader(leadBroker, a_topic, a_partition, a_port); continue; } numErrors = 0; long numRead = 0; for (MessageAndOffset messageAndOffset : fetchResponse.messageSet( a_topic, a_partition)) { long currentOffset = messageAndOffset.offset(); System.out.println("currentOffset="+currentOffset+"readOffset="+readOffset); // 必要判断,因为对于compressed message,会返回整个block,所以可能包含old的message if (currentOffset < readOffset) { System.out.println("Found an old offset: " + currentOffset + " Expecting: " + readOffset); continue; } // 获取下一个readOffset 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++; a_maxReads--; } if (numRead == 0) { try { Thread.sleep(1000); } catch (InterruptedException ie) { ie.printStackTrace(); } } } if (consumer != null) consumer.close(); } 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<>(); //build offset fetch request info requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1)); kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest( requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName); //取到offsets 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; } //取到的一组offset long[] offsets = response.offsets(topic, partition); //取第一个开始读 return offsets[0]; } /** * @Descrition :Finding the Lead Broker for a Topic and Partition 从活跃的Broker列表中找出指定Topic、Partition中的Leader Broker *思路就是,遍历每个broker,取出该topic的metadata,然后再遍历其中的每个partition metadata, * 如果找到我们要找的partition就返回根据返回的PartitionMetadata.leader().host()找到leader broker * @param a_oldLeader * @param a_topic * @param a_partition * @param a_port * @return * @throws Exception */ private String findNewLeader(String a_oldLeader, String a_topic, int a_partition, int a_port) throws Exception { for (int i = 0; i < 3; i++) { boolean goToSleep; PartitionMetadata metadata = findLeader(m_replicaBrokers, a_port, a_topic, a_partition); if (metadata == null) { goToSleep = true; } else if (metadata.leader() == null) { goToSleep = true; } else if (a_oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) { // first time through if the leader hasn‘t changed give // ZooKeeper a second to recover // second time, assume the broker did recover before failover, // or it was a non-Broker issue // goToSleep = true; } else { return metadata.leader().host(); } if (goToSleep) { try { Thread.sleep(1000); } catch (InterruptedException ie) { ie.printStackTrace(); } } } System.out.println("Unable to find new leader after Broker failure. Exiting"); throw new Exception("Unable to find new leader after Broker failure. Exiting"); } private PartitionMetadata findLeader(List<String> a_seedBrokers, int a_port, String a_topic, int a_partition) { PartitionMetadata returnMetaData = null; loop: //遍历每个broker for (String seed : a_seedBrokers) { //遍历每个broker SimpleConsumer consumer = null; try { //创建Simple Consumer, consumer = new SimpleConsumer(seed, a_port, 100000, 64 * 1024, "leaderLookup"); List<String> topics = Collections.singletonList(a_topic); TopicMetadataRequest req = new TopicMetadataRequest(topics); //发送TopicMetadata Request请求 TopicMetadataResponse resp = consumer.send(req); //取到Topic的Metadata List<TopicMetadata> metaData = resp.topicsMetadata(); for (TopicMetadata item : metaData) { //遍历每个partition的metadata for (PartitionMetadata part : item.partitionsMetadata()) { //确认是否是我们要找的partition if (part.partitionId() == a_partition) { returnMetaData = part; break loop; //找到就返回 } } } } catch (Exception e) { System.out.println("Error communicating with Broker [" + seed + "] to find Leader for [" + a_topic + ", " + a_partition + "] Reason: " + e); } finally { if (consumer != null) consumer.close(); } } if (returnMetaData != null) { m_replicaBrokers.clear(); for (Broker replica : returnMetaData.replicas()) { m_replicaBrokers.add(replica.host()); } } return returnMetaData; } }
It‘s all。
以上是关于Kafka Consumer应用与高级应用的主要内容,如果未能解决你的问题,请参考以下文章
KafkaMetrics - 无法绑定仪表:kafka.consumer*
kafkaThe group member needs to have a valid member id before actually entering a consumer group(代码片段