从未如此简单：10分钟带你逆袭Kafka！(12)

Kafka 的日志

Kafka 的日志分两种:

第一种日志是我们的 Kafka 的启动日志，就是我们排查问题，查看报错信息的日志。

第二种日志就是我们的数据日志，Kafka 是我们的数据是以日志的形式存在存盘中的，我们第二种所说的日志就是我们的 Partiton 与 Segment。

那我们就来说说备份和分区吧：我们创建一个分区，一个备份，那么 test 就应该在三台机器上或者三个数据目录只有一个 test-0。(分区的下标是从 0 开始的)

如果我们创建 N 个分区，我们就会在三个服务器上发现，test_0-n，如果我们创建 M 个备份，我们就会在发现，test_0 到 test_n 每一个都是 M 个。

Kafka API

使用 Kafka 原生的 API

①消费者自动提交

定义自己的生产者：

import org.apache.kafka.clients.producer.Callback; 

import org.apache.kafka.clients.producer.KafkaProducer; 

import org.apache.kafka.clients.producer.ProducerRecord; 

import org.apache.kafka.clients.producer.RecordMetadata; 

import java.util.Properties; 

/** 

 * @ClassName MyKafkaProducer 

 * @Description TODO 

 * @Author lingxiangxiang 

 * @Date 3:37 PM 

 * @Version 1.0 

 **/ 

public class MyKafkaProducer { 

    private org.apache.kafka.clients.producer.KafkaProducer<Integer, String> producer; 

    public MyKafkaProducer() { 

        Properties properties = new Properties(); 

        properties.put("bootstrap.servers", "192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094"); 

        properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"); 

        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 

        // 设置批量发送 

        properties.put("batch.size", 16384); 

        // 批量发送的等待时间50ms, 超过50ms, 不足批量大小也发送 

        properties.put("linger.ms", 50); 

        this.producer = new org.apache.kafka.clients.producer.KafkaProducer<Integer, String>(properties); 

    } 

    public boolean sendMsg() { 

        boolean result = true; 

        try { 

（编辑：ASP站长网）