作者：刘宇
CSDN博客地址：https://blog.csdn.net/liuyu973971883
有部分资料参考，如有侵权，请联系删除。如有不正确的地方，烦请指正，谢谢。

一、 Kafka生产者的api操作

1.1、添加依赖

<!-- 添加如下依赖，包含了kafka-clients -->
<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.11</artifactId>
	<version>1.1.1</version>
</dependency>

1.2、生产者案例

• 创建producer时的一些配置信息说明：

bootstrap.servers=192.168.40.101:9092,192.168.40.102:9092,192.168.40.103:9092 ## kafka的服务器
key.serializer=org.apache.kafka.common.serialization.IntegerSerializer ##Key的序列化器
value.serializer=org.apache.kafka.common.serialization.StringSerializer ##value的序列化器
acks=[0|-1|1|all] ##消息确认机制 0: 不做确认，直管发送消息即可 -1|all: 不仅leader需要将数据写入本地磁盘，并确认，还需要同步的等待其它followers进行确认 1:只需要leader进行消息确认即可，后期follower可以从leader进行同步
batch.size=1024 #每个分区内的用户缓存未发送record记录的空间大小
## 如果缓存区中的数据，没有沾满，也就是任然有未用的空间，那么也会将请求发送出去，为了较少请求次数，我们可以配置linger.ms大于0
linger.ms=10 ## 不管缓冲区是否被占满，延迟10ms发送request
buffer.memory=10240 #控制的是一个producer中的所有的缓存空间
retries=0 #发送消息失败之后的重试次数

• 生产者代码：

public class MyKafkaProducer {
	public static void main(String[] args) throws Exception { 
		/* 
			K: --->代表的是向topic中发送的每一条消息的key的类型，key可以为null
			V: --->代表的是向topic中发送的每一条消息的value的类型 
		*/ 
		Properties properties = new Properties(); 
		// properties.put("bootstrap.servers", "bigdata01:9092,bigdata02:9092,bigdata03:9092"); 
		properties.load(MyKafkaProducer.class.getClassLoader().getResourceAsStream("producer.proper ties"));
		Producer<Integer, String> producer = new KafkaProducer<Integer, String>(properties);
		int start = 10;
		int end = start + 10;
		for (int i = start; i < end; i++) {
			//发送数据
			ProducerRecord<Integer, String> record = new ProducerRecord("spark", "11111");
			producer.send(record); 
		}
		Thread.sleep(1000);
		producer.close();
	} 
}

• 创建producer.properties配置文件：

bootstrap.servers=192.168.40.101:9092,192.168.40.102:9092,192.168.40.103:9092
# specify the compression codec for all data generated: none, gzip, snappy, lz4
compression.type=none

# name of the partitioner class for partitioning events; default partition spreads data randomly
# 输入进入分区的方式
#partitioner.class=

# the maximum amount of time the client will wait for the response of a request
# 请求超时时间
#request.timeout.ms=

# how long `KafkaProducer.send` and `KafkaProducer.partitionsFor` will block for
# 使用send方法最大消息阻塞时间
#max.block.ms=

# the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together
linger.ms=5000

# the maximum size of a request in bytes
## 最大的请求大小
#max.request.size=

# the default batch size in bytes when batching multiple records sent to a partition
batch.size=1024

buffer.memory=10240
key.serializer=org.apache.kafka.common.serialization.IntegerSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

• 现象：
  因为我们在配置文件里面设置了linger.ms=5000，所以延迟等待5秒后一次性将数据发送到cluster

二、 Kafka消费者的api操作

2.1、添加依赖

<!-- 添加如下依赖，包含了kafka-clients -->
<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.11</artifactId>
	<version>1.1.1</version>
</dependency>

2.2、消费者案例

• 创建consumer.properties配置文件：

bootstrap.servers=192.168.40.101:9092,192.168.40.102:9092,192.168.40.103:9092
# consumer group id
group.id=test-consumer-group

# What to do when there is no initial offset in Kafka or if the current
# offset does not exist any more on the server: latest, earliest, none
auto.offset.reset=earliest
key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

• 消费者代码：

public class MyKafkaConsumer {
	public static void main(String[] args) throws Exception {
		//消费者
		Properties properties = new Properties();
		properties.load(MyKafkaConsumer.class.getClassLoader().getResourceAsStream("consumer.proper ties"));
		Consumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(properties);
		//订阅topic consumer.subscribe(Arrays.asList("spark"));
		//从kafka对应的topic中拉取数据
		while (true) {
			ConsumerRecords<Integer, String> consumerRecords = consumer.poll(1000);
			for (ConsumerRecord<Integer, String> record : consumerRecords) {
				Integer key = record.key();
				String value = record.value();
				int partition = record.partition();
				long offset = record.offset();
				String topic = record.topic();
				System.out.println(String.format("topic:%s\tpartition:%d\toffset:%d\tkey:%d\tvalue:%s", topic, partition, offset, key, value));
			}
		}
	}
}

三、 Kafka的分区策略

每一条producerRecord有，topic名称、可选的partition分区编号，以及一对可选的key和value组成。

• 如果指定的partition，那么直接进入该partition
• 如果没有指定partition，但是指定了key，使用key的hash选择partition
• 如果既没有指定partition，也没有指定key，使用轮询的方式进入partition

3.1、自定义分区之随机分区方式

自定义分区方式需要实现Partitioner接口。

• 代码：

public class RandomPartitioner implements Partitioner {
	public void close() {
	}
	public void configure(Map<String, ?> configs) {
	}
	private Random random = new Random();
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		Integer partitionCount = cluster.partitionCountForTopic(topic);//返回当前topic的partition个数
		int partition = random.nextInt(partitionCount);
		System.out.println("partition: " + partition); return partition;
	}
}

• 在生产者文件添加配置：

partitioner.class=com.brycen.kafka.partitioner.RandomPartitioner

3.2、自定义分区之hash分区方式

自定义分区方式需要实现Partitioner接口。

• 代码：

public class HashPartitioner implements Partitioner {
	public void close() {
	}
	public void configure(Map<String, ?> configs) {
	}
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		Integer partCount = cluster.partitionCountForTopic(topic);
		int partition = Math.abs(key.hashCode()) % partCount; System.out.println("key: " + key + "partition: " + partition);
		return partition;
	}
}

• 在生产者文件添加配置：

partitioner.class=com.brycen.kafka.partitioner.HashPartitioner

3.3、自定义分区之轮询分区方式

自定义分区方式需要实现Partitioner接口。

• 代码：

public class RoundRobinPartitioner implements Partitioner {
	public void close() {
	}
	public void configure(Map<String, ?> configs) {
	}
	
	//定义一个原子计数器
	private AtomicInteger count = new AtomicInteger();
	
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		int parCount = cluster.partitionCountForTopic(topic);
		int partition = count.getAndIncrement() % parCount;
		System.out.println("key: " + key + "\tpartition: " + partition);
		return partition;
	}
}

• 在生产者文件添加配置：

partitioner.class=com.brycen.kafka.partitioner.RoundRobinPartitioner