brokers和消费者使用zk来获取状态信息和追踪消息坐标。
每一个partition是一个有序的，不可变的消息序列。
只有当partition里面的file置换到磁盘文件以后，才开放给消费者来消费。
每一个partition是跨服务器地被复制到其他地方，为了容错的目的。
这个partition可以理解为hadoop中block的单位。
但是只有被选择为leader的服务器partition来服务消费者的读和生产者的写，
followers只是把数据同步过去。同步状态较好的被列入ISR，这些ISR和leader
信息都保存在zk中，当leader状态异常，ISR中的某一个Follower变成新的leader.
在整个kafka集群中，每一个服务器扮演一个双重角色，它可能是某个top的leader partition,
也同时可以是另一个topic的follower partition.这确保了集群的负载均衡。
每一个消费者代表一个进程，多个消费者组成一个消费者组。
一个topic中的一条消息只能被一个消费者组中的某一个消费者消费，如果需要被多个消费者消费，则这些消费者需要在不同的消费者组中。
原因可能是以消费者组的单位在zk中保持partition的offset.
kafka的设计中，broker是无状态的，这意味着它并不负责管理哪些消费者消费了哪些partition中的消息到什么位置，甚至谁消费的都不理会。
对于消息保持策略，kafka采用了基于时间的SLA，一个消息将会被自动删除当它达到了这个SLA.
kafka的复制策略有两种，同步和异步，同步会在lead replica和follower都完成消息的存储后才给producer发确认信息。
异步同步，只要lead replica收到了信息，就给producer发确认信息，如果这个时候lead replica的broker出问题，就会有风险。
生产者
kafka的message api for producer
从前面分析得知，数据被封装成消息，如何发送给kafka呢？首先需要获取这个topic的 lead partition。
消息可以一条一条发送，也可以批量压缩异步发送。即攒到一定的数量或一定的时间再发送。
Producer:Kafka provides the kafka.javaapi.producer.Producer class (classProducer<K,V>)。默认的分区策略是对key进行hash.
import java.util.Date; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; public class SimpleProducer { private static Producer<String, String> producer; public SimpleProducer() { Properties props = new Properties(); // Set the broker list for requesting metadata to find the lead broker props.put("metadata.broker.list", "192.168.146.132:9092, 192.168.146.132:9093, 192.168.146.132:9094"); //This specifies the serializer class for keys props.put("serializer.class", "kafka.serializer.StringEncoder"); // 1 means the producer receives an acknowledgment once the lead replica // has received the data. This option provides better durability as the // client waits until the server acknowledges the request as successful. props.put("request.required.acks", "1"); ProducerConfig config = new ProducerConfig(props); producer = new Producer<String, String>(config); } public static void main(String[] args) { int argsCount = args.length; if (argsCount == 0 || argsCount == 1) throw new IllegalArgumentException( "Please provide topic name and Message count as arguments"); String topic = (String) args[0]; String count = (String) args[1]; int messageCount = Integer.parseInt(count); System.out.println("Topic Name - " + topic); System.out.println("Message Count - " + messageCount); SimpleProducer simpleProducer = new SimpleProducer(); simpleProducer.publishMessage(topic, messageCount); } private void publishMessage(String topic, int messageCount) { for (int mCount = 0; mCount < messageCount; mCount++) { String runtime = new Date().toString(); String msg = "Message Publishing Time - " + runtime; System.out.println(msg); // Creates a KeyedMessage instance KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, msg); // Publish the message producer.send(data); } // Close producer connection with broker. producer.close(); } }