kafaka组件、配置、consumer和producer实例

快来打我* 2022-04-23 16:00 559阅读 0赞

## 一．    Kafka组件介绍    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70][] ##

如上图所示，一个典型的Kafka体系架构包括若干Producer（可以是服务器日志，业务数据，页面前端产生的page view等等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer (Group)，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在consumer group发生变化时进行rebalance。Producer使用push(推)模式将消息发布到broker，Consumer使用pull(拉)模式从broker订阅并消费消息。  
Producer：     消息和数据的生产者，向Kafka的一个topic发布消息。  
Consumer：     消息和数据的消费者，订阅topic并处理其发布的消息。  
ConsumerGroup：  每个Consumer属于一个特定的Consumer Group，一条消息可以发送到多个不同的Consumer Group，但是一个Consumer Group中只能有一个Consumer能够消费该消息  
Broker：         消息的代理，Kafka集群中的一个kafka服务节点称为一个broker，主要存储消息数据。存在硬盘中。  
zookeeper：      协调kafka的正常运行。  
Topic：          主题，Kafka处理的消息的不同分类(逻辑概念)。  
Partition：      Topic物理上的分组，一个topic在broker中被分为1个或者多个partition，分区在创建topic的时候指定。每个topic都是有分区的。  
Message：        消息，是通信的基本单位，每个消息都属于一个partition

## 分布式机制 ##

如下图所示，Kafka集群中有4个broker, 某topic有3个partition,且复制因子即副本个数也为3：![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 1][]

Kafka分布式保证的第一个特性就是：Kafka的Replication  
Kafka分布式保证的第二个特性就是：Kafka Leader Election

## 二．    Kafka安装集群配置文件的修改 ##

修改 kafka   $KAFKA\_HOME/config/server.properties 配置文件  
broker.id=0       //broker标识  
log.dirs=/data/kafka\_logs    //kafka具体数据的目录  
zookeeper.connect=hadoop100:2181,hadoop101:2181,hadoop102:2181   //zookeeper的节点信息  
port =9092          //broker server服务端口      
num.partitions=2    //topic在当前broker上的分片个数    
host.name=kafka01   //\#此处为本机IP，（重要）    
log.flush.interval.messages=10000 一个分区的消息数阀值  
log.flush.interval.ms=1000  
1：为了减少磁盘写入的次数,broker会将消息暂时buffer起来,当消息的个  
数达到一定阀值或者过了一定的时间间隔时,再flush到磁盘,这样减少了磁盘  
IO调用的次数。  
配置：Log Flush Policy

log.retention.hours=168  
log.retention.bytes=1073741824  
log.retention.check.interval.ms=300000  
kafka的消息保存一定时间（通常为7天）后会被删除。  
配置：Log Retention Policy  
新增：  
message.max.byte=5242880    //消息的最大值  
default.replication.factor = 3     //消息的副本数      
replica.fetch.max.bytes=5242880

##    三、Producer 配置及代码实例 ##

Kafka集群中由producer负责数据的产生，并发送到对应的Topic；Producer通过push的方式将数据发送到对应Topic的分区  
在发送一条消息时，可以指定这个消息的key，producer根据这个key和partition机制来判断这个消息发送到哪个partition。partition机制可以通过指定producer的partition.class这一参数来指定，该class必须实现kafka.producer.Partitioner接口。

Producer发送数据的方式分为sync(同步)和async(异步)两种，默认为同步方式，由参数\{producer.type\}决定；  
当为异步发送模式的时候Producer提供重试机制，默认失败重试发送3次

配置文件： producer.properties  
\#指定节点列表:指定当前producer需要关注的broker列表；  
metadata.broker.list=kafka01:9092,kafka02:9092,kafka03:9092 ，  
\#指定序列化处理类，将消息实体转化为byte\[\]  
serializer.class=kafka.serializer.DefaultEncoder   
\#指定分区处理类：Producer也根据用户设置的算法来根据消息的key来计算输入哪个partition\[默认是随机发送到不同分区\]，可以自定义  
partitioner.class = kafka.producer.DefaultPartioner

