程序中的Kafka中服务器有哪些接点呢？

有以下几个，希望能帮到你

第一：broker

Kafka 集群包含一个或多个服务器，服务器节点称为broker。

broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。

如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。

如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

第二：Topic

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

类似于数据库的表名

第三：Partition

topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，不同partition间的数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。

第四：Producer

生产者即数据的发布者，该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。

第五：Consumer

消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。

第六：Consumer Group

每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

第七：Leader

每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。

第八：Follower

Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。

以上是kf的主要使用接口，希望能帮助你，谢谢！

一．概念&原理

[if !supportLists]1. [endif]主题(topic)：主题是对消息的分类。

[if !supportLists]2. [endif]消息(message)：消息是kafka通信的基本单位。

[if !supportLists]3. [endif]分区(partition)： 一组 消息对应 一个 主题， 一个 主题对应 一个或多个 分区。每个分区为一系列有序消息组成的 有序队列 ；每个分区在物理上对应一个文件夹。

[if !supportLists]4. [endif]副本(replica)：每个分区有 一个或多个 副本，分区的副本分布在集群的 不同 代理（机器）上，以提高可用性；分区的副本与日志对象是一一对应的。

[if !supportLists]5. [endif]Kafka只保证一个 分区内 的消息 有序性 ，不保证跨分区消息的有序性。消息被追加到相应分区中， 顺序写入磁盘 ，效率非常高。

[if !supportLists]6. [endif]Kafka选取某个某个分区的 一个 副本作为leader副本，该分区的 其他 副本为follower副本。 只有leader副本负责处理客户端读/写请求 ，follower副本从leader副本同步数据。

[if !supportLists]7. [endif]任何发布到分区的消息都会追加到日志文件的尾部， 每条消息 在日志文件中的 位置 都对应一个 按序递增的偏移量 ；偏移量在一个分区下严格有序。

[if !supportLists]8. [endif]Kafka不允许对消息进行随机读写。

[if !supportLists]9. [endif]新版消费者将 消费偏移量 保存到kafka内部的一个主题中。

[if !supportLists]10. [endif]Kafka集群由 一个或多个代理 （Broker，也称为kafka实例）构成。可以在 一台 服务器上配置 一个或多个代理 ，每个代理具有唯一标识broker.id。

[if !supportLists]11. [endif]生产者将消息 发送给代理 （Broker）。

[if !supportLists]12. [endif]消费者以 拉取 （pull）方式拉取数据，每个消费者都属于一个消费组。

[if !supportLists]13. [endif]同一个主题的一条消息只能被 同一个消费组 下的某一个消费者消费，但 不同消费组 的消费者可以 同时 消费该消息。

[if !supportLists]14. [endif]消息 广播 ：指定各消费者属于不同消费组；消息 单播 ：指定各消费者属于同一个消费组。

[if !supportLists]15. [endif]Kafka启动时在Zookeeper上创建相应节点来保存 元数据 ，元数据包括：代理节点信息、集群信息、主题信息、分区状态信息、分区副本分配方案、动态配置等；

[if !supportLists]16. [endif]Kafka通过 监听 机制在节点注册监听器来监听节点元数据变化；

[if !supportLists]17. [endif]Kafka将数据写入 磁盘 ，以文件系统来存数据；

[if !supportLists]18. [endif]生产环境一般将zookeeper集群和kafka集群 分机架 部署；

[if !supportLists]二．[endif] Kafka Producer

配置：

/**

 * xTestProxy——KafkaConfigConstant

 *

 * @author  ZhangChi

 * @date  2018年6月20日---下午5:50:44

 * @version  1.0

 */

public   class  KafkaConfigConstant {

public   static   final  String KAFKA_CLUSTER  = "fa-common1.hangzhou-1.kafka.internal.lede.com:9200,fa-common2.hangzhou-1.kafka.internal.lede.com:9200,fa-common3.hangzhou-1.kafka.internal.lede.com:9200"

}

生产者配置：

