Kafka 消费者旧版低级 API

Kafka 消费者总共有 3 种 API,新版 API、旧版高级 API、旧版低级 API,新版 API 是在 kafka 0.9 版本后增加的,推荐使用新版 API,但由于旧版低级 API 可以对消息进行更加灵活的控制,所有在实际开发中使用的也较多,本文讨论消费者旧版低级 API 的基本使用。

旧版低级 API 处理以下场景更为方便:

  • 消息重复消费
  • 添加事务管理机制,保证 Exactly Once
  • 消费指定分区或者指定分区的某些片段

使用旧版低级 API的步骤:

  • 获取你要读取的topic的partition的元数据信息
  • 找到这个partition的leader节点,然后通过这个leader节点找到存有这个partition副本的节点
  • 构造消费请求,获取数据并处理
  • 手动管理偏移量
  • 识别并处理分区leader节点的改变

以下示例代码实现的功能是,指定主题和分区,从该分区的第一条记录开始读取数据,打印到控制台:

package com.bonc.rdpe.kafka110.consumer;

import java.nio.ByteBuffer;
import java.util.Arrays;
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.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.TopicMetadataResponse;
import kafka.javaapi.OffsetRequest;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;

/**
 * @Title SimpleConsumerTest.java 
 * @Description Kafka旧版消费者API——低级API
 * @Author YangYunhe
 * @Date 2018-06-26 13:16:29
 */
public class SimpleConsumerTest {
    
    // broker list
    private static final String BROKER_LIST = "192.168.42.101,192.168.42.102,192.168.42.103";
    // 连接超时时间:1min
    private static final int TIME_OUT = 60 * 1000;
    // 读取消息缓存区大小:1M
    private static final int BUFFER_SIZE = 1024 * 1024;
    // 每次获取消息的条数
    private static final int FETCH_SIZE = 100000;
    // 发生错误时重试的次数
    private static final int RETRIES_TIME = 3;
    // 允许发生错误的最大次数
    private static final int MAX_ERROR_NUM = 3;
    
    /**
     * 获取指定主题指定分区的元数据
     */
    private PartitionMetadata fetchPartitionMetadata(List<String> brokerList, int port, String topic, int partitionId) {
        SimpleConsumer consumer = null;
        TopicMetadataRequest metadataRequest = null;
        TopicMetadataResponse metadataResponse = null;
        List<TopicMetadata> topicMetadatas = null;
        
        try{
            /*
             * 循环是因为不确定传入的partition的leader节点是哪个
             */
            for(String host : brokerList) {
                // 1. 构建一个消费者,它是获取元数据的执行者
                consumer = new SimpleConsumer(host, port, TIME_OUT, BUFFER_SIZE, "dev3-client-001");  // 最后一个参数是 clientId
                // 2. 构造请求主题元数据的 request
                metadataRequest = new TopicMetadataRequest(Arrays.asList(topic));
                // 3. 发送请求获取元数据
                try {
                    metadataResponse = consumer.send(metadataRequest);
                } catch (Exception e) {
                    System.out.println("get TopicMetadataResponse failed!");
                    e.printStackTrace();
                    continue;
                }
                // 4. 获取主题元数据列表
                topicMetadatas = metadataResponse.topicsMetadata();
                // 5. 提取主题元数据列表中指定分区的元数据信息
                for(TopicMetadata topicMetadata : topicMetadatas) {
                    for(PartitionMetadata partitionMetadata : topicMetadata.partitionsMetadata()) {
                        if(partitionMetadata.partitionId() != partitionId) {
                            continue;
                        } else {
                            return partitionMetadata;
                        }
                    }
                }
            }
        } catch (Exception e) {
            System.out.println("Fetch PartitionMetadata failed!");
            e.printStackTrace();
        } finally {
            if(consumer != null) {
                consumer.close();
            }
        }
        
        return null;
        
    }
    
    /**
     * 根据分区的元数据信息获取它的leader节点
     */
    private String getLeader(PartitionMetadata metadata) {
        if(metadata.leader() == null) {
            System.out.println("can not find partition" + metadata.partitionId() + "'s leader!");
            return null;
        }
        return metadata.leader().host();
    }
    
