Kafka是一种高吞吐量的分布式发布订阅消息系统，是观察者模式的一种实现。
Kafka是一个分布式消息队列。Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群有多个kafka实例组成，每个实例(server)称为broker。
无论是kafka集群，还是consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性。

1. 通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
2. 高吞吐量 ：即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。
3. 支持通过Kafka服务器和消费机集群来分区消息。
4. 支持Hadoop并行数据加载。

一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
kafka/config/server.properties配置文件：

#broker的全局唯一编号，不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径	
log.dirs=/opt/module/kafka/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181

可以理解为一个队列。每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。
物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处。

Partition是物理上的概念，每个Topic包含一个或多个Partition.
为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序

负责发布消息到Kafka broker

消息消费者，向Kafka broker读取消息的客户端。
Kafak中同一消费者组不能同时消费一个分区，只能去消费不同分区，因此在生产环境中，消费者consumer的数量和分区数保持一致即可。

每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。
是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic；

kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka。

producer采用推（push）模式将消息发布到broker，每条消息都被追加（append）到分区（patition）中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）。

消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成
每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值。
1）分区的原因
（1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；
（2）可以提高并发，因为可以以Partition为单位读写了。
2）分区的原则
（1）指定了patition，则直接使用；
（2）未指定patition但指定key，通过对key的value进行hash出一个patition；
（3）patition和key都未指定，使用轮询选出一个patition。

同一个partition可能会有多个replication（对应 server.properties 配置中的 default.replication.factor=N）。没有replication的情况下，一旦broker 宕机，其上所有 patition 的数据都不可被消费，同时producer也不能再将数据存于其上的patition。引入replication之后，同一个partition可能会有多个replication，而这时需要在这些replication之间选出一个leader，producer和consumer只与这个leader交互，其它replication作为follower从leader 中复制数据。

1）producer先从zookeeper的 "/brokers/…/state"节点找到该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。
ACK可以设置 0 1 all等。0代表发送出去不接受确认信息，1代表只要leader回复确认信息即可，all代表kafka集群必须全部回复确认信息。

物理上把topic分成一个或多个patition（对应 server.properties 中的num.partitions=3配置），每个patition物理上对应一个文件夹（该文件夹存储该patition的所有消息和索引文件）

无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据：
1）基于时间：log.retention.hours=168
2）基于大小：log.retention.bytes=1073741824
需要注意的是，因为Kafka读取特定消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高 Kafka 性能无关。

kafka提供了两套consumer API：高级Consumer API和低级Consumer API。

1）高级API优点
高级API 写起来简单
不需要自行去管理offset，系统通过zookeeper自行管理。
不需要管理分区，副本等情况，系统自动管理。
消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据（默认设置1分钟更新一下zookeeper中存的offset）
可以使用group来区分对同一个topic 的不同程序访问分离开来（不同的group记录不同的offset，这样不同程序读取同一个topic才不会因为offset互相影响）
2）高级API缺点
不能自行控制offset（对于某些特殊需求来说）
不能细化控制如分区、副本、zk等

1）低级 API 优点
能够让开发者自己控制offset，想从哪里读取就从哪里读取。
自行控制连接分区，对分区自定义进行负载均衡
对zookeeper的依赖性降低（如：offset不一定非要靠zk存储，自行存储offset即可，比如存在文件或者内存中）
2）低级API缺点
太过复杂，需要自行控制offset，连接哪个分区，找到分区leader 等。

消费者是以consumer group消费者组的方式工作，由一个或者多个消费者组成一个组，共同消费一个topic。每个分区在同一时间只能由group中的一个消费者读取，但是多个group可以同时消费这个partition。有一个由三个消费者组成的group，有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。
在这种情况下，消费者可以通过水平扩展的方式同时读取大量的消息。另外，如果一个消费者失败了，那么其他的group成员会自动负载均衡读取之前失败的消费者读取的分区。

consumer采用pull（拉）模式从broker中读取数据。
push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息。
对于Kafka而言，pull模式更合适，它可简化broker的设计，consumer可自主控制消费消息的速率，同时consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。
pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直等待数据到达。为了避免这种情况，我们在我们的拉请求中有参数，允许消费者请求在等待数据到达的“长轮询”中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小）。

可以使用spring的kafka，依赖如下：

<dependency>
            <groupId>org.springframework.kafka</groupId>
           <artifactId>spring-kafka</artifactId>
        </dependency>

原生用法如下。

 <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>1.1.1</version>
        </dependency>

spring:
  kafka:
    properties:
      max.request.size: 10097152
      zookeeper.connection.timeout.ms: 100000
    consumer:
      bootstrap-servers: 10.10.10.10:9092
      enable-auto-commit: true
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        max.request.size: 10097152
        max.partition.fetch.bytes: 10097152
      group-id: demo-id

@Configuration
public class KafkaConfig {

    @Bean
    public Properties getProperties(){

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.91.159.28:9092,10.91.159.29:9092,10.91.159.30:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("lingers.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return properties;
    }

    @Bean
    public KafkaProducer<String, String> kafkaProducer(@Autowired @Qualifier("getProperties") Properties kafkaProperties) {
        return new KafkaProducer<String, String>(kafkaProperties);
    }
}

    @Autowired
    private Producer<String, String> producer;
    producer.send(new ProducerRecord<String, String>("luowenhui", JSONObject.toJSONString(userinfo)));

注意：给kafka里送的消息，一般一条条的发送数据，这样可以轮询分发给不同的consumer，提高性能。对于有关联的数据可以选择批量发送。

@Configuration
public class KafkaConfig {
    @Bean
    public Properties kafkaProperties()  {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.91.159.28:9092,10.91.159.29:9092,10.91.159.30:9092");
        properties.put("group.id", "hello");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        return properties;
    }

    @Bean
    public KafkaConsumer<String, String> kafkaConsumer0(@Autowired @Qualifier("kafkaProperties") Properties kafkaProperties) {
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(kafkaProperties);
        kafkaConsumer.assign(Arrays.asList(new TopicPartition("luo", 0)));//按区接收
        //消费者订阅的topic, 可同时订阅多个 
		//consumer.subscribe(Arrays.asList("first", "second","third"));
        return kafkaConsumer;
    }
}

assign方法和subscribe区别：
Kafka里的消费者组有两个使用的场景：

1. “队列模式”：在同一组的消费者共同消费一个主题的所有消息，而且确保一条消息只被一个消费者处理。一个主题的所有的分区会和一个消费组的所有消费者做关联：每个消费者和一到多个分区做关联，接收它们的消息。反向说，一个分区只会与一个消费者关联，它的消息不会被其它的消费者接收。通过assign实现。 最开始只有一个消费者时，所有的分区都分配给了它。当消息的规模增加时，我们就需要扩展消费者的数量，水平扩展处理能力，一直可以达到每个消费者只关联一个分区。大于分区数的消费者是会处在空闲状态，因为没有分配任何的分区。
2. “发布/订阅模式” 创建不同的消费者组意味一个主题的消息会发送给所有订阅它的消费者组，然后消费者组依照前面共同协作的场景进行分配。这往往是因为我们有不同的应用需求，比如一批交易数据，资金系统、ERP系统会消费它而风险监控也需要同时消费它。这就实现了数据的透明异步共用。

 while (true) {

            ConsumerRecords<String, String> records = consumer0.poll(100);
            for (ConsumerRecord<String, String> record : records) {//record.value()就是获取到的数据

            }
        }

Ctrl+C停止命令