/**

 * xTestProxy——HttpKafkaProducerFactory

 *

 * @author  ZhangChi

 * @date  2018年6月11日---下午2:37:51

 * @version  1.0

 */

public   class  HttpKafkaProducerFactory {

// 真正的KafkaProducer仅有一份

private   static  KafkaProducer kafkaProducer  = null

private   static  Properties property

public   static  KafkaProducer getKafkaProducer() {

if  ( kafkaProducer  == null ) {

synchronized  (HttpKafkaProducerFactory. class ) {

if  ( kafkaProducer  == null ) {

property  = buildKafkaProperty ()

kafkaProducer  = new  KafkaProducer( property )

}

}

}

return   kafkaProducer

}

public   static  Properties buildKafkaProperty() {

Properties props = new  Properties()

props.put(ProducerConfig. BOOTSTRAP_SERVERS_CONFIG , KafkaConfigConstant. KAFKA_CLUSTER )

props.put(ProducerConfig. ACKS_CONFIG , "all")

props.put(ProducerConfig. RETRIES_CONFIG , 0)

props.put(ProducerConfig. BATCH_SIZE_CONFIG , 16384)

props.put(ProducerConfig. BUFFER_MEMORY_CONFIG , 33554432)

props.put(ProducerConfig. LINGER_MS_CONFIG , 1)

props.put(ProducerConfig. KEY_SERIALIZER_CLASS_CONFIG , "org.apache.kafka.common.serialization.StringSerializer")

props.put(ProducerConfig. VALUE_SERIALIZER_CLASS_CONFIG ,

"org.apache.kafka.common.serialization.StringSerializer")

return  props

}

}

生产者线程组：

/**

 * xTestProxy——HttpKafkaProducerThread

 * 多线程每次new一个实例

 *

 * @author  ZhangChi

 * @date  2018年6月25日---下午2:09:39

 * @version  1.0

 */

public   class  HttpKafkaProducerThread implements  Runnable {

private   static  Logger logger  = LoggerFactory. getLogger ("HttpKafkaProducerThread")

private   final  String KAFKA_TOPIC = KafkaConstant. HTTP_REQ_RESP_TOPIC

private  String kafkaMessageJson

private  KafkaProducer producer

public  String messageType

public  String originalMessage

private   static  KafkaMessage kafkaMessage  = new  KafkaMessage()

public  HttpKafkaProducerThread(KafkaProducer producer, String messageType, String originalMessage) {

this .producer = producer

this .messageType = messageType

this .originalMessage = originalMessage

}

@Override

public   void  run() {

// TODO  Auto-generated method stub

/* 1.构建kafka消息*/

kafkaMessageJson = generateKafkaMessage( this .messageType, this .originalMessage)

/* 2.发送kafka消息*/

if  (kafkaMessageJson != null  &&!StringUtils. isEmpty (kafkaMessageJson)) {

logger .info("create message start:" + kafkaMessageJson)

producer.send( new  ProducerRecord( this .KAFKA_TOPIC, kafkaMessageJson))

} else  {

logger .info("kafkaMessageJson is null!")

}

}

private  String generateKafkaMessage(String messageType, String originalMessage) {

if  (StringUtils. isBlank (messageType) || StringUtils. isBlank (originalMessage)) {

return   null

}

kafkaMessage .setMessageId(KafkaMessageUtils. generateId ())

kafkaMessage .setMessageTime(KafkaMessageUtils. generateTime ())

kafkaMessage .setMessageType(messageType)

kafkaMessage .setMessage(originalMessage)

String kafkaMessageToJson = null

try  {

kafkaMessageToJson = KafkaMessageUtils. objectToJson ( kafkaMessage )

} catch  (JsonProcessingException e) {

// TODO  Auto-generated catch block

e.printStackTrace()

}

kafkaMessageJson = kafkaMessageToJson

return  kafkaMessageToJson

}

}

[if !supportLists]三．[endif] Kafka Consumer

消费者配置：

