golang基于TCP协议实现消息队列

ゞ浴缸里的玫瑰 2023-02-14 05:59 11阅读 0赞

1. 整体的设计图如下，用的生产消费的模型

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvV29ybGRZYW5nU29uZw_size_16_color_FFFFFF_t_70][]

2. sever端代码设计

思路：监听端口-->初始化消费类型类似kafka的topic-->product client生产数据-->生产的数据存到数组中-->consumer client-->从数组中消费数据

3. 整代码

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvV29ybGRZYW5nU29uZw_size_16_color_FFFFFF_t_70 1][]

4. 服务端代码

package main
    
    import (
    	"encoding/json"
    	"fmt"
    	"net"
    	"src/lib"
    )
    // 接收数据
    type Msg struct {
    	K string `json:"key"`
    	V string `json:"value"`
    	T string `json:"type"`
    }
    
    // 返回数据
    type Result struct {
    	Code int `json:"code"`
    	Data interface{} `json:"data"`
    	Status bool `json:"status"`
    }
    
    // 队列信息
    var mqInfoMap = make(map[string][]string)
    // 注册多个消息，类似kafka的topic
    var mqArr = []string{"mq_test1", "mq_test2"}
    
    func initData()  {
    	// 初始化队列信息
    	for i :=0; i < len(mqArr); i++  {
    		var arrStr = []string{}
    		mqInfoMap[mqArr[i]] = arrStr
    	}
    }
    
    
    
    func main() {
    	// 服务端端口
    	service := lib.Port
    	// 绑定
    	tcpAddr, err := net.ResolveTCPAddr("tcp", service)
    	lib.CheckError(err)
    	// 监听
    	listener, err := net.ListenTCP("tcp", tcpAddr)
    	lib.CheckError(err)
    	// 初始化数据
    	initData()
    	fmt.Println(mqInfoMap)
    	for {
    		// 接受
    		conn, err := listener.Accept()
    		if err != nil {
    			continue
    		}
    		// 创建 Goroutine
    		go handleClient(conn)
    	}
    }
    func handleClient(conn net.Conn) {
    	// 逆序调用 Close() 保证连接能正常关闭
    	defer conn.Close()
    	var buf [512]byte
    	for {
    		// 接收数据
    		n, err := conn.Read(buf[0:])
    		if err != nil {
    			return
    		}
    		rAddr := conn.RemoteAddr()
    		msg := string(buf[0:n])
    		fmt.Println("Receive from client", rAddr.String(), msg)
    		var m Msg
    		err = json.Unmarshal([]byte(msg), &m)
    		if(err != nil || m == Msg{}){
    			conn.Write([]byte("1"))
    			return
    		}
    
    		if m.T == "c"{
    			handleConsumer(conn, m)
    		}else {
    			handleProduct(conn, m)
    		}
    	}
    }
    
    func handleConsumer(conn net.Conn, m Msg) {
    	var result = Result{}
    	arr := mqInfoMap[m.K]
    	if len(arr) > 0 {
    		arr = append(arr[:0],arr[0+1:]... )
    		mqInfoMap[m.K] = arr
    		conn.Write([]byte("0"))
    		result.Data = 0
    		result.Status = true
    		jsonByte, _ := json.Marshal(result)
    		conn.Write(jsonByte)
    		return
    	}
    	// 没有可以消费的数据
    	result.Data = 1
    	result.Status = true
    	jsonByte, _ := json.Marshal(result)
    	conn.Write(jsonByte)
    }
    
    func handleProduct(conn net.Conn, m Msg) {
    	var result = Result{}
    	arr := mqInfoMap[m.K]
    	arr = append(arr, m.V)
    	mqInfoMap[m.K] = arr
    
    	result.Data = 0
    	result.Status = true
    	jsonByte, _ := json.Marshal(result)
    	conn.Write(jsonByte)
    }

5. 生产者客户端代码

package main
    
    import (
    	"fmt"
    	"net"
    	"os"
    	"src/lib"
    	"time"
    )
    
    func main() {
    	var buf [512]byte
    	// 绑定
    	tcpAddr, err := net.ResolveTCPAddr("tcp", "127.0.0.1" + lib.Port)
    	lib.CheckError(err)
    	// 连接
    	conn, err := net.DialTCP("tcp", nil, tcpAddr)
    	lib.CheckError(err)
    	rAddr := conn.RemoteAddr()
    	// 发送
    	msg := "{\"key\":\"mq_test1\",\"value\":\"hi server\"}"
    	n, err := conn.Write([]byte(msg))
    	lib.CheckError(err)
    	// 接收
    	n, err = conn.Read(buf[0:])
    	lib.CheckError(err)
    	fmt.Println("Reply form server", rAddr.String(), string(buf[0:n]))
    	time.Sleep(time.Second * 2)
    
    	conn.Close()
    	os.Exit(0)
    }

6. 消费者客户端代码

package main
    
    import (
    	"fmt"
    	"net"
    	"os"
    	"src/lib"
    	"time"
    )
    
    func main() {
    	var buf [512]byte
    	// 绑定
    	tcpAddr, err := net.ResolveTCPAddr("tcp", "127.0.0.1" + lib.Port)
    	lib.CheckError(err)
    	// 连接
    	conn, err := net.DialTCP("tcp", nil, tcpAddr)
    	lib.CheckError(err)
    	rAddr := conn.RemoteAddr()
    	// 发送
    	msg := "{\"key\":\"mq_test1\",\"type\":\"c\"}"
    	n, err := conn.Write([]byte(msg))
    	lib.CheckError(err)
    	// 接收
    	n, err = conn.Read(buf[0:])
    	lib.CheckError(err)
    	fmt.Println("Reply form server", rAddr.String(), string(buf[0:n]))
    	time.Sleep(time.Second * 2)
    
    	conn.Close()
    	os.Exit(0)
    }

7. lib包代码

package lib
    
    import (
    	"fmt"
    	"os"
    )
    // 参考文档：https://studygolang.com/articles/12282?fr=sidebar
    // 参考文档：https://studygolang.com/articles/24440?fr=sidebar
    
    
    // 消息数据类型
    type ReceiveData struct{
    	M map[string]string
    }
    
    const(
    	Port = ":5000" // 服务端接受的端口
    )
    
    
    func CheckError(err error) {
    	if err != nil {
    		fmt.Fprintf(os.Stderr, "Fatal error: %s", err.Error())
    		os.Exit(1)
    	}
    }

8. 仅供参考学习，如有疑问欢迎留言交流。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvV29ybGRZYW5nU29uZw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200601132249126.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvV29ybGRZYW5nU29uZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvV29ybGRZYW5nU29uZw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200601142430515.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvV29ybGRZYW5nU29uZw==,size_16,color_FFFFFF,t_70