前两篇中，学习了consumer端的consumer构建，topic订阅，消息轮询，位移管理，重平衡等知识点部分的学习。剩下的还有解序列化，多线程，独立consumer,新老consumer对比部分，接下来让我们开始吧。
Consumer端的解序列化是与producer端的序列化相对应的，consumer端从broker端获取的消息格式是字节数组，需要通过解序列化器将其转换为指定的对象类型。
kafka提供了以下deserializer：
ByteArrayDeserializer:无须操作，因为已经是字节数组。
ByteBufferDeserializer:将消息解序列化成ByteBuffer。
BytesDeserializer:将消息解序列化为kafka自定义的Bytes类。
DoubleDeserializer:解序列化Double类型。
IntegerDeserializer:解序列化Integer类型。
LongDeserializer:解序列化Long类型。
StringDeserializer:解序列化String类型。
Consumer端通过设置key.deserializer和value.deserializer参数来指定解序列化类型。如果涉及复杂的类型，用户可以自定义deserializer。
自定义deserializer整体上需要以下三步:
1).自定义或复用serializer的数据对象格式。
2).创建自定义deserializer类，实现org.apache.kafka.common.serialization.Deserializer接口，在deserializer方法中实现deserialize逻辑。
3).在构造KafkaConsumer的Properties对象中设置key.deserializer和value.deserializer的值。
kafkaProducer是线程安全的，用户可以在多个线程中使用同一个KafkaProducer实例，但是Consumer是非线程安全的，如果consumer想实现多线程消费，有以下两种方法:
1).多KafkaConsumer线程:每个线程维护一个KafkaConsumer，用户创建多个线程消费topic数据，每个线程创建自己的KafkaConsumer实例。每个线程负责消费固定数目的分区。实际中一般包含三个相关类:
ConsumerRunnable类:消费线程类，执行真正的消费任务。
ConsumerGroup类:消费线程管理类，创建多个线程类执行消费任务。
ConsumerMain类:主方法类。
2).单KafkaConsumer实例+多worker线程:实现思想为将消息的获取与消息的处理过程分开处理，实现解耦，全局维护一个或若干个consumer实例执行消息获取任务，然后将消息处理交由单独的worker线程进行处理,worker线程完成处理之后上报位移状态，由全局consumer提交位移。实际中一般也包含以下三个相关类:
ConsumerThreadHandler类:consumer多线程管理类，用于创建线程池及为每个线程分配消息集合，同时还用于consumer位移提交。
ConsumerWorker类:执行真正的消费逻辑并将消费的位移状态上报给ConsumerThreadHandler。
ConsumerMain类:主方法类。
这两种方式各自有自己的优缺点，具体如下:
多KafkaConsumer线程:实现简单，速度较快，没有线程间交互开销，便于位移管理，方便维护分区键的消息消费顺序。但是Socket连接开销大，consumer数量受限制于topic分区数，扩展性差，broker负载高，容易导致rebalance。
单KafkaConsumer实例+多worker线程:消息获取和消息处理过程分离，便于扩展consumer和worker数。但是难以维护分区内的消息顺序，处理链路较长，位移管理复杂，如果worker线程异常还可能导致消费数据丢失。
对于由多个consumer共同读取某个topic的场景来说，使用group是非常好的选择，但是对于需要精准控制消费的需求场景，则consumer group便失效了，此时需要使用独立consumer,即standalone consumer，standalone consumer之间是彼此独立的，任意一个consumer发生的变化对其他consumer不会造成任何影响。
使用standalone consumer是通过调用KafkaConsumer.assign方法实现的，与consumer group中的subscribe方法类似，当发生多次assign调用时，之前的会被覆盖，只有最后一个assign的调用会生效，另外，在同一个consumer应用中不要混合使用standalone consumer和consumer group。
在kafka中，新旧版本存在以下差别:
1).旧版本consumer是是使用scala编写，新版本是使用Java编写。新版本使用的API包名为org.apache.kafka.clients.consumer.*,主要使用类为KafkaConsumer。而旧版本使用的API包名为kafka.consumer.*,主要使用的类是ZookeeperConsumerConnector,SimpleConsumer。
2).旧版本consumer group的组成员管理，错误检测等功能，尤其是位移的提交依赖于ZooKeeper完成，依赖Zookeeper下的znode节点写入位移信息完成位移提交动作，而新版本则是kafka自身维护了一个内部topic用来代替Zookeeper完成上述功能。
3).在旧版本中，与新版本中的consumer group和standalone consumer相对应的是high-level consumer和low-level consumer。其中，high-level consumer使用的是consumer group，low-level consumer使用的是standalone consumer。
high-level。
4).新版本中，Consumer使用单线程轮询的方式来消费消息，而在旧版本中，Consumer使用的是多线程方式，用户可以指定多个线程来消费订阅的topic,consumer内部会自动创建多个线程进行消费，而不是新版本中需要手动创建多个线程。
另外，如果用户逻辑简单，不关心位移提交，只是单纯想消费数据，则优先建议使用high-level consumer+自动提交位移方法，反之，使用low-level consumer方式，该方式优点为:
可以细粒度控制分区消费，适用于以下场景:
消息需要重复读取。
只需消费部分分区的数据。
实现"精确一次"处理语义。
缺点是:
用户需要自己处理为已提交。
用户需要寻找分区的leader broker。
用户需要自己处理leader变更。
旧版本Kafka中consumer group依赖Zookeeper完成组成员管理，错误检测等功能，Consumer会在/consumers/groupId/ids/Consumer id目录下生成一个临时节点，即ephemeral node,如果Kafka和ZooKeeper的会话过期，该节点会自动被删除，可以通过这个来监听consumer的存活状态，一旦该节点失效，可以认为该consumer已经异常，会开启新一轮rebalance。
实现"精确一次"的核心思想是:将数据处理和位移提交放入一个事务中，在事务中，可以实现任何一个操作都可以回滚整个操作，从而实现精确一次的消息消费。
通常在kafka中，consumer设置线程数的个数不大于订阅的分区总数，一方面，因为如果创建超过分区数的线程会有线程无法消费任何分区。另一方面，创建多余的线程会造成额外的资源浪费。
本篇中，继前两篇后总结了consumer的解序列化，多线程消费，独立consumer,及新老consumer对比这几个方面的知识点，到此consumer的基本学习告一段落。
END
原文作者：DBA修行之路；如有侵权，请联系管理员删除。