Request.required.ack=1  
当producer.type为同步的时候，会等待接收方确认  
acks默认为1，表示需要leader已成功收到数据并得到确认  
acks：2，表示需要所有副本节点已成功收到数据并得到确认  
acks：0，表示不需要任何节点回复  
kafka producer 端 发送数据分为 同步（实时）和异步（达到设定发送条件）

producer.type = sync  
\#sync同步（默认），async异步可以提高发送吞吐量   
sync，单条发送   
async，buffer一堆请求后，再一起发送

\#重试次数  
message.send.max.retries = 3  
request.timeout.ms \#请求超时时间。默认值：10000

以batch的方式推送数据可以极大的提高处理效率，kafka producer可以将消息在内存中累计到一定数量后作为一个batch发送请求。batch的数量大小可以通过producer的参数（batch.num.messages）控制。通过增加batch的大小，可以减少网络请求和磁盘IO的次数，当然具体参数设置需要在效率和时效性方面做一个权衡  
batch.num.messages=500    
默认值：200。启用异步模式时，一个batch缓存的消息数量。达到这个数值时，producer才会发送消息。（每次批量发送的数量）  
queue.buffering.max.ms=5000   
默认值：5000。启用异步模式时，producer缓存消息的时间。比如我们设置成1000时，它会缓存1s的数据再一次发送出去，这样可以极大的增加broker吞吐量，但也会造成时效性的降低。   
queue.buffering.max.messages=20000    
默认值：10000。启用异步模式时，producer缓存队列里最大缓存的消息数量，如果超过这个值，producer就会阻塞或者丢掉消息。  
queue.enqueue.timeout.ms  
默认值：-1。当达到上面参数时producer会阻塞等待的时间。如果设置为0，buffer队列满时producer不会阻塞，消息直接被丢掉；若设置为-1，producer会被阻塞，不会丢消息。  
compression.codec=0   
\#消息是否压缩,0代表不压缩,1代表用gzip压缩,2代表用snappy压缩  
compressed.topics=    
\#如果要压缩消息，这里指定哪些topic要压缩消息，默认是empty，表示不压缩

Kafka的Producer API主要提供下列三个方法：  
　　public void send(KeyedMessage<K,V> message) 发送单条数据到Kafka集群  
　　public void send(List<KeyedMessage<K,V>> messages) 发送多条数据(数据集)到Kafka集群  
public void close() 关闭Kafka连接资源  
官网示例  
https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example

1.创建配置文件  
Properties props = new Properties();  
props.put("metadata.broker.list", brokerList);  
...  
2. 构建Kafka Producer Configuration上下文  
ProducerConfig config = new ProducerConfig(props);  
3. 构建Producer对象  
Producer<String, String> producer = new Producer<String, String>(config);  
4. 发送数据到服务器，并发线程发送  
  ExecutorService pool = Executors.newFixedThreadPool(numThreads);  
   pool.submit(new Thread(new Runnable() \{  
               @Override  
               public void run() \{  
                ...  
                        // 发送数据  
               KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, msg);  
               producer.send(message);  
                 
                \}  
            \}));

##    四.Consumer配置及代码实例     ##

在官方网站上，给出了这样一张图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 2][]  
consumer group是kafka提供的可扩展且具有容错性的消费者机制。既然是一个组，那么组内必然可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。当然，每个分区只能由同一个消费组内的一个consumer来消费。

