Kafka应用入门

Myth丶恋晨 2022-04-01 06:22 156阅读 0赞

**Kafka是一款基于发布和订阅的消息系统**  
运作方式如图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70]  
各个组件的作用:  
生产者:将数据依据主题，分区发往broker  
broker:broker接收来自生产者的消息，为休息设置偏移量，并将消息保存到磁盘  
zookeeper:管理各个分布式broker节点，为各个节点提供数据共享，数据一致性，选主服务。(可以看一下[Zookeeper和Kafka的关系，为啥Kafka依赖Zookeeper][Zookeeper_Kafka_Kafka_Zookeeper] )  
消费者:从主题分区中读取数据

**Kafka的发布订阅体现在，生产者将数据一个一个写入相应的主题，消费者通过订阅主题，从主题分区中读取一个一个数据。**  
下面是主题和分区图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 1]  
**消费者消费规则:**  
1.主题的一个分区只能供同个消费者群组的一个消费者消费。  
如图所示:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 2]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 3]  
2.如果分区数小于消费者数量，那么多出来的消费者啥也不干。  
如图所示  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 4]  
所以增加分区，往消费者群组里增加消费者是横向伸缩消费能力的主要方式。

**Kafka的安装**  
1.zookeeper的安装和使用  
先要装java1.8，然后下载zookeeper包，解压  
(1)先设置配置文件  
将conf目录下的zoo\_sample.cfg更名为zoo.cfg，默认端口为2181  
(2)使用bin/zkServer.sh start 启动zookeeper  
2.kafka安装  
解压缩包即可，默认端口9092，连接的zookeeper端口2181  
启动kafka  
bin/kafka-server-start.sh config/server.properties &  
也可以在一台机子上开启另外2个节点broker然后再启动  
cp config/server.properties config/server-1.properties  
cp config/server.properties config/server-2.properties  
bin/kafka-server-start.sh config/server-1.properties &  
bin/kafka-server-start.sh config/server-2.properties &  
关闭kafka  
bin/kafka-server-stop.sh

**简单实用Kafka**  
创建主题，只有一个副本，一个分区  
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test  
验证主题  
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test  
往测试主题上发布消息:  
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test  
读取消息  
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning(旧版本)  
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning(新版本2.1)//从头开始读  
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test //只读最新

查询某个group\_id的消费情况,假设这里有group\_id logstash\_1  
./kafka-consumer-groups.sh --group logstash\_1 --describe --bootstrap-server localhost:9092  
查看所有消费者  
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list  
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --list  
**使用java编写生产者和消费者**  
**pom.xml**

<?xml version="1.0" encoding="UTF-8"?>
    
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    
    	<groupId>com.zte.apt</groupId>
    	<artifactId>KafkaTest</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    
    	<name>KafkaTest</name>
    	
    	<url>http://www.example.com</url>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<maven.compiler.source>1.8</maven.compiler.source>
    		<maven.compiler.target>1.8</maven.compiler.target>
    	</properties>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.apache.kafka</groupId>
    			<artifactId>kafka-clients</artifactId>
    			<version>2.1.0</version>
    		</dependency>
    		<dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.25</version>
    </dependency>
    		
    	</dependencies>
    
    	<build>
    		<pluginManagement>
    			<plugins>
    			<plugin>  
                <groupId>org.apache.maven.plugins</groupId>  
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                <archive>  
                        <manifest>  
                            <mainClass>com.zte.apt.KafkaTest.KafkaProducerTest</mainClass>  
                        </manifest>  
                    </archive>
                    <descriptorRefs>  
                        <descriptorRef>jar-with-dependencies</descriptorRef>  
                    </descriptorRefs>  
                </configuration>  
                <executions>  
                    <execution>  
                        <id>make-assembly</id>  
                        <phase>package</phase>  
                        <goals>  
                            <goal>single</goal>  
                        </goals>  
                    </execution>  
                </executions>  
            </plugin>  
    			</plugins>
    		</pluginManagement>
    	</build>
    </project>

