仿照Kafka,从零开始自实现 MQ
仿照Kafka,从零开始自实现 MQ,实现了 Kafka 中 80% 的基础功能。学习 Kafka 的话如果只是看文章和源码,可能不久就会忘了,还是自己实现一个「精简版」的 Kafka 吧,
实现功能概览
1、基于内存Queue实现生产和消费API
- 1) 创建内存Queue, 作为底层消息存储
- 2) 定义Topic, 支持多个Topic
- 3) 定义Producer, 支持Send消息
- 4) 定义Consumer, 支持Poll消息
2、设计自定义Queue,实现消息确认和消费offset
- 1) 自定义内存Message数组模拟Queue。
- 2) 使用指针记录当前消息写入位置。
- 3) 对于每个命名消费者, 用指针记录消费位置
3、拆分broker和client(包括producer和consumer)
- 1) 将Queue保存到web server端
- 2) 设计消息读写API接口, 确认接口, 提交offset接口
- 3) producer和consumer通过httpclient访问Queue
- 4) 实现消息确认, offset提交
- 5) 实现consumer从offset增量拉取
项目目录
bitkylin-mq
项目设计及项目能力
Server
一、Topic
- 维护ArrayList用于模拟持久化消息「原因:消息需要随机访问」
- 设定消息队列容量,达到容量时无法再生产消息
- 当前消息的最大索引
二、ConsumerGroup
- 消费者组由消费者组名和topic名共同决定,即不同topic的消费者组相互独立,不会相互影响
- 需根据topic创建消费者组,即消费者组必须关联topic
- 消费者组创建后,默认从头完整消费关联topic的所有消息
- 同一个消费者组内,各个消费者总共消费一次「最少消费一次」所关联topic的所有消息
三、broker
- 一个broker关联一个ConsumerGroup列表和一个Topic列表
- 通过broker暴露的接口,可以展示关联ConsumerGroup列表和Topic列表的概览信息
- 通过broker暴露的接口,可以向一个topic中生产消息
- 通过broker暴露的接口,可以根据消费者组名和topic名消费消息
注:本次仅实现单个broker,broker后实现了topic和consumerGroup「消费者组」,细节结构图如下:
client
- 客户端通过topic名生产消息
- 客户端根据消费者组名和topic名消费消息
- 客户端消费消息时,可以同时获得消费者组的offset「偏移量」
- 客户端消费消息成功后,需手动更新消费者组的offset。若不更新,客户端默认无法消费后面的消息。
- 客户端消费消息失败时,不应更新消费者组的offset。此时客户端可以重复消费当条消息。
- 多个客户端可以使用同一个消费者组消费同一个topic;可以使用不同的消费者组消费同一个topic;可以使用不同的消费者组消费不同的topic
客户端工作示意图如下:
项目结构
本项目共提供四个module:
bitkylin-mq-server
bitkylin-mq-api
bitkylin-mq-client-producer
bitkylin-mq-client-consumer
各module的介绍如下:
1. bitkylin-mq-server
提供MQ服务端,提供broker以及其关联的ConsumerGroup和Topic等,主要实现如下功能:
- 展示MQ概览信息,包括topic和ConsumerGroup的详细信息
- 创建消费者组,创建消费者组后,即可使用该消费者组消费消息
- 生产消息,将消息发送至指定topic
- 基于指定消费者组消费消息,消费消息但不更新关联消费者组的offset
- 基于指定消费者组消费消息,消费消息且自动更新关联消费者组的offset
- 手动更新指定消费者组的偏移量
2. bitkylin-mq-api
提供供客户端使用的api,通过feignClient形式提供,客户端可直接使用,执行RPC,当前实现如下功能:
- 发送消息至指定topic
- 订阅指定topic的消息。自动创建消费者组,使用观察者模式轮询消息并消费。
3. bitkylin-mq-client-producer
消息生产客户端,通过feign-api生产消息,当前实现如下演示功能:
随机向topic名为「topic-1」和「topic-2」的topic中发送消息,每隔3秒发送一次消息。
4. bitkylin-mq-client-consumer
消息消费客户端,通过feign-api消费消息,当前实现如下演示功能:
- 创建消费者组「spring-group-1」订阅「topic-1」,并打印订阅的消息。
- 创建消费者组「spring-group-2」订阅「topic-2」,并打印订阅的消息。
代码演示
- 运行module「bitkylin-mq-server」,启动MQ的broker,启动消息服务。
- 运行module「bitkylin-mq-client-consumer」和「bitkylin-mq-client-producer」,开启消息订阅演示任务和消息发送演示任务。
- 此时可通过「bitkylin-mq-client-consumer」的控制台,看到消息不断被消费。
2021-01-24 01:55:58.008 INFO 2516 --- [pool-1-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-1: topic-1-msg:1
2021-01-24 01:56:00.996 INFO 2516 --- [pool-1-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-1: topic-1-msg:2
2021-01-24 01:56:04.000 INFO 2516 --- [pool-1-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-1: topic-1-msg:3
2021-01-24 01:56:07.004 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:4
2021-01-24 01:56:10.015 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:5
2021-01-24 01:56:13.011 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:6
2021-01-24 01:56:16.011 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:7
2021-01-24 01:56:19.006 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:8
2021-01-24 01:56:21.997 INFO 2516 --- [pool-1-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-1: topic-1-msg:9
2021-01-24 01:56:24.994 INFO 2516 --- [pool-1-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-1: topic-1-msg:10
2021-01-24 01:56:28.002 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:11
2021-01-24 01:56:30.991 INFO 2516 --- [pool-1-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-1: topic-1-msg:12
2021-01-24 01:56:34.014 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:13
2021-01-24 01:56:37.010 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:14
2021-01-24 01:56:40.004 INFO 2516 --- [pool-2-thread-1] .c.c.BitkylinMqClientConsumerApplication : 收到消息:spring-group-2: topic-2-msg:15
- 打开postman,发送如下请求创建专用于postman的消费者组:
POST http://localhost:8080/mq/broker/consumer-group/create{"groupName": "postman-group-1","topicName": "topic-1"
}
- 发送如下请求即可消费消息,且自动确认「无需手动更新消费者组的offset」
POST http://localhost:8080/mq/broker/message/simple-pull{"groupName": "postman-group-1","topicName": "topic-1"
}
可以发现,postman可以独立消费指定topic的消息,不受Spring程序消费的影响。当然,postman可以直接使用Spring程序一致的消费者组,以共同消费消息。
此时查询MQ的概览信息:
GET http://localhost:8080/mq/broker/overview
响应:
{"groupList": [{"groupName": "spring-group-1","topic": {"name": "topic-1","capacity": 1000,"maxIndex": 14},"offset": 15},{"groupName": "postman-group-1","topic": {"name": "topic-1","capacity": 1000,"maxIndex": 14},"offset": 5},{"groupName": "spring-group-2","topic": {"name": "topic-2","capacity": 1000,"maxIndex": 17},"offset": 18}]
}
局限性
- 每个topic的队列容量是固定的,队列满后拒绝生产消息,暂不支持清理历史消息。
- 消息消费未加锁,如果一个消费者组的多个消费者高并发消费消息,可能导致同一条消息被消费多次。
仿照Kafka,从零开始自实现 MQ相关推荐
- Kafka从零开始 (Kafka是什么?)
Kafka从零开始 <Kafka是什么?> 1. 百度定义 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka是一种高吞吐量的分布式发布订 ...
- ibm linux mq 发送消息_RabbitMq、ActiveMq、Kafka和Redis做Mq对比
一.RabbitMq RabbitMQ是一个Advanced Message Queuing Protocol(AMQP)的开源实现,由以高性能.可伸缩性出名的Erlang写成.RabbitMQ Se ...
- Kafka压力测试(写入MQ消息压测和消费MQ消息压测)
1.测试目的 本次性能测试在正式环境下单台服务器上Kafka处理MQ消息能力进行压力测试.测试包括对Kafka写入MQ消息和消费MQ消息进行压力测试,根据10w.100w和1000w级别的消息处理结果 ...
- kafka reassign 限速_简单明了,Kafka与MQ的区别
作为消息队列来说,企业中选择mq的还是多数,因为像Rabbit,Rocket等mq中间件都属于很成熟的产品,性能一般但可靠性较强, 而kafka原本设计的初衷是日志统计分析,现在基于大数据的背景下也可 ...
- Kafka与其他MQ对比
与其他MQ相比较,Kafka有一些优缺点,主要如下, 优点: 可扩展.Kafka集群可以透明的扩展,增加新的服务器进集群. 高性能.Kafka性能远超过传统的ActiveMQ.RabbitMQ等,Ka ...
- mq和kafka的不同之处
其实,作为消息队列来说,企业中选择mq的还是多数,因为像Rabbit,Rocket等mq中间件都属于很成熟的产品,性能一般但可靠性较强,而kafka原本设计的初衷是日志统计分析,现在基于大数据的背景下 ...
- 手记系列之六 ----- 分享个人使用kafka经验
前言 本篇文章主要介绍的关于本人从刚工作到现在使用kafka的经验,内容非常多,包含了kafka的常用命令,在生产环境中遇到的一些场景处理,kafka的一些web工具推荐等等.由于kafka这块的记录 ...
- websphere mq 查看队列中是否有数据_全网最全的 “消息队列”
消息队列的使用场景 以下介绍消息队列在实际应用常用的使用场景.异步处理.应用解耦.流量削锋和消息通讯四个场景. 1]异步处理:场景说明:用户注册后,需要发注册邮件和注册短信. 引入消息队列后架构如下: ...
- RabbitMQ和kafka从几个角度简单的对比--转
业界对于消息的传递有多种方案和产品,本文就比较有代表性的两个MQ(rabbitMQ,kafka)进行阐述和做简单的对比, 在应用场景方面, RabbitMQ,遵循AMQP协议,由内在高并发的erlan ...
最新文章
- Oxford Nanopore sequencing, hybrid error correction, and de novo assembly of a eukaryotic genome
- python拿什么做可视化界面好-手把手教你用Python做个可视化的“剪刀石头布”小游戏...
- 产品设计体会(9000)人人都是产品经理,系列说明
- reverse() 几种操作
- Web如何应对流量劫持?
- Flex 布局 - 容器
- python 批量修改密码
- ps查看所有php进程,ps命令就是最基本进程查看命令
- jeesite使用心得(二)
- 客户端脚本selectivizr,PIE,Respond.js
- socket.io 消息发送
- QC3.0手机充电器直接输出12V电压,自制诱骗线
- C++中的仿函数(functors)和仿函数适配器(adapter function)
- Vue中的filter过滤器是使用方法
- 金山词霸划译功能对IDE的影响
- 【Unity Shader 中Pass相关介绍_第一篇】
- 2018秋招暨年度总结
- Open Cascade 读取STEP、STL
- GPON网络架构简介
- Windows 8 客户端预览版(Consumer Preview)抢先体验