下载掌阅APP，畅读海量书库
立即打开
本节从架构设计的难点入手，提供突破难点的设计思路，给出架构设计难点的解决方案，并在后续的内容中对其具体的实现进行详细讲解。
本项目针对不同的场景，提供了同步线程模型和异步线程模型。
在这种线程模型中，客户端为每个消费者流使用一个线程，每个线程负责从Kafka队列里消费消息，并且在同一个线程里处理业务。我们把这些线程称为消费线程，并把这些线程所在的线程池叫作消息消费线程池。这种模型多用于处理轻量级别的业务，例如缓存查询、本地计算等。
在这种线程模型中，客户端为每个消费者流使用一个线程，每个线程负责从Kafka队列里消费消息，并且传递消费得到的消息到后端的异步线程池中，在异步线程池中处理业务。我们仍然把在前面负责消费消息的线程池称为消息消费线程池，把后面的异步线程池称为异步业务线程池。这种线程模型适合重量级的业务，例如在业务中有大量的 I/O 操作、网络 I/O 操作、复杂计算、对外部系统的调用等。
后端的异步业务线程池又细分为所有消费者流共享线程池和每个流独享线程池。
所有消费者流共享线程池与每个流独享线程池相比，会创建更少的线程池对象，能节省些许内存，但是，由于多个流共享同一个线程池，所以在数据量较大时，流之间的处理可能会互相影响。例如，一个业务使用了两个区和两个流，它们一一对应，通过生产者指定定制化的散列函数替换默认的 key-hash，实现了一个流（区）用来处理普通用户，另外一个流（区）用来处理VIP用户。如果两个流共享一个线程池，则当普通用户的消息大量产生时，VIP用户数量很少，并且排在了队列的后面，会产生饿死的情况。对于该场景可以使用多个topic来解决，一个是普通用户的topic，一个是VIP用户的topic，但是这样又要多维护一个topic，客户端发送时需要显式地判断topic的目标，也没有多少好处。
每个流独享线程池会使用不同的异步业务线程池来处理不同的流里面的消息，互相隔离、互相独立、不互相影响。在不同的流（区）的优先级不同或者消息在不同流（区）中不均衡的情况下表现会更好，当然，创建多个线程池会多使用些许内存，但这并不是一个大问题。
另外，异步业务线程池支持有确定的线程数量的线程池和线程数量可自动增减的线程池。
（1）若核心业务的硬件资源有保证，核心服务有专享的资源池，或者线上流量可预测，则请使用固定数量的线程池。
（2）非核心业务一般混合部署，资源互相调配，若存在线上流量不固定等情况，则请使用线程数量可自动增减的线程池。
对于在消息处理过程中产生的业务异常，当前在业务处理的上层捕捉了Throwable，在专用的错误恢复日志中记录了出错的消息，后续可根据错误恢复日志人工处理错误消息，也可重做或者清洗数据。这里也考虑实现异常的 Listener 体系结构，对异常处理采用监听者模式来实现异常处理器的可插拔等。当前采用简单易用的方式，默认仅打印错误日志。
在默认的异常处理逻辑中，在捕捉异常后会在专用的错误恢复日志中记录错误信息，然后继续处理下一条消息。考虑到可能存在上线失败或者上游消息格式出错等场景，会导致所有消息处理都出错，以至于堆满错误恢复日志，因此，我们需要借助报警和监控系统来解决这种大量异常导致的问题。
消费者本身是一个事件驱动的服务器，类似于 Tomcat：Tomcat接收HTTP请求并返回HTTP 响应，Consumer 则接收 Kafka 的消息，在处理业务后返回，也可以将处理结果发送到下一个消息队列。所以，消费者本身是非常复杂的，除了线程模型，异常处理、性能、稳定性、可用性等都需要我们考虑。既然消费者是一个后台的服务器，所以我们需要考虑如何优雅地关机，也就是说需要考虑在消费者服务器处理消息时如何关机，才不会因为处理中断而丢失消息。
优雅关机的重点在于：
● 如何知道JVM要退出？
● 如何阻止Daemon的线程在JVM退出后被杀掉？
● 如果Worker线程处于阻塞状态，则如何唤醒并退出？
对于第1个问题，如果一个后台程序运行在控制台的前台中，则可以通过Ctrl+C发送退出信号给JVM，也可以通过kill-2 PS_ID 或者 kill-15 PS_ID发送退出信号，但是不能发送kill-9 PS_ID，否则进程会无条件地强制退出。JVM在收到退出信号后，会调用注册的钩子，我们通过注册JVM退出钩子进行优雅关机。
对于第2个问题，线程分为Daemon线程和非Daemon线程，一个线程默认继承父线程的Daemon属性，如果当前线程池是由Daemon线程创建的，则Worker线程是Daemon线程。如果Worker线程是Daemon线程，则我们需要在JVM退出钩子时等待Worker线程完成当前手头处理的消息，再退出JVM。如果不是Daemon线程，则即使JVM收到退出信号，也得等待Worker线程退出后再退出，不会丢掉正在处理的消息。
对于第3个问题，在Worker线程从Kafka服务器消费消息时，Worker线程可能处于阻塞状态，这时需要中断线程以退出，在这种场景下没有消息丢失的情况。在Worker线程处理业务时有可能有阻塞，例如I/O、网络I/O，在指定的退出时间内没有完成，我们也需要中断线程以退出，这时会产生一个 InterruptedException，在异常处理的默认处理器中被捕捉并写入错误日志，Worker线程随后退出。