Kafka

# 在解压的kafka目录下（推荐），创建一个存放日志（也存放数据）的信息
mkdir logs

# 修改server.properties
# 每一个实例的唯一辨识（int）
broker.id=0
# 是否可以删除topic
delete.topic.enable=true
# 设置日志打印的位置为创建的日志目录
log.dirs=/opt/kafka/logs
# 缓存数据的时间为7天、大小为1G
log.retention.hours=168
log.segment.bytes=1073741824
# 修改zk集群
zookeeper.connect=${id}:${port},${id}:${port},${id}:${port}

# 集群分发，修改broker.id
xsync kafka/

# bin/下的命令
kafka-server-start.sh
kafka-server-stop.sh
# 对topic的操作
kafka-topics.sh
# 测试时，控制台的消费
kafka-console-consumer.sh
kafka-console-producer.sh

--topic  # 定义topic名
--replication-factor  # 定义副本数
--partitions  # 定义分区数
--daemon  # 后台启动

# kafka依赖zk，先启动zk，并确定zk状态为leader/foller
zkstart.sh
zk/bin/zkServer.sh status

# 使用kafka自带的zk，-daemon表示后台启动
./zookeeper-server-start.sh -daemon ../config/zookeeper.properties

# 指定配置文件启动kafka，阻塞进程
bin/kafka-server-start.sh config/server.properties

# 查看进程
jps -l

# 创建分区数为x（kafka/logs下的一个目录），副本数为x（不能超过集群的节点数），名称为x的topic
bin/kafka-topics.sh --create --zookeeper ${id}:${port} --partitions ${number} --replication-factor ${number} --topic ${name}
# 查看topic数量，显示为topic name
bin/kafka-topics.sh --list --zookeeper ${id}:${port}
# 删除topic，能创建同名的即删除成功
bin/kafka-topics.sh --zookeeper ${id}:${port} --delete --topic ${name}

# 发送消息，连接kafka集群（kafka默认9092），写进topic
bin/kafka-console-producer.sh --broker-list ${id}:${port} --topic ${name}
# 消费消息，连接zk集群，读取topic，不加--from-beginning表示只获取最新的
bin/kafka-console-consumer.sh --zookeeper ${id}:${port} --from-beginning --topic ${name}

# 新版本的kafka中consumer的offset存储在本地，提升效率，不交由zk保存，会报警告！bootstrap（附属于）本地kafka集群，名为__consumer_offset的topic中。
bin/kafka-console-consumer.sh --bootstrap-server ${id}:${port} --from-beginning --topic ${name}

# 查看topic的详情,Isr为选举（其中某个与宕机的Leader节点数据最相近，作为新的Leader），ReplicationFactor为副本，值为broker.id
bin/kafka-topics.sh --zookeeper ${id}:${port} --describe-topic ${name}

# 分区的好处：不同分区放在不同节点，实现了负载；消费者组只能消费不同分区数据，提高了并发度（topic的分区数应与一个消费者组中的消费者个数相同）
Broker1：Topic A-Partition 0-Leader、Topic A-Partition 1-Follower
                         Replication A/0                    Replication A/1
Broker2：Topic A-Partition 0-Follower、Topic A-Partition 1-Leader

分区的原因：
（1）方便在集群中扩展，每个partition可以通过调整以适应它所在的及其，而每个topic又可以由多个partition组成，因此整个集群就可以适应任意大小的数据；
（2）可以提高并发，同一个消费者组不能读取同一个分区的数据，因此可以以partition为单位读写。

分区的原则：
（1）指定了partition，则直接使用；
（2）未指定partition但指定key，通过对key的value进行hash出一个partition；
（3）partition和key都未指定，使用轮询选出一个partition。

