在前一篇文章爬虫架构|利用Kafka处理数据推送问题（1）中对Kafka做了一个介绍，以及环境搭建，最后是选择使用阿里云的Kafka，这一篇文章继续说使用阿里云的Kafka的一些知识。
一、发布者最佳实践
发布的完整代码（根据自己的业务做相应处理）：
代码语言：javascript复制package com.yimian.controller.kafka;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Future;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import com.alibaba.fastjson.JSON;
import com.yimian.model.SpiderData;
/**
* 生产者
*
* @author huangtao
*
*/
@Controller
@RequestMapping(value = "kafka/producer")
public class KafkaProducerController {
private static Producer<String, String> producer;
private static Properties kafkaProperties;
static {
// 设置sasl文件的路径
JavaKafkaConfigurer.configureSasl();
// 加载kafka.properties
kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
// 设置接入点，请通过控制台获取对应Topic的接入点
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
// 设置SSL根证书的路径，请记得将XXX修改为自己的路径
// 与sasl路径类似，该文件也不能被打包到jar中
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
// 根证书store的密码，保持不变
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
// 接入协议，目前支持使用SASL_SSL协议接入
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
// SASL鉴权方式，保持不变
props.put(SaslConfigs.SASL_MECHANISM, "ONS");
// Kafka消息的序列化方式
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");
// 请求的最长等待时间
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
// 构造Producer对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可；
// 如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过5个
producer = new KafkaProducer<String, String>(props);
}
/**
* 发送消息给kafka
* @param topic
* @param msg
*/
public static void sendMsgToKafka(String topic, SpiderData msg) {
try {
// 发送消息，并获得一个Future对象
Future<RecordMetadata> metadataFuture = producer.send(new ProducerRecord<String, String>(topic, String.valueOf(new Date().getTime()),
JSON.toJSONString(msg)));
// 同步获得Future对象的结果
RecordMetadata recordMetadata = metadataFuture.get();
System.out.println("Produce ok:" recordMetadata.toString());
} catch (Exception e) {
/**
* 要考虑重试～
* 在分布式环境下，由于网络等原因，偶尔的发送失败是常见的。这种失败有可能是消息已经发送成功，但是 Ack 失败，也有可能是确实没发送成功。
* 消息队列 Kafka 是 VIP 网络架构，会主动掐掉空闲连接（一般 30 秒没活动），也就是说，不是一直活跃的客户端会经常收到”connection rest by peer”这样的错误，因此建议都考虑重试。
*/
// 参考常见报错:
// https://help.aliyun.com/document_detail/68168.html?spm=a2c4g.11186623.6.567.2OMgCB
System.out.println("error occurred");
e.printStackTrace();
}
}
@RequestMapping(value = "init", produces = "text/html;charset=UTF-8")
@ResponseBody
public void init() {
// 构造一个Kafka消息
String topic = kafkaProperties.getProperty("topic"); // 消息所属的Topic，请在控制台申请之后，填写在这里
SpiderData data = new SpiderData();
data.setDescUrl("www.baidu.com");
data.setTitle("百度");
sendMsgToKafka(topic, data);
}
}Kafka的发送非常简单，代码片段如下：
代码语言：javascript复制Future<RecordMetadata> metadataFuture = producer.send(new ProducerRecord<String, String>(
topic, \ topic
null, \ 分区编号，这里最好为 null，交给 producer 去分配
System.currentTimeMillis(), \时间戳
String.valueOf(message.hashCode()), \ key，可以在控制台通过这个 Key 查找消息，这个 key 最好唯一；
message)); \ value，消息内容message可以是一个JSON类型的对象，如上面例子中的JSON.toJSONString(new SpiderData())
1.1、Key 和 Value
Kafka 0.10.0.0 的消息字段只有两个：Key 和 Value。为了便于追踪，重要消息最好都设置一个唯一的 Key。通过 Key 追踪某消息，打印发送日志和消费日志，了解该消息的发送和消费情况；更重要的是，您可以在控制台可以根据 Key 查询消息的内容。
1.2、失败重试
在分布式环境下，由于网络等原因，偶尔的发送失败是常见的。这种失败有可能是消息已经发送成功，但是 Ack 失败，也有可能是确实没发送成功。
消息队列 Kafka 是 VIP 网络架构，会主动掐掉空闲连接（一般 30 秒没活动），也就是说，不是一直活跃的客户端会经常收到”connection rest by peer”这样的错误，因此建议都考虑重试。
1.3、异步发送
需要注意的是这个接口是异步发送的；如果你想得到发送的结果，可以调用metadataFuture.get(timeout, TimeUnit.MILLISECONDS)。
1.4、线程安全
Producer 是线程安全的，且可以往任何 Topic 发送消息。一般一个应用，对应一个 Producer 就足够了。
1.5、Ack
消息队列 Kafka 没有考虑这个参数，都认为是“all”，即所有消息同步到 Slave 节点后才会返回成功的确认消息给客户端。
1.6、Batch
Batch 的基本思路是：把消息缓存在内存中，并进行打包发送。Kafka 通过 Batch 来提高吞吐，但同时也会增加延迟，生产时应该对两者予以权衡。
在构建 Producer 时，需要考虑以下两个参数：
batch.size: 发往每个 Partition 的消息个数缓存量达到这个数值时，就会触发一次网络请求，把消息真正发往服务器；linger.ms: 每个消息待在缓存中的最大时间，超过这个时间，就会忽略batch.size的限制，立即把消息发往服务器。
由此可见，Kafka 什么时候把消息真正发往服务器，是通过上面两个参数共同决定的；
batch.size 有助于提高吞吐，linger.ms 有助于控制延迟。您可以根据具体业务进行调整。
1.7、OOM
结合 Kafka Batch 的设计思路，Kafka 会缓存消息并打包发送，如果缓存太多，则有可能造成 OOM。
buffer.memory: 所有缓存消息的总体大小超过这个数值后，就会触发把消息发往服务器。此时会忽略batch.size和linger.ms的限制。buffer.memory的默认数值是 32M，对于单个 Producer 来说，可以保证足够的性能。需要注意的是，如果你在同一个 JVM 中启动多个 Producer，那么每个 Producer 都有可能占用32M 缓存空间，此时便有可能触发 OOM。- 在生产时，一般没有必要启动多个 Producer；如果特殊情况需要，则需要考虑
buffer.memory的大小，避免触发 OOM。
1.8、分区顺序
单个分区内，消息是按照发送顺序储存的，是基本有序的。
但消息队列 Kafka 并不保证单个分区内绝对有序，所以在某些情况下，会发生少量消息乱序。比如：消息队列 Kafka 为了提高可用性，某个分区挂掉后把消息 Failover 到其它分区。
二、订阅者最佳实践
消费的完整代码（根据自己的业务做相应处理）：
代码语言：javascript复制package com.yimian.controller.kafka;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yimian.model.SpiderData;
/**
* 消费者
*
* @author huangtao
*
*/
@Controller
@RequestMapping(value = "kafka/consumer")
public class KafkaConsumerController {
private static Consumer<String, String> consumer;
static {
// 设置sasl文件的路径
JavaKafkaConfigurer.configureSasl();
// 加载kafka.properties
Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
// 设置接入点，请通过控制台获取对应Topic的接入点
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
// 设置SSL根证书的路径，请记得将XXX修改为自己的路径
// 与sasl路径类似，该文件也不能被打包到jar中
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
// 根证书store的密码，保持不变
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
// 接入协议，目前支持使用SASL_SSL协议接入
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
// SASL鉴权方式，保持不变
props.put(SaslConfigs.SASL_MECHANISM, "ONS");
// 两次poll之间的最大允许间隔
// 请不要改得太大，服务器会掐掉空闲连接，不要超过30000
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 25000);
// 每次poll的最大数量
// 注意该值不要改得太大，如果poll太多数据，而不能在下次poll之前消费完，则会触发一次负载均衡，产生卡顿
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
// 消息的反序列化方式
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
// 当前消费实例所属的消费组，请在控制台申请之后填写
// 属于同一个组的消费实例，会负载消费消息
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
// 构造消息对象，也即生成一个消费实例
consumer = new KafkaConsumer<String, String>(props);
// 设置消费组订阅的Topic，可以订阅多个
// 如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样
List<String> subscribedTopics = new ArrayList<String>();
// 如果需要订阅多个Topic，则在这里add进去即可
// 每个Topic需要先在控制台进行创建
subscribedTopics.add(kafkaProperties.getProperty("topic"));
consumer.subscribe(subscribedTopics);
}
@RequestMapping(value = "init", produces = "text/html;charset=UTF-8")
@ResponseBody
public void init() {
// 循环消费消息
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(1000);
// 必须在下次poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG
// 建议开一个单独的线程池来消费消息，然后异步返回结果
for (ConsumerRecord<String, String> record : records) {
JSONObject jsonMsg = JSON.parseObject(record.value());
SpiderData spiderData = JSONObject.toJavaObject(jsonMsg, SpiderData.class);
System.out.println(spiderData.toString());
}
} catch (Exception e) {
try {
Thread.sleep(1000);
} catch (Throwable ignore) {
}
// 参考常见报错:
// https://help.aliyun.com/document_detail/68168.html?spm=a2c4g.11186623.6.567.2OMgCB
e.printStackTrace();
}
}
}
}消费时把JSON数据反序列化：
代码语言：javascript复制for (ConsumerRecord<String, String> record : records) {
JSONObject jsonMsg = JSON.parseObject(record.value());
SpiderData spiderData = JSONObject.toJavaObject(jsonMsg, SpiderData.class);
}2.1、消费消息基本流程
Kafka 订阅者在订阅消息时的基本流程是：
- Poll 数据
- 执行消费逻辑
- 再次 poll 数据
2.2、负载消费
每个 Consumer Group 可以包含多个消费实例，也即可以启动多个 Kafka Consumer，并把参数 group.id 设置成相同的值。属于同一个 Consumer Group 的消费实例会负载消费订阅的 topic。
示例1：Consumer Group A 订阅了 Topic A，并开启三个消费实例 C1、C2、C3，则发送到 Topic A 的每条消息最终只会传给 C1、C2、C3 的某一个。Kafka 默认会均匀地把消息传给各个消息实例，以做到消费负载均衡。
Kafka 负载消费的内部原理是，把订阅的 Topic 的分区，平均分配给各个消费实例。因此，消费实例的个数不要大于分区的数量，否则会有实例分配不到任何分区而处于空跑状态。这个负载均衡发生的时间，除了第一次启动上线之外，后续消费实例发生重启、增加、减少等变更时，都会触发一次负载均衡。
消息队列 Kafka 分区的数量至少是 16 个，已经足够满足大部分用户的需求，且云上服务会根据容量调整分区数。
2.3、多个订阅
一个 Consumer Group 可以订阅多个 Topic。一个 Topic 也可以被多个 Consumer Group 订阅，且各个 Consumer Group 独立消费 Topic 下的所有消息。
示例1：Consumer Group A 订阅了 Topic A，Consumer Group B 也订阅了 Topic A，则发送到 Topic A 的每条消息，不仅会传一份给 Consumer Group A 的消费实例，也会传一份给 Consumer Group B 的消费实例，且这两个过程相互独立，相互没有任何影响。
2.4、消费位点
每个 Topic 会有多个分区，每个分区会统计当前消息的总条数，这个称为最大位点 MaxOffset。Kafka Consumer 会按顺序依次消费分区内的每条消息，记录已经消费了的消息条数，称为ConsumerOffset。
剩余的未消费的条数（也称为消息堆积量） = MaxOffset - ConsumerOffset
2.5、位点提交
Kafka 消费者有两个相关参数：
enable.auto.commit：默认值为 true。auto.commit.interval.ms： 默认值为 1000，也即 1s。
这两个参数组合的结果就是，每次 poll 时，再拉取数据前会预先做下面这件事：
- 检查上次提交位点的时间，如果距离当前时间已经超过 auto.commit.interval.ms，则启动位点提交动作；
因此，如果 enable.auto.commit 设置为 true，需要在每次 poll 时，确保前一次 poll 出来的数据已经消费完毕，否则可能导致位点跳跃；
如果想自己控制位点提交，则把 enable.auto.commit 设为 false，并调用 commit(offsets)函数自行控制位点提交。
2.6、消息重复以及消费幂等
Kafka 消费的语义是 “At Lease Once”， 也就是至少投递一次，保证消息不丢，但是不会保证消息不重复。在出现网络问题、客户端重启时均有可能出现少量重复消息，此时应用消费端，如果对消息重复比较敏感(比如说订单交易类)，则应该做到消息幂等。
以数据库类应用为例，常用做法是：
- 发送消息时，传入 key 作为唯一流水号ID；
- 消费消息时，判断 key 是否已经消费过，如果已经消费过了，则忽略，如果没消费过，则消费一次；
当然，如果应用本身对少量消息重复不敏感，则不需要做此类幂等检查。
2.7、消费失败
Kafka 是按分区一条一条消息顺序向前消费推进的，如果消费端拿到某条消息后消费逻辑失败，比如应用服务器出现了脏数据，导致某条消息处理失败，等待人工干预，该怎么办呢？
- 如果失败后一直尝试再次执行消费逻辑，则有可能造成消费线程阻塞在当前消息，无法向前推进，造成消息堆积；
- 由于 Kafka 自身没有处理失败消息的设计，实践中通常会打印失败的消息、或者存储到某个服务（比如创建一个 Topic 专门用来放失败的消息），然后定时 check 失败消息的情况，分析失败原因，根据情况处理。
2.8、消费阻塞以及堆积
消费端最常见的问题就是消费堆积，最常造成堆积的原因是：
- 消费速度跟不上生产速度，此时应该提高消费速度，详情见本文下一节<提高消费速度>；
- 消费端产生了阻塞；
消费端拿到消息后，执行消费逻辑，通常会执行一些远程调用，如果这个时候同步等待结果，则有可能造成一直等待，消费进程无法向前推进。
消费端应该竭力避免堵塞消费线程，如果存在等待调用结果的情况，设置等待的超时时间，超过时间后，作消费失败处理。
2.9、提高消费速度
提高消费速度有两个办法：
- 增加 Consumer 实例个数；
- 增加消费线程；
增加 Consumer 实例，可以在进程内直接增加（需要保证每个实例一个线程，否则没有太大意义），也可以部署多个消费实例进程；需要注意的是，实例个数超过分区数量后就不再能提高速度，将会有消费实例不工作；
增加 Consumer 实例本质上也是增加线程的方式来提升速度，因此更加重要的性能提升方式是增加消费线程，最基本的方式如下：
- 定义一个线程池；
- poll 数据；
- 把数据提交到线程池进行并发处理；
- 等并发结果返回成功再次poll数据执行。
2.10消息过滤
Kafka 自身没有消息过滤的语义。实践中可以采取以下两个办法：
- 如果过滤的种类不多，可以采取多个 Topic 的方式达到过滤的目的；
- 如果过滤的种类多，则最好在客户端业务层面自行过滤。
实践中根据业务具体情况进行选择，可以综合运用上面两种办法。
2.11、消息广播
Kafka 自身没有消息广播的语义，可以通过创建不同的 Consumer Group来模拟实现。
2.12、订阅关系
同一个 Consumer Group 内，各个消费实例订阅的 Topic 最好保持一致，避免给排查问题带来干扰。
参考资料：阿里云消息队列Kafka的帮助文档