    /**
     * 重新寻找partition的leader节点的方法
     */
    private String findNewLeader(List<String> brokerList, String oldLeader, String topic, int partition, int port) throws Exception {
        for (int i = 0; i < 3; i++) {
            boolean goToSleep = false;
            PartitionMetadata metadata = fetchPartitionMetadata(brokerList, port, topic, partition);
            if (metadata == null) {
                goToSleep = true;
            } else if (metadata.leader() == null) {
                goToSleep = true;
            } else if (oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) {
                // 这里考虑到 zookeeper 还没有来得及重新选举 leader 或者在故障转移之前挂掉的 leader 又重新连接的情况
                goToSleep = true;
            } else {
                return metadata.leader().host();
            }
            if (goToSleep) {
                Thread.sleep(1000);
            }
        }
        System.out.println("Unable to find new leader after Broker failure!");
        throw new Exception("Unable to find new leader after Broker failure!");
    }
    
    /**
     * 获取指定主题指定分区的消费偏移量
     */
    private long getOffset(SimpleConsumer consumer, String topic, int partition, long beginTime, String clientName) {
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
        Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<>();
        /*
         * PartitionOffsetRequestInfo(beginTime, 1)用于配置获取offset的策略
         * beginTime有两个值可以取
         *     kafka.api.OffsetRequest.EarliestTime(),获取最开始的消费偏移量,不一定是0,因为segment会删除
         *     kafka.api.OffsetRequest.LatestTime(),获取最新的消费偏移量
         * 另一个参数 1 暂不清楚有什么意义
         */
        requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(beginTime, 1));
        // 构造获取offset的请求
        OffsetRequest request = new OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
        OffsetResponse response = consumer.getOffsetsBefore(request);
        if(response.hasError()) {
            System.out.println("get offset failed!" + response.errorCode(topic, partition));
            return -1;
        }
        long[] offsets = response.offsets(topic, partition);
        if(offsets == null || offsets.length == 0) {
            System.out.println("get offset failed! offsets is null");
            return -1;
        }
        return offsets[0];
    }
    
    /**
     * 处理数据的方法
     */
    public void consume(List<String> brokerList, int port, String topic, int partition) {
        SimpleConsumer consumer = null;
        try {
            // 1. 获取分区元数据信息
            PartitionMetadata metadata = fetchPartitionMetadata(brokerList, port, topic, partition);
            if(metadata == null) {
                System.out.println("can not find metadata!");
                return;
            }
            // 2. 找到分区的leader节点
            String leaderBroker = getLeader(metadata);
            String clientId = topic + "-" + partition + "-" + "client";
            
            // 3. 创建一个消费者用于消费消息
            consumer = new SimpleConsumer(leaderBroker, port, TIME_OUT, BUFFER_SIZE, clientId);
            
            // 4. 配置获取offset的策略为,获取分区最开始的消费偏移量
            long offset = getOffset(consumer, topic, partition, kafka.api.OffsetRequest.EarliestTime(), clientId);
            
            int errorCount = 0;
            FetchRequest request = null;
            FetchResponse response = null;
            
            while(offset > -1) {
                // 运行过程中,可能因为处理错误,把consumer置为 null,所以这里需要再实例化
                if(consumer == null) {
                    consumer = new SimpleConsumer(leaderBroker, port, TIME_OUT, BUFFER_SIZE, clientId);
                }
                // 5. 构建获取消息的request
                request = new FetchRequestBuilder().clientId(clientId).addFetch(topic, partition, offset, FETCH_SIZE).build();
                // 6. 获取响应并处理
                response = consumer.fetch(request);
                if(response.hasError()) {
                    errorCount ++;
                    if(errorCount > MAX_ERROR_NUM) {
                        break;
                    }
                    short errorCode = response.errorCode(topic, partition);
                    
                    if(ErrorMapping.OffsetOutOfRangeCode() == errorCode) {
                        // 如果是因为获取到的偏移量无效,那么应该重新获取
                        // 这里简单处理,改为获取最新的消费偏移量
                        offset = getOffset(consumer, topic, partition, kafka.api.OffsetRequest.LatestTime(), clientId);
                        continue;
                    } else if (ErrorMapping.OffsetsLoadInProgressCode() == errorCode) {
                        Thread.sleep(300000);
                        continue;
                    } else {
                        consumer.close();
                        consumer = null;
                        // 更新leader broker
                        leaderBroker = findNewLeader(brokerList, leaderBroker, topic, partition, port); 
                        continue;
                    }
                // 如果没有错误
                } else {
                    errorCount = 0;  // 清空错误记录
                    long fetchCount = 0;
                    // 处理消息
                    for(MessageAndOffset messageAndOffset : response.messageSet(topic, partition)) {
                        long currentOffset = messageAndOffset.offset();
                        if(currentOffset < offset) {
                            System.out.println("get an old offset[" + currentOffset + "], excepted offset is offset[" + offset + "]");
                            continue;
                        }
                        offset = messageAndOffset.nextOffset();
                        ByteBuffer payload = messageAndOffset.message().payload();
                        byte[] bytes = new byte[payload.limit()];
                        payload.get(bytes);
                        
                        // 把消息打印到控制台
                        System.out.println("message: " + new String(bytes, "UTF-8") + ", offset: " + messageAndOffset.offset());
                        
                        fetchCount++;
                    }
                    
                    if (fetchCount == 0) {
                        Thread.sleep(1000);
                    }
                    
                }
            }
            
        } catch (Exception e) {
            System.out.println("exception occurs when consume message");
            e.printStackTrace();
        } finally {
            if (consumer != null) {
                consumer.close();
            }
        }
    }
    
    public static void main(String[] args) {
        SimpleConsumerTest sct = new SimpleConsumerTest();
        sct.consume(Arrays.asList(BROKER_LIST.split(",")), 9092, "dev3-yangyunhe-topic001", 0);
    }
    
}