kafka读取特定消息的时间复杂度为O（1），即与文件大小无关，删除过期文件与提高kafka性能无关。

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>${kafka.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.12</artifactId>
  <version>${kafka.version}</version>
</dependency>

1.高级API：带或不带回调函数的生产者
public class Producer {
    public static void main(String[] args) {
        // 设置配置文件
        Properties props = new Properties();
        // Kafka集群
        props.put("bootstrap.servers", "localhost:9092");
        // 应答级别，all可以写出-1
        props.put("acks", "all");
        // 重试次数
        props.put("retries", 0);
        // 批量大小和提交延迟，决定发送时刻
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        // 缓存
        props.put("buffer.memory", 33554432);
        // KV序列化类
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 发送数据，第二个参数可以加上回调函数，重写onCompletion(RecordMetadata, exception)函数
        producer.send(new ProducerRecord<>("first", String.valueOf(123)), (metadata, exception) -> {
            if (exception == null) {
                System.out.println(metadata.partition() + "-" + metadata.offset());
            } else {
                System.out.println("ERROR:" + exception);
            }
        });

        producer.close();
    }
}


// 2.低级API：自定义分区的生产者
// 实现Partition类，重写方法partition、close、configure，配置文件需要匹配生产者
props.put("partitioner.class", "${全类名}");

// 1.高级API
public class Consumer {
    public static void main(String[] args) {
        // 配置信息
        Properties props = new Properties();
        // kafka集群
        props.put("bootstrap.servers", "localhost:9092");
        // 消费者组id
        props.put("group.id", "test");
        // 设置自动提交offset
        props.put("enable.auto.commit", "true");
        // 提交延迟
        props.put("auto.commit.interval.ms", "1000");
        // KV的反序列化
        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);

        // 指定Topic
        consumer.subscribe(Arrays.asList("first", "second"));

        // 获取数据结束，JVM自动退出
        while (true) {
        
            // 获取数据，参数为获取延迟
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100);

            // 打印数据
            for (ConsumerRecord<String, String> record : consumerRecords) {
                System.out.println(record.topic() + "-" + record.partition() + "-" + record.value());
            }
        }
    }
}


// 2.低级API：读取指定topic、partition（找leader）、offset的数据
/** 
 * 主要步骤：
 * （1*）findLeader()，根据指定的分区从主题元数据中找到主副本；
 * （2）getLastOffset()，获取分区最新的消费进度；
 * （3*）run()，从主副本拉取分区的消息；
 * （4）findNewLeader()，识别主副本的变化，重试。
 **/
 // 找分区leader（元数据信息）
private BrokerEndPoint findLeader(List<String> brokers, int port, String topic, int partition) {
    for (String broker : brokers) {
        // 创建获取分区Leader的消费者对象，链接到具体某一个节点
        SimpleConsumer getLeader = new SimpleConsumer(broker, port, 1000,
                1024*4, "getLeader");

        // 创建一个主题元数据的信息请求
        TopicMetadataRequest topicMetadataRequest =
                new TopicMetadataRequest(Collections.singletonList(topic));

        // 获取返回值
        TopicMetadataResponse topicMetadataResponse = getLeader.send(topicMetadataRequest);

        // 解析元数据返回值
        List<TopicMetadata> topicsMetadata = topicMetadataResponse.topicsMetadata();
        for (TopicMetadata topicMetadata : topicsMetadata) {
            List<PartitionMetadata> partitionsMetadata = topicMetadata.partitionsMetadata();
            for (PartitionMetadata partitionMetadata : partitionsMetadata) {
                if (partition == partitionMetadata.partitionId()) {
                    return partitionMetadata.leader();
                }
            }
        }
    }
    return null;
}

// 获取数据
private void run(List<String> brokers, int port, String topic, int partition, long offset) {
    // 获取分区Leader
    BrokerEndPoint leader = findLeader(brokers, port, topic, partition);
    if (leader == null) {
        return;
    }
    String leaderHost = leader.host();

    // 创建获取数据的消费者
    SimpleConsumer simpleConsumer = new SimpleConsumer(leaderHost, port, 1000,
            1024 * 4, "getData");

    // 创建获取数据的对象（可以获取多个数据.addFetch()）
    FetchRequest fetchRequest =
            new FetchRequestBuilder().addFetch(topic, partition, offset, 1024 * 4).build();

    // 获取返回值
    FetchResponse fetchResponse = simpleConsumer.fetch(fetchRequest);

    // 解析返回值，创建获取数据的对象时可以多次.addFetch
    ByteBufferMessageSet messageAndOffsets = fetchResponse.messageSet(topic, partition);
    for (MessageAndOffset messageAndOffset : messageAndOffsets) {
        // offset可以自行保存
        long offset1 = messageAndOffset.offset();
        ByteBuffer payload = messageAndOffset.message().payload();

        // 自行反序列化
        byte[] bytes = new byte[payload.limit()];
        payload.get(bytes);

        System.out.println(offset1 + "-" + new String(bytes));
    }
}