**生产者**  
1创建Properties配置对象，配置broker地址，键，键值的序列化等等  
2.创建生产者KafkaProducer  
3.创建记录对象ProducerRecord对象,往ProducerRecord存放主题，键，键值  
4.KafkaProducer发送ProducerRecord对象，其中发送方式有:  
(1)fire-and-forget发送并忘记，并不关心是否正常到达，虽然生产者有自动尝试重发机制，但有时候也会丢失一些信息  
(2)send(ProducerRecord)同步发送，它会返回一个future对象，调用get()方法进行等候，就可以知道消息是否发送成功  
(3)send(ProducerRecord,Callback)异步发送，多一个回调函数  
5.代码

package com.zte.KafkaTest;
    
    import java.util.Properties;
    import java.util.concurrent.Future;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    
    public class KafkaProducerTest { 
    	public static void main(String[] args) { 
    		Properties props = new Properties();
    		props.put("bootstrap.servers", "10.41.41.202:9092");
    		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    		Producer<String, String> producer = new KafkaProducer<>(props);
    		//主题的名字，发送的键，值
    		ProducerRecord<String, String> record = new ProducerRecord<>("topictest", "keyhaha", "valuehehe");
    		try
    		{ 
    			for (int i = 0; i < 100; i++)
    			     producer.send(new ProducerRecord<String, String>("topictest", Integer.toString(i), Integer.toString(i)));
    			System.out.println("已经发送完毕");
    			//异步发送，为了发送成功，主函数最好等待一小段时间
    			Thread.sleep(5000);
    			
    			//同步发送.调用get等待
    // Future<RecordMetadata> result = producer.send(record);
    // System.out.println(result.get().partition());
    		}
    		catch(Exception e)
    		{ 
    			e.printStackTrace();
    		}
    	}
    }

**消费者**  
向消费者群组增加消费者是横向伸缩消费能力的主要方式，一个主题的每一个分区只能分配给同一群组的一个消费者  
所以我们有必要为主题创建大量的分区，在负载增长时可以加入更多的消费者，不过要注意，不要让消费者的数量超过主题分区的数量，多余的消费者只会被闲置。  
1.创建Properties配置对象，配置broker地址，groupid 群组名字，键，键值的反序列化等等  
2.创建消费者KafkaConsumer  
3.订阅consumer.subscribe(Collections.singletonList(“主题，也可使用正则”))  
4.轮询consumer.poll(100),返回ConsumerRecords记录对象，  
5.提交偏移量，即更新分区当前的位置  
(1)自动提交设置enable.auto,commit设置为true,每隔一段时间提交一次  
(2)同步提交commitSync，阻塞  
(3)异步提交commitAsync，非阻塞  
(4)也可以使用异步和同步结合提交

package com.zte.KafkaTest;
    
    import java.util.Collections;
    import java.util.Properties;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    public class KafkaConsumerTest { 
    
    	public static void main(String[] args) 
    	{ 
    		Properties props = new Properties();
    		props.put("bootstrap.servers", "10.41.41.202:9092");
    		//指定消费者所属群组的名字
    		props.put("group.id", "group1");
    		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    		//订阅单个主题
    		consumer.subscribe(Collections.singletonList("topictest"));
    // //订阅多个主题
    // consumer.subscribe(Arrays.asList("foo", "bar"));
    // //使用正则订阅主题
    // consumer.subscribe("test.*");
    		try
    		{ 
    			while (true) 
    			{ 
    		         ConsumerRecords<String, String> records = consumer.poll(100);
    		         for (ConsumerRecord<String, String> record : records)
    		         { 
    		             System.out.printf("tppic=%s,partition=%s,offset=%d, key=%s, value =%s%n",record.topic(),record.partition(),record.offset(),record.key(),record.value());
    		         }
    		     }
    		}
    		finally
    		{ 
    			consumer.close();
    		}
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70]: /images/20220401/df220f0533034faa819879d986b56d35.png
[Zookeeper_Kafka_Kafka_Zookeeper]: https://blog.csdn.net/u011311291/article/details/85264046
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 1]: /images/20220401/91909e6771514db2a197087c4701c2f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20181229152035789.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 3]: /images/20220401/e7aaf61806c540698c37c83fc96c5ae4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEzMTEyOTE_size_16_color_FFFFFF_t_70 4]: /images/20220401/d988e3a7582c41eb82eeecbd62a6cdfc.png