当前位置 : 首页 » 文章分类 :  开发  »  Apache-kafka

Apache-kafka

kafka笔记

https://kafka.apache.org/


KafkaEmbedded kafka单元测试

Spring Kafka - Embedded Unit Test Example
https://www.codenotfound.com/spring-kafka-embedded-unit-test-example.html

code-not-found/spring-kafka
https://github.com/code-not-found/spring-kafka

Simple embedded Kafka test example with spring boot
https://stackoverflow.com/questions/48753051/simple-embedded-kafka-test-example-with-spring-boot


System property ‘java.security.auth.login.config’ is not set

原因是找不到 kafka 账号配置文件,可能原因是配置文件是通过启动脚本的 jvm参数传入的,但本地启动spring boot时忘了指定这个参数,就可能出此问题。
解决方法是在idea中配置Run Configuration,把kafka配置文件路径加入 jvm 参数中。
VM options: -Djava.security.auth.login.config=/Users/user/IdeaProjects/uds/uds-message-consumer/conf/test/kafka.conf

Caused by: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set
    at org.apache.kafka.common.security.JaasContext.defaultContext(JaasContext.java:131)
    at org.apache.kafka.common.security.JaasContext.load(JaasContext.java:96)
    at org.apache.kafka.common.security.JaasContext.load(JaasContext.java:78)
    at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:100)
    at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:58)
    at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:88)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:374)

Kafka “Login module not specified in JAAS config”
https://stackoverflow.com/questions/45756543/kafka-login-module-not-specified-in-jaas-config


基本概念

Kafka是一个分布式流数据系统,使用Zookeeper进行集群的管理。与其他消息系统类似,整个系统由生产者、Broker Server和消费者三部分组成,生产者和消费者由开发人员编写,通过API连接到Broker Server进行数据操作。我们重点关注三个概念:

Topic

Topic,是Kafka下消息的类别,类似于RabbitMQ中的Exchange的概念。这是逻辑上的概念,用来区分、隔离不同的消息数据,屏蔽了底层复杂的存储方式。对于大多数人来说,在开发的时候只需要关注数据写入到了哪个topic、从哪个topic取出数据。

Partition

Partition,是Kafka下数据存储的基本单元,这个是物理上的概念。同一个topic的数据,会被分散的存储到多个partition中,这些partition可以在同一台机器上,也可以是在多台机器上,比如下图所示的topic就有4个partition,分散在两台机器上。这种方式在大多数分布式存储中都可以见到,比如MongoDB、Elasticsearch的分片技术,其优势在于:有利于水平扩展,避免单台机器在磁盘空间和性能上的限制,同时可以通过复制来增加数据冗余性,提高容灾能力。为了做到均匀分布,通常partition的数量通常是Broker Server数量的整数倍。

Consumer Group

Consumer Group,同样是逻辑上的概念,是Kafka实现单播和广播两种消息模型的手段。同一个topic的数据,会广播给不同的group;同一个group中的worker,只有一个worker能拿到这个数据。换句话说,对于同一个topic,每个group都可以拿到同样的所有数据,但是数据进入group后只能被其中的一个worker消费。group内的worker可以使用多线程或多进程来实现,也可以将进程分散在多台机器上,worker的数量通常不超过partition的数量,且二者最好保持整数倍关系,因为Kafka在设计时假定了一个partition只能被一个worker消费(同一group内)。

consumer group是kafka提供的可扩展且具有容错性的消费者机制。既然是一个组,那么组内必然可以有多个消费者或消费者实例(consumer instance),它们共享一个公共的ID,即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。当然,每个分区只能由同一个消费组内的一个consumer来消费。理解consumer group记住下面这三个特性就好了:

1)consumer group下可以有一个或多个consumer instance,consumer instance可以是一个进程,也可以是一个线程
2)group.id是一个字符串,唯一标识一个consumer group
3)consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)


kafka partition

kafka partition

Kafka中可以将Topic从物理上划分成一个或多个分区(Partition),每个分区在物理上对应一个文件夹,以”topicName_partitionIndex”的命名方式命名,该文件夹下存储这个分区的所有消息(.log)和索引文件(.index),这使得Kafka的吞吐率可以水平扩展。

生产者在生产数据的时候,可以为每条消息指定Key,这样消息被发送到broker时,会根据分区规则选择被存储到哪一个分区中,如果分区规则设置的合理,那么所有的消息将会被均匀的分布到不同的分区中,这样就实现了负载均衡和水平扩展。另外,在消费者端,同一个消费组可以多线程并发的从多个分区中同时消费数据(后续将介绍这块)。

Consumer分区分配机制

分配partition的策略:

range:对于每个topic,会将topic的partition编上序号排好序,然后consumer线程以字典序排序。然而把partition的总数除以consumer线程的总数来决定分配给每个线程的partition数目。如果无法除尽,将余数再均分给排序靠前的几个线程,即这些线程都会多出额外的一个partition。
round-robin:这个策略把所有partition和所有consumer线程都列出来。然后它以循环制分配partition给线程。如果所有consumer实例的订阅是相同的,那么partition会均匀分布。这个分配策略只有当以下情况成立时才可用:a.每个topic在一个consumer实例中有同样的stream数目。b.在group中的每个consumer实例订阅的topic的集合是相同的。
如果所有consumer实例有相同的consumer group,那么这个就像传统的队列,负载均衡到所有consumer上。假如多个consumer实例都有多个线程,且属于同一个group,那么一个topic的所有partition会均匀分配给所有线程。

接收消息的顺序只能保证一个partition之内是有序的,一个consumer接收多个partition的话是无法保证消息全局有序的,即consumer接收的消息的顺序可能跟producer发送的顺序不同。

Producer分区机制

当指定partition key的时候,分配partition的策略:

hash:由消息所提供的key来进行hash,然后分发到对应的partition。这是默认使用的partition机制。
自定义:自己实现partition接口,并在配置中用参数partitioner.class指定这个实现。
当没有指定partition key的时候,分配partition的策略:

随机:把每个消息随机分发到一个partition中。在10分钟内,该partition不会切换。所以,当producer数目小于partition时,在一定时间内会有部分partition没有收到数据。

Kafka分配分区的机制
http://lsr1991.github.io/2015/07/09/kafka-partition-mechanism/

kafka-manager

kafka-manager
https://github.com/yahoo/kafka-manager

kafka集群管理工具kafka-manager部署安装
https://www.jianshu.com/p/c24ed08dfa63

Apache Kafka 入门 - Kafka-manager的基本配置和运行
https://blog.csdn.net/isea533/article/details/73727485


上一篇 Java8 Optional 笔记

下一篇 Apache-HttpComponents

阅读
1,844
阅读预计7分钟
创建日期 2018-08-27
修改日期 2018-08-27
类别
百度推荐