Upload Project Structure

Author: Adwin2

cmd/SERVER/main.go

@@ -0,0 +1,5 @@
+module ryanMQ
+
+go 1.22.5
+
+require github.com/golang/snappy v0.0.4

config.yaml

@@ -0,0 +1,2 @@
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=

data/topics/my_test_topic_0/000001.index

@@ -0,0 +1,8 @@
+/*
+ 消息路由
+ 根据消息Key哈希到指定分区 （实现负载均衡）
+
+ ​关键技术：
+ 用 sync.Map 缓存主题元数据
+ 分区选择算法：hash(key) % partition_num
+*/

data/topics/my_test_topic_0/000001.log

@@ -0,0 +1 @@
+# 消息调度中枢

go.mod

@@ -0,0 +1,3 @@
+/*
+ 副本同步 分布式支持，暂不考虑
+*/

go.sum

@@ -0,0 +1,4 @@
+/*
+ 主题管理：创建、删除主题
+ 维护主题(Topic)与分区(Partition)的映射关系
+*/

internal/core/broker/dispatch.go

@@ -0,0 +1,4 @@
+/*
+ 持久化消费者Offset到 LevelDB/BoltDB
+ 提供Commit(Offset) 和 Fetch()接口
+*/

internal/core/broker/readme

@@ -0,0 +1,9 @@
+/*
+ 分区元信息
+
+ 记录Leader/Follower 节点 （分布式需求）
+ 维护ISR列表(In-Sync Replica)
+
+ ​关键技术：
+ 用嵌入式数据库（如 BoltDB）存储元数据
+*/

internal/core/broker/replication.go

@@ -0,0 +1,10 @@
+/*
+ 处理消费者请求
+
+ 根据Offset读取消息
+ 支持长轮询 (Long Polling)
+
+ ​关键技术：
+ 基于 io.Reader/io.Writer 实现协议解析
+ 用 goroutine 池处理并发请求
+*/

internal/core/broker/topic_manager.go

@@ -0,0 +1,88 @@
+/*
+消息编解码
+
+定义二进制协议格式(参考Kafka)
+处理压缩(Snappy)
+*/
+package protocol
+
+import (
+	"bytes"
+	"encoding/binary"
+	"ryanMQ/internal/utils/log"
+
+	"github.com/golang/snappy"
+)
+
+const (
+	ProduceRequestType       = byte(0x01)
+	FetchRequestType         = byte(0x02)
+	OffsetCommitType         = byte(0x03)
+	TopicMetadataRequestType = byte(0x04)
+)
+
+func EncodeProduceRequest(req *ProduceRequest) ([]byte, error) {
+	buf := new(bytes.Buffer) //写入缓冲区
+
+	//主题 ：2 字节长度(int16类型) + 数据
+	topicBytes := []byte(req.Topic)
+	//大端字节序存储
+	if err := binary.Write(buf, binary.BigEndian, int16(len(topicBytes))); err != nil {
+		return nil, err
+	}
+	buf.Write(topicBytes)
+
+	//分区号：4字节
+	binary.Write(buf, binary.BigEndian, req.Partition)
+
+	//消息列表 (每条消息： 4字节 长度(len) + 数据)
+	for _, msg := range req.Messages {
+		binary.Write(buf, binary.BigEndian, int32(len(msg)))
+		buf.Write(msg)
+	}
+
+	//构建最终请求（添加请求头和长度）
+	header := RequestHeader{
+		Length:      int32(buf.Len()),
+		RequestType: ProduceRequestType, //生产者请求
+	}
+	headerBytes, err := encodeHeader(header)
+	if err != nil {
+		log.Error("encode header error: %v", err)
+		return nil, err
+	}
+
+	return append(headerBytes, buf.Bytes()...), nil
+}
+
+func encodeHeader(header RequestHeader) ([]byte, error) {
+	buf := new(bytes.Buffer)
+
+	if err := binary.Write(buf, binary.BigEndian, header.Length); err != nil {
+		return nil, err
+	}
+
+	if err := binary.Write(buf, binary.BigEndian, header.RequestType); err != nil {
+		return nil, err
+	}
+
+	return buf.Bytes(), nil
+}
+
+func DecodeProduceRequest(data []byte) (*ProduceRequest, error) {
+	buf := bytes.NewBuffer(data)
+	req := &ProduceRequest{}
+
+	//解析请求头
+
+}
+
+func EncodeConsumeRequest()
+
+func CompressWithSnappy(data []byte) ([]byte, error) {
+	return snappy.Encode(nil, data), nil
+}
+
+func DecompressWithSnappy(data []byte) ([]byte, error) {
+	return snappy.Decode(nil, data)
+}

internal/core/metadata/offset_manager.go

@@ -0,0 +1,6 @@
+/*
+ 处理生产者请求
+
+ 批量写入消息
+ 返回写入成功的Offset
+*/

internal/core/metadata/partition_meta.go