运行结果为:

message: t13229543255|13229543255|2017-10-04 05:04:19|2017-10-04 05:04:19|01001|x13229543255|75165|UaiOKGnr|wx|1003|0, offset: 0
message: t15554236866|15554236866|2017-10-04 20:50:36|2017-10-04 20:50:36|01001|x15554236866|66815|cLhDPEfl|wx|1001|0, offset: 1
message: t13053448010|13053448010|2018-04-01 18:22:27|2018-04-01 18:22:27|01001|x13053448010|67860|UTpzF05R|wx|1005|1, offset: 2
message: t13016064334|13016064334|2017-07-01 15:10:31|2017-07-01 15:10:31|01001|x13016064334|78549|nzbMDJXs|wx|1005|0, offset: 3
message: t13025257802|13025257802|2018-01-21 18:49:55|2018-01-21 18:49:55|01001|x13025257802|39067|zOHLBrje|wx|1003|0, offset: 4

......

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏阿杜的世界

《七周七并发模型》阅读笔记(一)一、线程与锁——第一天二、线程与锁——第二天三、线程与锁——第三天

线程与锁模型其实是对底层硬件运行过程的形式化,这种形式化既是该模型最大的优点,也是它最大的缺点。我们借助Java语言来学习线程与锁模型,不过内容也适用于其他语言...

13820
来自专栏芋道源码1024

面试问烂的 Spring AOP 原理

来源:https://www.jianshu.com/p/e18fd44964eb

24340
来自专栏向前进

【笔记】HybridApp中使用Promise化的JS-Bridge

背景: HybridApp,前端采用JS-bridge的方式调用Native的接口,如获取设备信息、拍照、人脸识别等 前端封装了调用库,每次调用Native接口...

35440
来自专栏偏前端工程师的驿站

asp.net 解码gb2312下urlencode后的字符串

公司网站前期的网页用了gb2312保存用户数据,而我负责的部分用的是utf8,今天恰好要获取前期录入的数据于是毫无悬念地出现乱码问题,经过一番网上的搜索还是找不...

22150
来自专栏从零开始学自动化测试

python笔记9-多线程Threading之阻塞(join)和守护线程(setDaemon)

前言 今天小编YOYO请xiaoming和xiaowang吃火锅,吃完火锅的时候会有以下三种场景: - 场景一:小编(主)先吃完了,xiaoming(客)和xi...

39460
来自专栏Java编程技术

使用数据库悲观锁实现不可重入的分布式锁

在同一个jvm进程中时,可以使用JUC提供的一些锁来解决多个线程竞争同一个共享资源时候的线程安全问题,但是当多个不同机器上的不同jvm进程共同竞争同一个共享资源...

7910
来自专栏java 成神之路

RocketMQ 底层通信机制 源码分析

RocketMQ 底层通讯是使用Netty来实现的。 下面我们通过源码分析下RocketMQ是怎么利用Netty进行通讯的。

17820
来自专栏Android相关

处理器结构--ReorderBuffer

Reorder Buffer用来保存在乱序执行之前的(OOOE)指令执行顺序,当指令集合在乱序执行后按照原有指令顺序将结果提交。

15140
来自专栏分布式系统进阶

Influxdb的Meta data分析

Influxdb定义了一个Service:Precreator Serivec(services/precreator/service.go),实现比较简单,周...

14320
来自专栏JackieZheng

RabbitMQ入门-Routing直连模式

Hello World模式,告诉我们如何一对一发送和接收消息; Work模式,告诉我们如何多管齐下高效的消费消息; Publish/Subscribe模式,告...

288100

扫码关注云+社区

领取腾讯云代金券