标签：spring mvc+my batis kafka dubbo+zookeerper restful redis分布式缓存
如下图：
图.1
如图.1中，kafka 相关名词解释如下：
1.producer： 消息生产者，发布消息到 kafka 集群的终端或服务。 2.broker： kafka 集群中包含的服务器。 3.topic： 每条发布到 kafka 集群的消息属于的类别，即 kafka 是面向 topic 的。 4.partition： partition 是物理上的概念，每个 topic 包含一个或多个 partition。kafka 分配的单位是 partition。 5.consumer： 从 kafka 集群中消费消息的终端或服务。 6.Consumer group： high-level consumer API 中，每个 consumer 都属于一个 consumer group，每条消息只能被 consumer group 中的一个 Consumer 消费，但可以被多个 consumer group 消费。 7.replica： partition 的副本，保障 partition 的高可用。 8.leader： replica 中的一个角色， producer 和 consumer 只跟 leader 交互。 9.follower： replica 中的一个角色，从 leader 中复制数据。 10.controller： kafka 集群中的其中一个服务器，用来进行 leader election 以及 各种 failover。 12.zookeeper： kafka 通过 zookeeper 来存储集群的 meta 信息。
kafka 在 zookeeper 中的存储结构如下图所示：
图.2
producer 采用 push 模式将消息发布到 broker，每条消息都被 append 到 patition 中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。
producer 发送消息到 broker 时，会根据分区算法选择将其存储到哪一个 partition。其路由机制为：
1. 指定了 patition，则直接使用； 2. 未指定 patition 但指定 key，通过对 key 的 value 进行hash 选出一个 patition 3. patition 和 key 都未指定，使用轮询选出一个 patition。
附上 Java 客户端分区源码，一目了然：
//创建消息实例 public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) { if (topic == null) throw new IllegalArgumentException("Topic cannot be null"); if (timestamp != null && timestamp < 0) throw new IllegalArgumentException("Invalid timestamp " + timestamp); this.topic = topic; this.partition = partition; this.key = key; this.value = value; this.timestamp = timestamp;
} //计算 patition，如果指定了 patition 则直接使用，否则使用 key 计算 private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
Integer partition = record.partition(); if (partition != null) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic()); int lastPartition = partitions.size() - 1; if (partition < 0 || partition > lastPartition) { throw new IllegalArgumentException(String.format("Invalid partition given with record: %d is not in the range [0...%d].", partition, lastPartition));
} return partition;
} return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
} // 使用 key 选取 patition public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement();
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition();
} else { return DefaultPartitioner.toPositive(nextValue) % numPartitions;
}
} else { //对 keyBytes 进行 hash 选出一个 patition return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}producer 写入消息序列图如下所示：
图.3
流程说明：
1. producer 先从 zookeeper 的 "/brokers/.../state" 节点找到该 partition 的 leader 2. producer 将消息发送给该 leader 3. leader 将消息写入本地 log 4. followers 从 leader pull 消息，写入本地 log 后 leader 发送 ACK 5. leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset） 并向 producer 发送 ACK
愿意了解更多技术分享的可关注：mingli.com
朋友需要请加球球：二零四二八四九二三七
一般情况下存在三种情况：
1. At most once 消息可能会丢，但绝不会重复传输 2. At least one 消息绝不会丢，但可能会重复传输 3. Exactly once 每条消息肯定会被传输一次且仅传输一次
当 producer 向 broker 发送消息时，一旦这条消息被 commit，由于 replication 的存在，它就不会丢。但是如果 producer 发送数据给 broker 后，遇到网络问题而造成通信中断，那 Producer 就无法判断该条消息是否已经 commit。虽然 Kafka 无法确定网络故障期间发生了什么，但是 producer 可以生成一种类似于主键的东西，发生故障时幂等性的重试多次，这样就做到了 Exactly once，但目前还并未实现。所以目前默认情况下一条消息从 producer 到 broker 是确保了 At least once，可通过设置 producer 异步发送实现At most once。
标签：spring mvc+my batis kafka dubbo+zookeerper restful redis分布式缓存
原文地址：http://12945747.blog.51cto.com/12935747/1931370
