一. 为什么使用消息系统

1. 解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。看这么个场景。A 系统发送数据到 BCD 三个系统，通过接口调用发送，如果 E 系统也要这个数据呢？那如果 C 系统现在不需要了呢？A 系统负责人几乎崩溃。A 系统要时时刻刻考虑 BCDE 四个系统如果挂了该咋办？要不要重发，要不要把消息存起来？
2. 异步通信： 允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。
3. 缓冲：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
4. 顺序保证:在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka保证一个Partition内的消息的有序性）
   5.峰值处理能力: 数据量急剧增加的时候，每次消费数据的能力不变，不会对系统造成影响。

二. kafka

三. kafka的定义

Kafka是一个分布式的发布订阅消息系统，由Scala写成，最初是由LinkedIn公司开发，后来捐赠给了Apache软件基金会。

四. kafka的架构

(1) Producer : 消息生产者，就是向kafka broker发消息的客户端；
(2) Consumer : 消息消费者，向kafka broker取消息的客户端;
(3) Topic : 可以理解为一个队列, 消息的分类。
(4) Broker : 一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic；
(5) Partition : 为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序；
(6) Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka。
(7) Consumer Group （CG）：这是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic；

五. kafka环境搭建

1. 参数配置server.properties

#broker的全局唯一编号，不能重复
broker.id=0
#kafka运行日志存放的路径	
log.dirs=/opt/module/kafka/logs
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop102:2181,hadoop103:2181

2. kafka通过zookeeper来管理

zookeeper环境搭建教程：

https://blog.csdn.net/qinqinde123/article/details/102854529

六. kafka的操作命令

1. 查看topic的列表

bin/kafka-topics.sh --zookeeper node102:2181 --list

2. 创建topic

bin/kafka-topics.sh --zookeeper node102:2181 
--create --replication-factor 3 --partitions 1 --topic first

3. 删除topic

bin/kafka-topics.sh --zookeeper hadoop102:2181 
--delete --topic first
需要server.properties中设置delete.topic.enable=true，重启

4. 发送消息

bin/kafka-console-producer.sh 
--broker-list node102:9092 --topic first

5. 消费者

bin/kafka-console-consumer.sh
--zookeeper node102:2181 --from-beginning --topic first

6. 查看某个Topic的详情

bin/kafka-topics.sh --zookeeper node102:2181
--describe --topic first

七. kafka生产过程分析

1.存储方式

(1) Kafka作为一个支持大数据量写入写出的消息队列，由于是基于Scala和Java实现的，而Scala和Java均需要在JVM上运行，所以如果是基于内存的方式，即JVM的堆来进行数据存储则需要开辟很大的堆来支持数据读写，从而会导致GC频繁影响性能。考虑到这些因素，kafka是使用磁盘而不是kafka服务器broker进程内存来进行数据存储，并且基于磁盘顺序读写和MMAP技术来实现高性能。
(2) 整个分区的数据不是由一个数据文件存放的，而是由多个segments组成的，即上面看到的0000.log文件是其中一个segment文件，文件名是以该文件的第一个数据相对于该分区的全局offset命名的。每当segment文件达到一定的大小，则会创建一个新的segment文件，具体大小在server.properties配置：默认为1G。

(3)当broker接收到producer发送过来的消息时，需要根据消息的主题和分区信息，将该消息写入到该分区当前最后的segment文件中，文件的写入方式是追加写。
由于是对segment文件追加写，故实现了对磁盘文件的顺序写，避免磁盘随机写时的磁盘寻道的开销，同时由于是追加写，故写入速度与磁盘文件大小无关.

2. 写入顺序

(1) producer先从zookeeper找到该partition的leader
(2) producer将消息发送给该leader
(3) leader将消息写入本地log
(4) followers从leader pull消息，写入本地log后向leader发送ACK
(5)leader收到所有ISR中的replication的ACK后，增加HW（high watermark，最后commit 的offset）并向producer发送ACK

3. 页缓存PageCache

虽然消息写入是磁盘顺序写入，没有磁盘寻道的开销，但是如果针对每条消息都执行一次磁盘写入，则也会造成大量的磁盘IO，影响性能。
所以在消息写入方面，broker基于MMAP技术，即内存映射文件，将消息先写入到操作系统的页缓存中，由页缓存直接映射到磁盘文件，不需要在用户空间和内核空间直接拷贝消息，所以也可以认为消息传输是发送在内存中的。

4.数据存入分区的原则

(1) 指定了partition，则直接写入
(2) 未指定partition但指定了key，根据key求hash值，处于分区个数，取余(hashcode % partions)，指定分区。
(3) partition和key均未指定，则使用轮询指定一个partition

5. 存储策略

无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据：
1）基于时间：log.retention.hours=168
2）基于大小：log.retention.bytes=1073741824

八. zookeeper中存储的数据

九. 创建生产者的代码

public class NewProducer {

	public static void main(String[] args) {
		
		Properties props = new Properties();
		// Kafka服务端的主机名和端口号
		props.put("bootstrap.servers", "hadoop103:9092");
		// 等待所有副本节点的应答
		props.put("acks", "all");
		// 消息发送最大尝试次数
		props.put("retries", 0);
		// 一批消息处理大小
		props.put("batch.size", 16384);
		// 请求延时
		props.put("linger.ms", 1);
		// 发送缓存区内存大小
		props.put("buffer.memory", 33554432);
		// key序列化
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		// value序列化
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

		Producer<String, String> producer = new KafkaProducer<>(props);
		for (int i = 0; i < 50; i++) {
			producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), "hello world-" + i));
		}
		producer.close();
	}
}

十. 创建消费者的代码

public class CustomNewConsumer {

	public static void main(String[] args) {

		Properties props = new Properties();
		// 定义kakfa 服务的地址，不需要将所有broker指定上 
		props.put("bootstrap.servers", "hadoop102:9092");
		// 制定consumer group 
		props.put("group.id", "test");
		// 是否自动确认offset 
		props.put("enable.auto.commit", "true");
		// 自动确认offset的时间间隔 
		props.put("auto.commit.interval.ms", "1000");
		// key的序列化类
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		// value的序列化类 
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		// 定义consumer 
		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
		
		// 消费者订阅的topic, 可同时订阅多个 
		consumer.subscribe(Arrays.asList("first", "second","third"));

		while (true) {
			// 读取数据，读取超时时间为100ms 
			ConsumerRecords<String, String> records = consumer.poll(100);
			
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
		}
	}
}

kafka拦截器

待写