注意：配置基于Kafka 0.8.2.1

broker配置

#非负整数，用于唯一标识broker
broker.id 0

#kafka持久化数据存储的路径，可以指定多个，以逗号分隔
log.dirs /tmp/kafka-logs

#broker接收连接请求的端口
port 9092

#指定zk连接字符串，[hostname:port]以逗号分隔
zookeeper.connect

#单条消息最大大小控制，消费端的最大拉取大小需要略大于该值
message.max.bytes 1000000

#接收网络请求的线程数
num.network.threads 3

#用于执行请求的I/O线程数
num.io.threads 8

#用于各种后台处理任务（如文件删除）的线程数
background.threads 10

#待处理请求最大可缓冲的队列大小
queued.max.requests 500

#配置该机器的IP地址
host.name 

#默认分区个数
num.partitions  1

#分段文件大小，超过后会轮转
log.segment.bytes 1024 * 1024 * 1024

#日志没达到大小，如果达到这个时间也会轮转
log.roll.{ms,hours}   168

#日志保留时间
log.retention.{ms,minutes,hours}

#不存在topic的时候是否自动创建
auto.create.topics.enable true

#partition默认的备份因子
default.replication.factor 1

#如果这个时间内follower没有发起fetch请求，被认为dead，从ISR移除
replica.lag.time.max.ms   10000

#如果follower相比leader落后这么多以上消息条数，会被从ISR移除
replica.lag.max.messages  4000

#从leader可以拉取的消息最大大小
replica.fetch.max.bytes 1024 * 1024

#从leader拉取消息的fetch线程数
num.replica.fetchers 1

#zk会话超时时间
zookeeper.session.timeout.ms  6000

#zk连接所用时间
zookeeper.connection.timeout.ms

#zk follower落后leader的时间
zookeeper.sync.time.ms 2000

#是否开启topic可以被删除的方式
delete.topic.enable false

producer配置

#参与消息确认的broker数量控制，0代表不需要任何确认 1代表需要leader replica确认 -1代表需要ISR中所有进行确认
request.required.acks 0

#从发送请求到收到ACK确认等待的最长时间（超时时间）
request.timeout.ms  10000

#设置消息发送模式，默认是同步方式， async异步模式下允许消息累计到一定量或一段时间又另外线程批量发送，吞吐量好但丢失数据风险增大
producer.type sync

#消息序列化类实现方式，默认是byte[]数组形式
serializer.class kafka.serializer.DefaultEncoder

#kafka消息分区策略实现方式，默认是对key进行hash
partitioner.class kafka.producer.DefaultPartitioner

#对发送的消息采取的压缩编码方式，有none|gzip|snappy
compression.codec none

#指定哪些topic的message需要压缩
compressed.topics  null

#消息发送失败的情况下，重试发送的次数 存在消息发送是成功的，只是由于网络导致ACK没收到的重试，会出现消息被重复发送的情况
message.send.max.retries 3

#在开始重新发起metadata更新操作需要等待的时间
retry.backoff.ms 100

#metadata刷新间隔时间，如果负值则失败的时候才会刷新，如果0则每次发送后都刷新，正值则是一种周期行为
topic.metadata.refresh.interval.ms 600 * 1000

#异步发送模式下，缓存数据的最长时间，之后便会被发送到broker
queue.buffering.max.ms 5000

#producer端异步模式下最多缓存的消息条数
queue.buffering.max.messages 10000

#0代表队列没满的时候直接入队，满了立即扔弃，-1代表无条件阻塞且不丢弃
queue.enqueue.timeout.ms -1

#一次批量发送需要达到的消息条数，当然如果queue.buffering.max.ms达到的时候也会被发送
batch.num.messages 200

consumer配置

#指明当前消费进程所属的消费组，一个partition只能被同一个消费组的一个消费者消费
group.id

#针对一个partition的fetch request所能拉取的最大消息字节数，必须大于等于Kafka运行的最大消息
fetch.message.max.bytes  1024 * 1024

#是否自动周期性提交已经拉取到消费端的消息offset
auto.commit.enable true

#自动提交offset到zookeeper的时间间隔
auto.commit.interval.ms  60 * 1000

#消费均衡的重试次数
rebalance.max.retries  4

#消费均衡两次重试之间的时间间隔
rebalance.backoff.ms 2000

#当重新去获取partition的leader前需要等待的时间
refresh.leader.backoff.ms   200

#如果zookeeper上没有offset合理的初始值情况下获取第一条消息开始的策略smallest|largeset
auto.offset.reset largest

#如果其超时，将会可能触发rebalance并认为已经死去
zookeeper.session.timeout.ms  6000

#确认zookeeper连接建立操作客户端能等待的最长时间
zookeeper.connection.timeout.ms 6000

1.maven:

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.10</artifactId>
  <version>0.8.2.1</version>