@@ -0,0 +1 @@
+# 协议处理

internal/core/protocol/consume_handler.go

@@ -0,0 +1,31 @@
+package protocol
+
+type RequestHeader struct {
+	Length      int32 //消息长度
+	RequestType byte  //请求类型
+}
+
+// 生产者请求
+type ProduceRequest struct {
+	Topic     string
+	Partition int32
+	Messages  [][]byte //多条消息， 支持批量
+}
+
+type ProduceResponse struct {
+	Status byte  //成功/失败
+	Offset int64 //写入的Offset
+}
+
+type FetchRequest struct {
+	Topic       string
+	Partition   int32
+	StartOffset int64 //消费起始位置
+	MaxMsgNum   int32 //最多消费(拉取)多少条
+}
+
+type FetchResponse struct {
+	Status     byte     //成功/失败
+	NextOffset int64    //下次拉取的起始位置
+	Messages   [][]byte //实际返回的消息
+}

internal/core/protocol/encode_decode.go

@@ -0,0 +1,10 @@
+/*
+ 维护分区下的多个segment
+ 根据时间/空间策略清理旧数据 retention策略
+
+ ​关键技术：
+ 使用 os.File + bufio.Writer 实现缓冲写入
+ ​零拷贝读取：syscall.Sendfile 直接发送文件内容
+ ​内存映射：syscall.Mmap 加速索引文件读取
+
+*/

internal/core/protocol/produce_handler.go

@@ -0,0 +1 @@
+# 存储引擎

internal/core/protocol/readme

@@ -0,0 +1,6 @@
+/*
+ 管理单个分段文件 (.log + .index)
+ 实现消息追加 func Append(msg []byte)
+ 构建稀疏索引 4KB/条
+ 处理文件滚动 达到 segment.max.size 后创建新文件
+*/

internal/core/protocol/types.go

@@ -0,0 +1,4 @@
+/*
+预写日志 (Write-Ahead Log, 用于未刷盘数据的恢复)
+
+*/

internal/core/storage/patition.go

@@ -0,0 +1,10 @@
+/*
+ 连接管理
+
+ 封包/拆包 （处理粘包）
+ 心跳维护 （保活机制）
+
+
+ 基于 net.Listener 实现非阻塞 IO
+ 使用 goroutine-per-connection 模型（Go 高并发优势）
+*/

internal/core/storage/readme

@@ -0,0 +1,6 @@
+/*
+ TCP服务器
+
+ 监听端口，接受连接
+ 为每个连接启动处理协程
+*/

internal/core/storage/segment.go

@@ -0,0 +1,19 @@
+package log
+
+import (
+	"log"
+)
+
+var (
+	DebugEnabled = true //默认开启 Debug
+)
+
+func Debug(format string, v ...any) {
+	if DebugEnabled {
+		log.Printf("[DEBUG] "+format, v...)
+	}
+}
+
+func Error(format string, v ...any) {
+	log.Printf("[ERROR] "+format, v...)
+}

internal/core/storage/wal.go

@@ -0,0 +1,9 @@
+/*
+consumer.go：消费者接口
+Subscribe(topic string) <-chan Message
+Ack(offset int) 手动提交 Offset
+
+关键技术：
+基于 net.Dial 实现客户端通信
+用 channel 实现消息推送
+*/

internal/network/tcp/connection.go

@@ -0,0 +1,5 @@
+/*
+producer.go：生产者接口
+Send(msg []byte) (offset int, err error)
+支持同步/异步发送
+*/

internal/network/tcp/server.go

@@ -0,0 +1,26 @@
+# 存储设计
+
+- [ ] 分段文件 + 内存映射（mmap）减少磁盘IO
+- [ ] 索引文件 跳表结构  (offset --> 物理地址)
+- [ ] 多副本同步写入 (ISR)
+
+顺序写入： 避免随机IO
+分段存储（segment 对应一个 offset 快速定位 清理过期数据） 稀疏索引
+直接利用操作系统页缓存 磁盘文件映射到内存
+
+零拷贝传输 ： sendfile系统调用，直接将磁盘数据通过DMA传输到网卡
+
+作batch单位传输 将数据写到页缓存，操作系统异步刷盘 flush.internal.ms 控制刷盘频率
+
+：ELK日志系统
+
+按时间和bytes大小删除过期segment
+**故障容错和恢复**
+​Checksum 校验：每条消息包含 CRC32 校验码，防止数据损坏。
+​恢复日志（Recovery Log）​：Broker 重启时通过读取 Segment 文件和索引重建状态，而非依赖外部数据库。
+​可借鉴点
+​自包含存储：将元数据与数据统一存储（如将索引与数据文件绑定），避免外部依赖。
+​数据完整性：在关键路径添加校验机制（如网络传输、持久化存储）。
+内存**池化** 减少GC垃圾回收压力
+事件驱动模型 :
+FileChannel.mmap 频繁读写时 将文件映射到内存