一个kafka cluster中的某个topic，有4个partition。有两个consumer group (A and B)订阅了该topic。 Consumer Group A有2个consumer：c1、c2，Consumer Group B有4个consumer：c3,c4,c5,c6。经过分区分配后，consumer与partition的订阅关系如下：  
Topic 中的4个partition在Consumer Group A中的分配情况如下：  
C1 订阅p0,p3  
C2 订阅p1,p2  
Topic 中的4个partition在Consumer Group B中的分配情况如下：  
C3 订阅p0  
C4 订阅p3  
C5 订阅p1  
C6 订阅p2  
High Level Consumer API：高度抽象的Kafka消费者API；将底层具体获取数据、更新offset、设置偏移量等操作屏蔽掉，直接将操作数据流的处理工作提供给编写程序的人员。优点是：操作简单；缺点：可操作性太差，无法按照自己的业务场景选择处理方式。(入口类：ConsumerConnector)  
Lower Level Consumer API：通过直接操作底层API获取数据的方式获取Kafka中的数据，需要自行给定分区、偏移量等属性。优点：可操作性强；缺点：代码相对而言比较复杂。(入口类：SimpleConsumer)

## High Level Consumer ##

很多时候，客户程序只是希望从Kafka读取数据，不太关心消息offset的处理。同时也希望提供一些语义，例如同一条消息只被某一个Consumer消费（单播）或被所有Consumer消费（广播）。因此，Kafka Hight Level Consumer提供了一个从Kafka消费数据的高层抽象，从而屏蔽掉其中的细节并提供丰富的语义。

## Consumer Group    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 3][] ##

很多传统的Message Queue都会在消息被消费完后将消息删除，一方面避免重复消费，另一方面可以保证Queue的长度比较短，提高效率。  
Kafka并不删除已消费的消息，为了实现传统Message Queue消息只被消费一次的语义，Kafka保证每条消息在同一个Consumer Group里只会被某一个Consumer消费。与传统Message Queue不同的是，  
Kafka还允许不同Consumer Group同时消费同一条消息，这一特性可以为消息的多元化处理提供支持。  
示例1：  
创建一个Topic (名为topic1)，再创建一个属于group1的Consumer实例，并创建三个属于group2的Consumer实例，然后通过Producer向topic1发送Key分别为1，2，3的消息。结果发现属于group1的Consumer收到了所有的这三条消息，同时group2中的3个Consumer分别收到了Key为1，2，3的消息，如下图所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 4][]

## Rebalance ##

Consumer会自动发起reblance  
如下例所示，如果topic1有0，1，2共三个Partition，当group1只有一个Consumer(名为consumer1)时，该 Consumer可消费这3个Partition的所有数据。  
　　 增加一个Consumer(consumer2)后，其中一个Consumer（consumer1）可消费2个Partition的数据（Partition 0和Partition 1），另外一个Consumer(consumer2)可消费另外一个Partition（Partition 2）的数据。  
　　再增加一个Consumer(consumer3)后，每个Consumer可消费一个Partition的数据。consumer1消费partition0，consumer2消费partition1，consumer3消费partition2。  
　　 再增加一个Consumer（consumer4）后，其中3个Consumer可分别消费一个Partition的数据，另外一个Consumer（consumer4）不能消费topic1的任何数据。  
　　此时关闭consumer1，其余3个Consumer可分别消费一个Partition的数据。  
　　接着关闭consumer2，consumer3可消费2个Partition，consumer4可消费1个Partition。  
　　再关闭consumer3，仅存的consumer4可同时消费topic1的3个Partition。

配置文件： Consumer.properties  
group.id=test-group   
(必需)consumer group id

zookeeper.connect=zk01:2181,zk02:2181,zk03:2181    
(必需)zookeeper连接服务器地址

zookeeper.session.timeout.ms=5000    
zookeeper的session的过期时间    
zookeeper.connection.timeout.ms 默认值：6000  
client连接到ZK server的超时时间。

zookeeper.sync.time.ms = 200    
ZooKeeper集群中leader和follower之间的同步

fetch.message.max.bytes 默认值：1024 \* 1024  
每一个获取某个topic的某个partition的请求，得到最大的字节数，每一个partition的要被读取的数据会加载入内存，所以这可以帮助控制consumer使用的内存。这个值的设置不能小于在server端设置的最大消息的字节数，否则producer可能会发送大于consumer可以获取的字节数限制的消息。

