如何实现一个简单易用且可靠的消息队列框架？

功能特性

1. 简单易用

简化了Kafka客户端API的使用方法,特别是对消费端开发,消费端开发者只需要实现MessageHandler接口或者相关子类,在实现中处理消息完成业务逻辑,并且在主线程中启动封装的消费端服务器即可.它提供了各种常用的MessageHandler,框架自动转换消息到领域对象模型或者JSON对象等数据结构,让开发者更专注于业务处理.如果使用服务源码注解的方式声明消息处理机的后台,可以将一个通用的服务方法直接转变成具有完善功能的处理Kafka消息队列的处理机,使用起来极其简单,代码看起来一目了然,在框架级别通过多种线程池技术保证了处理机的高性能.

在使用方面,它提供了多种使用方式：

1. 直接使用Java API;
2. 与Spring环境无缝集成;
3. 服务源码注解,通过注解声明方式启动Kafka消息队列的处理机.

除此之外,它基于注解提供了消息处理机的模板项目,可以根据模板项目通过配置快速开发Kafka的消息处理机.

2. 高性能

为了在不同的业务场景下实现高性能,它提供不同的线程模型：

为了在不同的业务场景下实现高性能,它提供不同的线程模型：

1. 适合轻量级服务的同步线程模型;
2. 适合IO密集型服务的异步线程模型(细分为所有消费者流共享线程池和每个流独享线程池).

另外,异步模型中的线程池也支持确定数量线程的线程池和线程数量可伸缩的线程池.

3. 高稳定性

框架级别处理了常见的异常,计入错误日志,可用于错误手工恢复或者洗数据,并实现了优雅关机和重启等功能.

架构设计

1. 线程模型

同步线程模型

在这种线程模型中,客户端为每一个消费者流使用一个线程,每个线程负责从Kafka队列里消费消息,并且在同一个线程里进行业务处理.我们把这些线程称为消费线程,把这些线程所在的线程池叫做消息消费线程池.这种模型之所以在消息消费线程池处理业务,是因为它多用于处理轻量级别的业务,例如：缓存查询、本地计算等.

异步线程模型

在这种线程模型中,并且传递消费得到的消息到后端的异步线程池,在异步线程池中处理业务.我们仍然把前面负责消费消息的线程池称为消息消费线程池,把后面的异步线程池称为异步业务线程池.这种线程模型适合重量级的业务,例如：业务中有大量的IO操作、网络IO操作、复杂计算、对外部系统的调用等.

后端的异步业务线程池又细分为所有消费者流共享线程池和每个流独享线程池.下面详细介绍下.

所有消费者流共享线程池：所有消费者流共享线程池对比每个流独享线程池,创建更少的线程池对象,能节省些许的内存,但是,由于多个流共享同一个线程池,在数据量较大的时候,流之间的处理可能互相影响.例如,一个业务使用2个区和两个流,他们一一对应,通过生产者指定定制化的散列函数替换默认的key-hash,实现一个流(区)用来处理普通用户,另外一个流(区)用来处理VIP用户,如果两个流共享一个线程池,当普通用户的消息大量产生的时候,VIP用户只有少量,并且排在了队列的后头,就会产生饿死的情况.这个场景是可以使用多个topic来解决,一个普通用户的topic,一个VIP用户的topic,但是这样又要多维护一个topic,客户端发送的时候需要显式的进行判断topic目标,也没有多少好处.

每个流独享线程池：每个流独享线程池使用不同的异步业务线程池来处理不同的流里面的消息,互相隔离,互相独立,不互相影响,对于不同的流(区)的优先级不同的情况,或者消息在不同流(区)不均衡的情况下表现会更好,当然,创建多个线程池会多使用些许内存,但是这并不是一个大问题.

另外,异步业务线程池支持确定数量线程的线程池和线程数量可伸缩的线程池.

• 核心业务硬件资源有保证,核心服务有专享的资源池,或者线上流量可预测,请使用固定数量的线程池.
• 非核心业务一般混布,资源互相调配,线上流量不固定等情况请使用线程数量可伸缩的线程池.

2. 异常处理

对于消息处理过程中产生的业务异常,当前在业务处理的上层捕捉了Throwable,在专用的错误恢复日志中记录出错的消息,后续可根据错误恢复日志人工处理错误消息,也可重做或者洗数据.TODO：考虑实现异常Listener体系结构,对异常处理实现监听者模式,异常处理器可插拔等,默认打印错误日志.

（编辑：衡阳站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!