mac 搭建kafka系列教程

brew install kafka

/usr/local/Cellar/zookeeper
/usr/local/Cellar/kafka

/usr/local/etc/kafka/server.properties
/usr/local/etc/kafka/zookeeper.properties

#创建topic 
kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
#查看创建的topic
kafka-topics --list --zookeeper localhost:2181
#终端1发送消息
kafka-console-producer --broker-list localhost:9092 --topic test

#终端2接受消息
kafka-console-consumer --bootstrap-server localhost:9092 --topic test --from-beginning

#指定分区消费消息
kafka-console-consumer --bootstrap-server localhost:9092 --topic test --partition 1 --from-beginning
#添加分区
kafka-topics --alter --zookeeper localhost:2181 --partitions 10  --topic test    //添加10个
#启动kafka，指定配置文件，后台启动并打印日志到 /usr/local/etc/kafka/kafka.log
 nohup kafka-server-start /usr/local/etc/kafka/server.properties > /usr/local/etc/kafka/kafka.log 2>&1 &

# topic：
1.kafka集群会将每个topic进行分区，每个分区都是一个排序且不可改变的队列，新的消息会进入队尾并分配一个唯一ID，官方称之为偏移量(offset)
2.无论消息是否被消费，集群都会保留消息,有一个配置的时间（过期时间）,超过这个时间后，消息会被清除
3.消费端唯一记录的元信息就是自己在topic中的位置(offset)，
4.分布式的原因：第一集群可以容纳大量的数据 第二：可以并行的处理

at last once：至少消费一次（对一条消息有可能多次消费，有可能会造成重复消费数据）
     原因：Proudcer产生数据的时候，已经写入在broker中，但是由于broker的网络异常，没有返回ACK，这时Producer,认为数据没有写入成功，此时producer会再次写入，相当于一条数据，被写入了多次。
     
at most once：最多消费一次，对于消息，有可能消费一次，有可能一次也消费不了
    原因：producer在产生数据的时候，有可能写数据的时候不成功，此时broker就跳过这个消息，那么这条数据就会丢失，导致consumer无法消费。
    
exactly once：有且仅有一次。这种情况是我们所需要的，也就是精准消费一次。
 kafka中消费语义的场景
   at last once：可以先读取数据，处理数据，最后记录offset，当然如果在记录offset之前就crash,新的consumer会重复的来消费这条数据，导致了”最少一次“
   at most once：可以先读取数据，然后记录offset，最后在处理数据，这个方式，就有可能在offset后，还没有及时的处理数据，就crash了，导致了新的consumer继续从这个offset处理，那么刚刚还没来得及处理的数据，就永远不会被处理，导致了”最多消费一次“
   exactly once：可以通过将提交分成两个阶段来解决：保存了offset后提交一次，消息处理成功后，再提交一次。

kafka中如何实现精准写入数据？
A：Producer 端写入数据的时候保证幂等性操作：
幂等性：对于同一个数据无论操作多少次都只写入一条数据，如果重复写入，则执行不成功
B：broker写入数据的时候，保证原子性操作, 要么写入成功，要么写入失败。（不成功不断进行重试）

/**
     * The offset commit behavior enumeration.
     */
    public enum AckMode {
        /**
         * Commit after each record is processed by the listener.
         */
        RECORD,
        /**
         * Commit whatever has already been processed before the next poll.
         */
        BATCH,
        /**
         * Commit pending updates after
         * {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
         */
        TIME,
        /**
         * Commit pending updates after
         * {@link ContainerProperties#setAckCount(int) ackCount} has been
         * exceeded.
         */
        COUNT,
        /**
         * Commit pending updates after
         * {@link ContainerProperties#setAckCount(int) ackCount} has been
         * exceeded or after {@link ContainerProperties#setAckTime(long)
         * ackTime} has elapsed.
         */
        COUNT_TIME,
        /**
         * User takes responsibility for acks using an
         * {@link AcknowledgingMessageListener}.
         */
        MANUAL,
        /**
         * User takes responsibility for acks using an
         * {@link AcknowledgingMessageListener}. The consumer is woken to
         * immediately process the commit.
         */
        MANUAL_IMMEDIATE,
    }

RECORD
每处理一条commit一次
BATCH(默认)
每次poll的时候批量提交一次，频率取决于每次poll的调用频率
TIME 
每次间隔ackTime的时间去commit(跟auto commit interval有什么区别呢？)
COUNT 
累积达到ackCount次的ack去commit
COUNT_TIME
ackTime或ackCount哪个条件先满足，就commit
MANUAL
listener负责ack，但是背后也是批量上去
MANUAL_IMMEDIATE
listner负责ack，每调用一次，就立即commit