Kafka学习

时间  2019-11-07
标签 kafka 学习 栏目 Kafka 繁體版
原文   原文链接

其余更多java基础文章：
java基础学习(目录)java

学习资料：kafka数据可靠性深度解读apache

Kafka概述

  • Kafka是一个分布式消息队列。Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群有多个kafka实例组成，每一个实例(server)成为broker。
  • 不管是kafka集群，仍是producer和consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性。
  • 在流式计算中，Kafka通常用来缓存数据，Spark经过消费Kafka的数据进行计算。

Kafka架构

  • Producer ：消息生产者，就是向kafka broker发消息的客户端。
  • Consumer ：消息消费者，向kafka broker取消息的客户端
  • Topic ：能够理解为一个队列。
  • Consumer Group （CG）：这是kafka用来实现一个topic消息的广播（发给全部的consumer）和单播（发给任意一个consumer）的手段。一个topic能够有多个CG。topic的消息会复制（不是真的复制，是概念上的）到全部的CG，但每一个partion只会把消息发给该CG中的一个consumer。若是须要实现广播，只要每一个consumer有一个独立的CG就能够了。要实现单播只要全部的consumer在同一个CG。用CG还能够将consumer进行自由的分组而不须要屡次发送消息到不一样的topic。
  • Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker能够容纳多个topic。
  • Partition：为了实现扩展性，一个很是大的topic能够分布到多个broker（即服务器）上，一个topic能够分为多个partition，每一个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的总体（多个partition间）的顺序。

分区

消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成，其组织结构以下图所示：bootstrap

咱们能够看到，每一个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每个消息都被赋予了一个惟一的offset值。缓存

  • 分区的缘由
    1. 方便在集群中扩展，每一个Partition能够经过调整以适应它所在的机器，而一个topic又能够有多个Partition组成，所以整个集群就能够适应任意大小的数据了；
    2. 能够提升并发，由于能够以Partition为单位读写了。
  • 分区的原则
    1. 指定了patition，则直接使用；
    2. 未指定patition但指定key，经过对key的value进行hash出一个patition
    3. patition和key都未指定，使用轮询选出一个patition。

副本（Replication）

同一个partition可能会有多个replication（对应 server.properties 配置中的 default.replication.factor=N）。没有replication的状况下，一旦broker 宕机，其上全部 patition 的数据都不可被消费，同时producer也不能再将数据存于其上的patition。引入replication以后，同一个partition可能会有多个replication，而这时须要在这些replication之间选出一个leader，producer和consumer只与这个leader交互，其它replication做为follower从leader 中复制数据。bash

写入流程

  1. producer先从zookeeper的 "/brokers/.../state"节点找到该partition的leader
  2. producer将消息发送给该leader
  3. leader将消息写入本地log
  4. followers从leader pull消息，写入本地log后向leader发送ACK
  5. leader收到全部ISR中的replication的ACK后，增长HW（high watermark，最后commit 的offset）并向producer发送ACK

ACK，HW，ISR等能够阅读kafka数据可靠性深度解读学习
简单来讲：服务器

  • HW是HighWatermark的缩写，是指consumer可以看到的此partition的位置
  • ISR (In-Sync Replicas)，这个是指副本同步队列。全部的副本（replicas）统称为Assigned Replicas，即AR。ISR是AR中的一个子集，由leader维护ISR列表，follower从leader同步数据有一些延迟（包括延迟时间replica.lag.time.max.ms和延迟条数replica.lag.max.messages两个维度），任意一个超过阈值都会把follower剔除出ISR, 存入OSR（Outof-Sync Replicas）列表，新加入的follower也会先存放在OSR中。AR=ISR+OSR。
  • ack机制：在kafka发送数据的时候，每次发送消息都会有一个确认反馈机制，确保消息正常的可以被收到。

Kafka消费过程分析

kafka提供了两套consumer API：高级Consumer API和低级API。架构

高级API

  • 高级API优势
    • 高级API 写起来简单
    • 不须要去自行去管理offset，系统经过zookeeper自行管理
    • 不须要管理分区，副本等状况，系统自动管理
    • 消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据（默认设置1分钟更新一下zookeeper中存的的offset）
    • 可使用group来区分对同一个topic的不一样程序访问分离开来（不一样的group记录不一样的offset，这样不一样程序读取同一个topic才不会由于offset互相影响）
  • 高级API缺点
    • 不能自行控制offset（对于某些特殊需求来讲）
    • 不能细化控制如分区、副本、zk等

