一．概念&原理
[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();
}
}
}
先逐个初始化消费者实例，然后将这些消费者加入到消费组列表中。消费组启动后，会循环产生消费者线程。
