服务端技术实战系列——Kafka篇

Posted 2023-02-14

参考技术A 一．概念&原理

[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 = 0; i < consumerNumber; i++)

/**

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

 */

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







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