private   static  Properties buildKafkaProperty() {

Properties properties = new  Properties()

// 测试环境kafka的端口号是9200

properties.put(ConsumerConfig. BOOTSTRAP_SERVERS_CONFIG , KafkaConfigConstant. KAFKA_CLUSTER )

// 消费组名称

properties.put(ConsumerConfig. GROUP_ID_CONFIG , KafkaConfigConstant. GROUP_ID )

properties.put(ConsumerConfig. CLIENT_ID_CONFIG , "test")

// 从头消费

properties.put(ConsumerConfig. AUTO_OFFSET_RESET_CONFIG , "earliest")

// 自动提交偏移量

properties.put(ConsumerConfig. ENABLE_AUTO_COMMIT_CONFIG , "true")

// 时间间隔1s

properties.put(ConsumerConfig. AUTO_COMMIT_INTERVAL_MS_CONFIG , "1000")

properties.put(ConsumerConfig. KEY_DESERIALIZER_CLASS_CONFIG ,

"org.apache.kafka.common.serialization.StringDeserializer")

properties.put(ConsumerConfig. VALUE_DESERIALIZER_CLASS_CONFIG ,

"org.apache.kafka.common.serialization.StringDeserializer")

return  properties

}

消费者线程组：

/**

 * AnalysisEngine——HttpKafkaConsumerGroup

 *

 * @author  ZhangChi

 * @date  2018年6月11日---下午6:20:47

 * @version  1.0

 */

@Service("httpKafkaConsumerGroup")

public   class  HttpKafkaConsumerGroup {

@Autowired

private  RequestAnalyzer requestAnalyzer

@Autowired

private  EsDocumentServiceImpl esDocumentServiceImpl

@Autowired

private  AnalysisEngineClient analysisEngineClient

@Autowired

private  MongoTemplate mongoTemplate

private  List httpKafkaConsumerList = new  ArrayList()

public   void  initHttpKafkaConsumerGroup( int  consumerNumber, RunModeEnum mode) {

for  ( int  i = 0i <consumerNumberi++) {

/**

 * 将注入的服务当做构造参数，这样保证每个子线程都能拿到服务实例而不是空指针！

 */

HttpKafkaConsumer consumerThread = new  HttpKafkaConsumer(requestAnalyzer, esDocumentServiceImpl, mode, analysisEngineClient, mongoTemplate)

httpKafkaConsumerList.add(consumerThread)

}

}

public   void  consumeGroupStart() {

for  (HttpKafkaConsumer item : httpKafkaConsumerList) {

LogConstant. runLog .info("httpKafkaConsumerList size : " + httpKafkaConsumerList.size())

Thread consumerThread = new  Thread(item)

consumerThread.start()

}

}

}

先逐个初始化消费者实例，然后将这些消费者加入到消费组列表中。消费组启动后，会循环产生消费者线程。

 

Kafka是一个分布式发布-订阅publish-subscribe消息传递系统，设计目标是快速、可伸缩和耐用冗余。

它是一个在一个较高的抽象水平上描述的但是又非常简单的系统，虽然当你更深地挖掘时会令人难以置信的技术细节。卡夫卡文档出色的解释了系统中许多设计和实现的微妙之处，所以我们不会在这里试图解释他们了。

像许多发布-订阅消息传递系统，卡夫卡能保存源消息的数据。生产者将输入写入到主题topic，而消费者则从主题topic中读取写入数据。因为卡夫卡是一个分布式系统，所以topic主题会实现跨多个节点分区和复制。

消息是简单byte数组，开发者能够使用它们存在对象的任何格式，如String或JSON和Avro，每个消息能够有一个Key，因此，生产者保证拥有相同key的消息到达同一个分区，而对主题topic的消费，可以使用多个消费者来配置消费组，每个消费组中消费者能够从他订阅的那个主题Topic所在分区子集中读取消息，这样每个消息发送给消费组中一个消费者，使用相同key的所有消息能够到达相同的消费者。

Kafka的特点是它将每个主题topic分区都看成一个日志log(一个有顺序的消息集合)，一个分区中的每个消息被分配一个唯一的偏移量offset.

