網站首頁 編程語言 正文
前言
趁著有空余時間，就想著擼一個mini的生產-消費消息隊列，說干就干了。自己是個javer，這次實現，特意換用了go。沒錯，是零基礎上手go，順便可以學學go。
前置知識：
- go基本語法
- 消息隊列概念，也就三個：生產者、消費者、隊列
目的
- 沒想著實現多復雜，因為時間有限，就mini就好，mini到什么程度呢
- 使用雙向鏈表數據結構作為隊列
- 有多個topic可供生產者生成消息和消費者消費消息
- 支持生產者并發寫
- 支持消費者讀，且ok后，從隊列刪除
- 消息不丟失（持久化）
- 高性能（先這樣想）
設計
整體架構
協議
通訊協議底層使用tcp，mq是基于tcp自定義了一個協議，協議如下
type Msg struct {
Id int64
TopicLen int64
Topic string
// 1-consumer 2-producer 3-comsumer-ack 4-error
MsgType int64 // 消息類型
Len int64 // 消息長度
Payload []byte // 消息
}
Payload使用字節數組，是因為不管數據是什么，只當做字節數組來處理即可。Msg承載著生產者生產的消息，消費者消費的消息，ACK、和錯誤消息，前兩者會有負載，而后兩者負載和長度都為空
協議的編解碼處理，就是對字節的處理，接下來有從字節轉為Msg，和從Msg轉為字節兩個函數
func BytesToMsg(reader io.Reader) Msg {
m := Msg{}
var buf [128]byte
n, err := reader.Read(buf[:])
if err != nil {
fmt.Println("read failed, err:", err)
}
fmt.Println("read bytes:", n)
// id
buff := bytes.NewBuffer(buf[0:8])
binary.Read(buff, binary.LittleEndian, &m.Id)
// topiclen
buff = bytes.NewBuffer(buf[8:16])
binary.Read(buff, binary.LittleEndian, &m.TopicLen)
// topic
msgLastIndex := 16 + m.TopicLen
m.Topic = string(buf[16: msgLastIndex])
// msgtype
buff = bytes.NewBuffer(buf[msgLastIndex : msgLastIndex + 8])
binary.Read(buff, binary.LittleEndian, &m.MsgType)
buff = bytes.NewBuffer(buf[msgLastIndex : msgLastIndex + 16])
binary.Read(buff, binary.LittleEndian, &m.Len)
if m.Len <= 0 {
return m
}
m.Payload = buf[msgLastIndex + 16:]
return m
}
func MsgToBytes(msg Msg) []byte {
msg.TopicLen = int64(len([]byte(msg.Topic)))
msg.Len = int64(len([]byte(msg.Payload)))
var data []byte
buf := bytes.NewBuffer([]byte{})
binary.Write(buf, binary.LittleEndian, msg.Id)
data = append(data, buf.Bytes()...)
buf = bytes.NewBuffer([]byte{})
binary.Write(buf, binary.LittleEndian, msg.TopicLen)
data = append(data, buf.Bytes()...)
data = append(data, []byte(msg.Topic)...)
buf = bytes.NewBuffer([]byte{})
binary.Write(buf, binary.LittleEndian, msg.MsgType)
data = append(data, buf.Bytes()...)
buf = bytes.NewBuffer([]byte{})
binary.Write(buf, binary.LittleEndian, msg.Len)
data = append(data, buf.Bytes()...)
data = append(data, []byte(msg.Payload)...)
return data
}
隊列
使用container/list，實現先入先出，生產者在隊尾寫，消費者在隊頭讀取
package broker
import (
"container/list"
"sync"
)
type Queue struct {
len int
data list.List
}
var lock sync.Mutex
func (queue *Queue) offer(msg Msg) {
queue.data.PushBack(msg)
queue.len = queue.data.Len()
}
func (queue *Queue) poll() Msg{
if queue.len == 0 {
return Msg{}
}
msg := queue.data.Front()
return msg.Value.(Msg)
}
func (queue *Queue) delete(id int64) {
lock.Lock()
for msg := queue.data.Front(); msg != nil; msg = msg.Next() {
if msg.Value.(Msg).Id == id {
queue.data.Remove(msg)
queue.len = queue.data.Len()
break
}
}
lock.Unlock()
}
方法offer往隊列里插入數據，poll從隊列頭讀取數據素，delete根據消息ID從隊列刪除數據。這里使用Queue結構體對List進行封裝，其實是有必要的，List作為底層的數據結構，我們希望隱藏更多的底層操作，只給客戶提供基本的操作
delete操作是在消費者消費成功且發送ACK后，對消息從隊列里移除的，因為消費者可以多個同時消費，所以這里進入臨界區時加鎖（em，加鎖是否就一定會影響對性能有較大的影響呢）
broker
broker作為服務器角色，負責接收連接，接收和響應請求
package broker
import (
"bufio"
"net"
"os"
"sync"
"time"
)
var topics = sync.Map{}
func handleErr(conn net.Conn) {
defer func() {
if err := recover(); err != nil {
println(err.(string))
conn.Write(MsgToBytes(Msg{MsgType: 4}))
}
}()
}
func Process(conn net.Conn) {
handleErr(conn)
reader := bufio.NewReader(conn)
msg := BytesToMsg(reader)
queue, ok := topics.Load(msg.Topic)
var res Msg
if msg.MsgType == 1 {
// comsumer
if queue == nil || queue.(*Queue).len == 0{
return
}
msg = queue.(*Queue).poll()
msg.MsgType = 1
res = msg
} else if msg.MsgType == 2 {
// producer
if ! ok {
queue = &Queue{}
queue.(*Queue).data.Init()
topics.Store(msg.Topic, queue)
}
queue.(*Queue).offer(msg)
res = Msg{Id: msg.Id, MsgType: 2}
} else if msg.MsgType == 3 {
// consumer ack
if queue == nil {
return
}
queue.(*Queue).delete(msg.Id)
}
conn.Write(MsgToBytes(res))
}
MsgType等于1時，直接消費消息；MsgType等于2時是生產者生產消息，如果隊列為空，那么還需創建一個新的隊列，放在對應的topic下；MsgType等于3時，代表消費者成功消費，可以
刪除消息
我們說消息不丟失，這里實現不完全，我就實現了持久化（持久化也沒全部實現）。思路就是該topic對應的隊列里的消息，按協議格式進行序列化，當broker啟動時，從文件恢復
持久化需要考慮的是增量還是全量，需要保存多久，這些都會影響實現的難度和性能（想想Kafka和Redis的持久化），這里表示簡單實現就好：定時器定時保存
func Save() {
ticker := time.NewTicker(60)
for {
select {
case <-ticker.C:
topics.Range(func(key, value interface{}) bool {
if value == nil {
return false
}
file, _ := os.Open(key.(string))
if file == nil {
file, _ = os.Create(key.(string))
}
for msg := value.(*Queue).data.Front(); msg != nil; msg = msg.Next() {
file.Write(MsgToBytes(msg.Value.(Msg)))
}
_ := file.Close()
return false
})
default:
time.Sleep(1)
}
}
}
有一個問題是，當上面的delete操作時，這里的file文件需不需要跟著delete掉對應的消息？答案是需要刪除的，如果不刪除，只能等下一次的全量持久化來覆蓋了，中間就有臟數據問題
下面是啟動邏輯
package main
import (
"awesomeProject/broker"
"fmt"
"net"
)
func main() {
listen, err := net.Listen("tcp", "127.0.0.1:12345")
if err != nil {
fmt.Print("listen failed, err:", err)
return
}
go broker.Save()
for {
conn, err := listen.Accept()
if err != nil {
fmt.Print("accept failed, err:", err)
continue
}
go broker.Process(conn)
}
}
生產者
package main
import (
"awesomeProject/broker"
"fmt"
"net"
)
func produce() {
conn, err := net.Dial("tcp", "127.0.0.1:12345")
if err != nil {
fmt.Print("connect failed, err:", err)
}
defer conn.Close()
msg := broker.Msg{Id: 1102, Topic: "topic-test", MsgType: 2, Payload: []byte("我")}
n, err := conn.Write(broker.MsgToBytes(msg))
if err != nil {
fmt.Print("write failed, err:", err)
}
fmt.Print(n)
}
消費者
package main
import (
"awesomeProject/broker"
"bytes"
"fmt"
"net"
)
func comsume() {
conn, err := net.Dial("tcp", "127.0.0.1:12345")
if err != nil {
fmt.Print("connect failed, err:", err)
}
defer conn.Close()
msg := broker.Msg{Topic: "topic-test", MsgType: 1}
n, err := conn.Write(broker.MsgToBytes(msg))
if err != nil {
fmt.Println("write failed, err:", err)
}
fmt.Println("n", n)
var res [128]byte
conn.Read(res[:])
buf := bytes.NewBuffer(res[:])
receMsg := broker.BytesToMsg(buf)
fmt.Print(receMsg)
// ack
conn, _ = net.Dial("tcp", "127.0.0.1:12345")
l, e := conn.Write(broker.MsgToBytes(broker.Msg{Id: receMsg.Id, Topic: receMsg.Topic, MsgType: 3}))
if e != nil {
fmt.Println("write failed, err:", err)
}
fmt.Println("l:", l)
}
消費者這里ack時重新創建了連接，如果不創建連接的話，那服務端那里就需要一直從conn讀取數據，直到結束。思考一下，像RabbitMQ的ack就有自動和手工的ack，如果是手工的ack，必然需要一個新的連接，因為不知道客戶端什么時候發送ack，自動的話，當然可以使用同一個連接，but這里就簡單創建一條新連接吧
啟動
先啟動broker，再啟動producer，然后啟動comsumer，OK，能跑，能實現發送消息到隊列，從隊列消費消息
總結
整體雖然簡單，但畢竟是使用go實現的，就是看似一頓操作猛如虎，實質慌如狗。第一時間就被go的gopath和go mod困擾住，后面語法的使用，比如指針，傳值傳引用等，最頭疼的就是類型轉換，作為一個javer，使用go進行類型轉換，著實被狠狠得虐了一番。
原文鏈接：https://juejin.cn/post/7041085344481017887
相關推薦
- 2023-04-08 Linux下動靜態庫的打包與使用指南(C/C++)_C 語言
- 2022-12-12 C++?Boost?Chrono實現計時碼表流程詳解_C 語言
- 2022-11-21 C#基礎知識之字符串和正則表達式_C#教程
- 2022-05-13 python實現簡易圖書管理系統_python
- 2023-01-31 C#實現偽裝文件夾功能_C#教程
- 2021-12-07 C語言實現頁面置換算法(FIFO、LRU)_C 語言
- 2023-03-25 Python?Flask-Login模塊使用案例詳解_python
- 2022-09-23 Pandas時間類型轉換與處理的實現示例_python
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤：Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列，發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速