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();
}
}