实战Kafka之异地双活--MirrorMaker1.0

实战Kafka之异地双活--MirrorMaker1.0,第1张

异地灾备可以防止一个机房出现灾难而导致整个系统无法提供服务的问题。

异地双活在异地灾备的基础上减少了数据的冗余,并且提高了对数据双向同步及同步实时性的要求。

Kafka作为在双活中扮演了很重要的角色,一方面是两个区域的正常业务的消息数据分发、另一方面则是elasticsearch等中间件集群的双活方案有时需要依赖kafka的双活。那么在保证实时性的前提下做好kafka的双活呢?下面博主抛砖引玉给出自己实现的基于Kafka自带组件Mirror-Maker的两种方案,读者可以根据自己需要进行改进。

MirrorMaker是Kafka官方提供的用来做跨机房同步的组件。在kafka的安装目录的bin目录下有一个kafka-mirror-maker.sh文件就是MirrorMaker的入口。

对于问题2的解决办法:

2、对于没有加密的kafka请使用xiao5aha/mirror-maker:v9版本

最近公司因为两次机房故障决定部署同城双机房,方案确定为双活单写

两个机房A.B都正常提供服务,所有写操作定位到A机房

// TODO 单写与多写的比较

// TODO 双活和冷备的比较

可以看到 两种实现方式 在B集群或专线崩溃时 受到的影响都较小

两个实现方式主要的区别是

第一种方法需要线下操作且需要保证zookeeper集群可用

第二种方法需要管理偏移量和mirrormaker

一.概念&原理

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

}

}

}

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

 


欢迎分享,转载请注明来源:夏雨云

原文地址:https://www.xiayuyun.com/zonghe/655790.html

(0)
打赏 微信扫一扫微信扫一扫 支付宝扫一扫支付宝扫一扫
上一篇 2023-07-23
下一篇2023-07-23

发表评论

登录后才能评论

评论列表(0条)

    保存