</dependency>
2.kafka生产者代码：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
 
import java.util.Properties;
 
/**
 * 
 * @author FromX
 *
 */
public class KProducer {
 
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //kafka服务器地址
        props.put("bootstrap.servers", "slave1.com:6667,slave2.com:6667,slave3.com:6667");
        //ack是判断请求是否为完整的条件（即判断是否成功发送）。all将会阻塞消息，这种设置性能最低，但是最可靠。
        props.put("acks", "1");
        //retries,如果请求失败，生产者会自动重试，我们指定是0次，如果启用重试，则会有重复消息的可能性。
        props.put("retries", 0);
        //producer缓存每个分区未发送消息，缓存的大小是通过batch.size()配置设定的。值较大的话将会产生更大的批。并需要更多的内存(因为每个“活跃”的分区都有一个缓冲区)
        props.put("batch.size", 16384);
        //默认缓冲区可立即发送，即便缓冲区空间没有满；但是，如果你想减少请求的数量，可以设置linger.ms大于0.这将指示生产者发送请求之前等待一段时间
        //希望更多的消息补填到未满的批中。这类似于tcp的算法，例如上面的代码段，可能100条消息在一个请求发送，因为我们设置了linger时间为1ms，然后，如果我们
        //没有填满缓冲区，这个设置将增加1ms的延迟请求以等待更多的消息。需要注意的是，在高负载下，相近的时间一般也会组成批，即使是linger.ms=0。
        //不处于高负载的情况下，如果设置比0大，以少量的延迟代价换取更少的，更有效的请求。
        props.put("linger.ms", 1);
        //buffer.memory控制生产者可用的缓存总量，如果消息发送速度比其传输到服务器的快，将会耗尽这个缓存空间。当缓存空间耗尽，其他发送调用将被阻塞，阻塞时间的阈值
        //通过max.block.ms设定，之后他将抛出一个TimeoutExecption。
        props.put("buffer.memory", 33554432);
        //key.serializer和value.serializer示例：将用户提供的key和value对象ProducerRecord转换成字节，你可以使用附带的ByteArraySerizlizaer或StringSerializer处理简单的byte和String类型.
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //设置kafka的分区数量
        props.put("kafka.partitions", 12);
        
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 50; i++){
            System.out.println("key-->key"+i+"  value-->vvv"+i);
            producer.send(new ProducerRecord<String, String>("aaa", "key"+i, "vvv"+i));
            Thread.sleep(1000);
        }
              
        producer.close();
    }
}
3.kafka消费者代码：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
 
import java.util.Arrays;
import java.util.Properties;
 
/**
 * 
 * @author FromX
 *
 */
public class KConsumer {
 
    public KafkaConsumer<String, String>  getConsmer() {
        Properties props = new Properties();
        //设置kafka服务器
        props.put("bootstrap.servers", "c1.wb3.com:6667,n1.wb1.com:6667");
        //消费者群组ID，发布-订阅模式，即如果一个生产者，多个消费者都要消费，那么需要定义自己的群组，同一个群组内的消费者只有一个能消费到消息
        props.put("group.id", "test");
        //true，消费者的偏移量将在后台定期提交；false关闭自动提交位移，在消息被完整处理之后再手动提交位移
        props.put("enable.auto.commit", "true");
        //如何设置为自动提交（enable.auto.commit=true）,这里设置自动提交周期
        props.put("auto.commit.interval.ms", "1000");
        //session.timeout.ms:在使用kafka的组管理时，用于检测消费者故障的超时
        props.put("session.timeout.ms", "30000");
        //key.serializer和value.serializer示例：将用户提供的key和value对象ProducerRecord转换成字节，你可以使用附带的ByteArraySerizlizaer或StringSerializer处理简单的byte和String类型.
        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);
        return consumer;
    }
 
    
    
    public static void main(String[] args) {
        KConsumer kconsumer =  new KConsumer();
        KafkaConsumer<String, String> consumer = kconsumer.getConsmer();
        
        consumer.subscribe(Arrays.asList("aaa"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.println("offset =  "+record.offset()+", key = "+record.key()+", value = "+ record.value());
        }
    }
}
4.官网文档地址：http://kafka.apache.org/documentation.html#configuration

5.极限情况的数据丢失现象

a:即使将ack设置为"all"也会在一定情况下丢失消息，因为kafka的高性能特性，消息在写入kafka时并没有落盘而是写入了OS buffer中，使用Os的脏页刷新策略周期性落盘，就算落盘 仍然会有raid buffer。前者机器宕机数据丢失，后者机器跳电数据丢失。

b:对数据可靠性比较高的场景建议offset手动提交，自动提交当遇到业务系统上线关闭时，消息读取并且offset已经提交，但是数据没有储存或者仍没来得及消费时，消息状态在内存中无法保留，重启应用会跳过消息，致使消息丢失。
---------------------