Kafka并不试图跟踪哪个消息被哪个消费者读取，而是只是保留未被读取的消息。

Kafka保留所有消息的时间，而消费者负责跟踪它们在每个日志(日志是一个消息序列集合代表一个topic分区)中的位置，

因此Kafka能够支持大量的消费者，使用最小的代价保留大量的数据。

Kafka如何工作?

假设我们正在开发一个大型多人在线游戏。在这些游戏中,玩家在一个虚拟的世界中互相合作和竞争。通常玩家互相发生贸易，比如交换物品和金钱，所以游戏开发者重要的是要确保球员不会欺骗，在下面两种情况下的交易将标记为特殊：如果贸易数量明显大于正常的球员如果IP玩家登录是不同于过去的20场比赛使用的IP。除了对实时交易进行标记以外，我们也想加载数据到Apache

Hadoop，我们的数据,科学家们可以用它来训练和测试新算法。

基于游戏服务器内存数据缓存进行实时事件标记是最好的，能够让我们达到迅速决定，特别是对那些最活跃玩家。如果我们分区游戏服务器，我们的系统有多个游戏服务器和数据集，，包括过去登录的20个玩家和近20在内存的交易,。

我们的游戏服务器必须执行两个截然不同的角色：第一个是接受和执行用户操作，第二是实时处理贸易信息并标记可疑事件。为了有效执行第二个角色功能，每个用户整个贸易的历史事件都驻留在一个单独的服务器内存中。因为接收用户操作(第一个角色功能)的服务器可能没有他的贸易历史，这意味着我们必须通过服务器之间的消息实现第二个角色功能。为了让两个角色功能保持松散耦合,我们使用卡夫卡在服务器之间传递消息,您将看到如下：

卡夫卡有几个特性：

可伸缩性、数据分区,低延迟,并且能够处理大量不同的消费者。我们为登录和交易配置了一个Topic主题。我们把它们配置成一个topic主题的原因是：只有我们获得已经登录信息(我们可以确保玩家从他平时IP登录)后，才能确保他的交易是有效的。卡夫卡可以在在一个主题topic中维护这个前后顺序，而不是在两个topic之间。

当用户登录后进行交易，接受服务器立即发送事件到Kafka. 消息是将user id作为key, 事件作为值.

这能确保同一个用户的所有交易和登录发送到Kafka同样的分区. 每个事件处理服务器都运行一个Kafka消费者,

其每个消费者都被配置为同样组的一部分，这样，每个服务器从少量Kafka分区读取消息，所有关于某个特定用户的数据都能发往相同事件处理服务器，当事件处理服务器从Kafka读取一个用户交易时，将这个事件加入到用户事件历史缓存本地内存缓存，这样就无需额外网络磁盘开销直接标记那些可以事件。

重要的是我们为每个事件处理服务器创建一个分区，或者每个事件处理服务器上每核对应一个多线程应用，(记住 Kafka大部分是用少于 10,000

个分区实现所有主题topic，这样我们不能为每个用户创建一个分区，因为用户数是不断增加的。)

这好像是一种迂回方式处理事件：从游戏服务器发送事件到Kafka,

另外一台服务器读取这个事件再处理它，这种设计解耦了两种角色功能，允许我们为每个角色功能安排其需要的容量与能力。.另外，这样做不会增加处理时间，因为Kafka是设计为高吞吐量和低延迟的，即使只有一个小的三台节点服务器的集群环境，也能每秒处理接近一百万个事件，平均延迟只有3ms.

当服务器标识出一个事件作为可疑，它会发送这个标记了的事件到新的Kafka

topic，比如其主题名称为Alerts，这时报警服务器或仪表监控服务器会接受到这个事件，同时另外一个单独过程会从Alerts主题中读取这个事件，将它写入到Hadoop进行更进一步分析。

因为Kafka并不跟踪确认每个消费者的消息，它就能用很少的性能影响处理成千上万的消费者。Kafka甚至可以处理批量消费者：每一个小时处理过程唤醒激活处理一个队列中所有消息，根本就不会影响系统的吞吐量或延迟。

---