rebalance.max.retries 默认值：4  
当一个新的consumer加入一个consumer group时，会有一个rebalance的操作，导致每一个consumer和partition的关系重新分配。如果这个重分配失败的话，会进行重试，此配置就代表最大的重试次数。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 5][]

rebalance.backoff.ms 默认值：2000  
在rebalance重试时的backoff时间。

topic = PMCS\_RESULT  
指定topic

consumer.id=xxx  
当前consumer的标识 ，如果不给定的话，默认自动产生一个随机的id

auto.commit.enable 默认值：true  
如果设为true，consumer会定时向ZooKeeper发送已经获取到的消息的offset。当consumer进程挂掉时，已经提交的offset可以继续使用，让新的consumer继续工作。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 6][]

auto.commit.interval.ms 默认值：60 \* 1000  
consumer向ZooKeeper发送offset的时间间隔。

auto.offset.reset 默认值：largest  
在Consumer在ZooKeeper中发现没有初始的offset时或者发现offset不在范围呢，该怎么做：  
\* smallest : 自动把offset设为最小的offset。  
\* largest : 自动把offset设为最大的offset。  
\* anything else: 抛出异常。

官网示例  
https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example  
1.创建配置文件  
// 1. 构建属性对象  
    Properties prop = new Properties();  
        // 2. 添加相关属性  
prop.put("group.id", groupId); // 指定分组id  
ConsumerConfig consumerConfig = new ConsumerConfig(props);  
2. Kafka数据消费对象  
    private ConsumerConnector consumer = Consumer.createJavaConsumerConnector(consumerConfig  
);  
3. 获取对应topic的数据流数据  
指定Topic  
    Map<String, Integer> map = new HashMap<String, Integer>();  
    topicCountMap.put(topic, numThreads);  
  指定数据的解码器  
    StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());  
    StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());  
从返回结果中获取对应topic的数据流处理器  
Map<String, List<KafkaStream<String, String>>> topicMessageStreams = consumerConnector.createMessageStreams(map, keyDecoder, valueDecoder);  
         
4.构建数据输出对象  
 for(List<KafkaStream<String, String>> streams :topicMessageStreams.values())\{  
      for (final KafkaStream<String, String> stream : streams) \{  
      executor.submit(new Runnable() \{  
      public void run() \{  
            // 1. 获取数据迭代器  
           ConsumerIterator<String, String> it = stream.iterator();  
// 2. 迭代输出数据  
           while (it.hasNext()) \{  
// 2.1 获取数据值  
           MessageAndMetadata<String, String> data = it.next();  
           String receiveData = data.message();  
                      \}  
                   \}  
               \});  
            \}  
        \}

借鉴：  
https://blog.csdn.net/u013256816/article/details/71091774  
http://frankfan915.iteye.com/blog/2219715  
https://blog.csdn.net/yajunandchunyun/article/details/55509220  
http://www.cnblogs.com/liuming1992/p/6432626.html  
http://www.infoq.com/cn/articles/kafka-analysis-part-1  
配置：  
http://damacheng009.iteye.com/blog/2088020  
概念：  
http://www.cnblogs.com/huxi2b/p/6223228.html

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70]: /images/20220224/1c68d71061c34cc9b55e79c1e5c0c75b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 1]: /images/20220224/c3772f9774cc4ef1bdb2a2a2fe91892c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 2]: /images/20220224/a2215aca380f4bd6bd4161213ba704ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 3]: /images/20220224/e94f2a5758994d45af78bfbaae195834.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 4]: /images/20220224/06f96f68fcfc4d6d89dd43ea86653f76.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 5]: /images/20220224/92feb34171644c0db039eec2c910ffe5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 6]: /images/20220224/334319b8b15a4913ba27a61f3e84fed4.png