Kafka应用入门

Kafka是一款基于发布和订阅的消息系统
运作方式如图:

各个组件的作用:
生产者:将数据依据主题，分区发往broker
broker:broker接收来自生产者的消息，为休息设置偏移量，并将消息保存到磁盘
zookeeper:管理各个分布式broker节点，为各个节点提供数据共享，数据一致性，选主服务。(可以看一下Zookeeper和Kafka的关系，为啥Kafka依赖Zookeeper )
消费者:从主题分区中读取数据

Kafka的发布订阅体现在，生产者将数据一个一个写入相应的主题，消费者通过订阅主题，从主题分区中读取一个一个数据。
下面是主题和分区图:

消费者消费规则:
1.主题的一个分区只能供同个消费者群组的一个消费者消费。
如图所示:

2.如果分区数小于消费者数量，那么多出来的消费者啥也不干。
如图所示

所以增加分区，往消费者群组里增加消费者是横向伸缩消费能力的主要方式。

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-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)

使用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>
	<!-- FIXME change it to the project's website -->
	<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><!-- lock down plugins versions to avoid using Maven 
				defaults (may be moved to parent pom) -->
			<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();
		}
	}
}