低级API

  • 低级 API 优势
    • 可以开发者本身控制offset，想从哪里读取就从哪里读取。
    • 自行控制链接分区，对分区自定义进行负载均衡
    • 对zookeeper的依赖性下降（如：offset不必定非要靠zk存储，自行存储offset便可，好比存在文件或者内存中）
  • 低级API缺点
    • 太过复杂，须要自行控制offset，链接哪一个分区，找到分区leader 等。

消费者组

消费者是以consumer group消费者组的方式工做，由一个或者多个消费者组成一个组，共同消费一个topic。每一个分区在同一时间只能由group中的一个消费者读取，可是多个group能够同时消费这个partition。并发

在图中，有一个由三个消费者组成的group，有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也能够叫作某个消费者是某个分区的拥有者。 在这种状况下，消费者能够经过水平扩展的方式同时读取大量的消息。另外，若是一个消费者失败了，那么其余的group成员会自动负载均衡读取以前失败的消费者读取的分区。

JAVA中使用Kafka

生产者

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaProducerDemo {

    public static void main(String[] args){
        //test();
        test2();
    }

    public static void test(){
        Properties props= new Properties();
        props.put("bootstrap.servers", "172.26.40.181:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer producer = new KafkaProducer(props);
        for(int i = 0; i < 10; i++){
            producer.send(new ProducerRecord("first",Integer.toString(i), Integer.toString(i)));
        }

        producer.close();
    }

    /**
     * 带回调函数
     */
    public static void test2(){
        Properties props = new Properties();
        // Kafka服务端的主机名和端口号
        props.put("bootstrap.servers", "172.26.40.181:9092");
        // 等待全部副本节点的应答
        props.put("acks", "all");
        // 消息发送最大尝试次数
        props.put("retries", 0);
        // 一批消息处理大小
        props.put("batch.size", 16384);
        // 增长服务端请求延时
        props.put("linger.ms", 1);
// 发送缓存区内存大小
        props.put("buffer.memory", 33554432);
        // key序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //拦截器
        List<String> interceptor = new ArrayList<>();
        interceptor.add("com.hiway.practice.kafka.TimeInterceptor");
        interceptor.add("com.hiway.practice.kafka.CounterInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptor);

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        for (int i = 0; i < 50; i++) {

            kafkaProducer.send(new ProducerRecord<String, String>("first", "hello" + i), new Callback() {

                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {

                    if (metadata != null) {

                        System.err.println(metadata.partition() + "---" + metadata.offset());
                    }
                }
            });
        }

        kafkaProducer.close();
    }

}
复制代码

消费者

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class CustomNewConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        // 定义kakfa 服务的地址，不须要将全部broker指定上
        props.put("bootstrap.servers", "172.26.40.181:9092");
        // 制定consumer group
        props.put("group.id", "test");
        // 是否自动确认offset
        props.put("enable.auto.commit", "true");
        // 自动确认offset的时间间隔
        props.put("auto.commit.interval.ms", "1000");
        // key的序列化类
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // value的序列化类
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 定义consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 消费者订阅的topic, 可同时订阅多个
        consumer.subscribe(Arrays.asList("first"));

        while (true) {
            // 读取数据，读取超时时间为100ms
            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}
复制代码

拦截器

public class TimeInterceptor implements ProducerInterceptor<String,String> {


    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return new ProducerRecord<>(record.topic(),record.partition(),record.timestamp(),record.key(),System.currentTimeMillis() + "," + record.value().toString());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}

public class CounterInterceptor implements ProducerInterceptor<String, String> {
    private int errorCounter = 0;
    private int successCounter = 0;

    @Override
    public void configure(Map<String, ?> configs) {

    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // 统计成功和失败的次数
        if (exception == null) {
            successCounter++;
        } else {
            errorCounter++;
        }
    }

    @Override
    public void close() {
        // 保存结果
        System.out.println("Successful sent: " + successCounter);
        System.out.println("Failed sent: " + errorCounter);
    }
}
复制代码

错误总结

1. Uncaught error in kafka producer I/O thread错误
这个问题主要是服务器上的kafka版本和IDEA中的kafka版本不一致致使的。负载均衡

2.producer发送数据到集群上无反应分布式

将kafka/config/server.properties文件中advertised.listeners改成以下属性。172.26.40.181是我虚拟机的IP。改完后重启，OK了。Java端的代码终于能通讯了 advertised.listeners=PLAINTEXT://172.26.40.181:9092 advertised.listeners上的注释是这样的：

#Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
复制代码

意思就是说：hostname、port都会广播给producer、consumer。若是你没有配置了这个属性的话，则使用listeners的值，若是listeners的值也没有配置的话，则使用 java.net.InetAddress.getCanonicalHostName()返回值(这里也就是返回localhost了)。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程