源码实践_实现一个mqtt(1)

目录

• 源码实践_实现一个mqtt(1)
  • 写这个原因
  • 设计topic
  • 报文
  • 测试报文

源码实践_实现一个mqtt(1)

https://gitee.com/maomaomaoge/opmq

写这个原因

因为用开源的emqx，背后的语言是erlang，不是计算的语言，但是天生的分布式，单机测试吞吐非常容易卡死，部署分布式，说实话，我不会，小公司也没有那个人力去部署，一般用一个单机搞，能多大，就多大，所以，我想实现一个单机最大效率的mq(当然这和机器的网卡和cpu有关)，只是支持简单的发布订阅即可，不正那么多弯弯绕，小公司用不到的东西

设计topic

因为我没台关注topic的解析，但是可以为以后考虑到是topic主导，还是tcp连接主导。也就是用连接去找topic，还是topic去找tcp，最终敲定topic下面带着tcp切片。数据结构用谷歌的Btree，因为这个数据结构在内存查询很快。

下面是初步实现的topic和tcp的关系，结合了btree

package opmq

import (
	"github.com/google/btree"
	"net"
)

type Topic struct {
	Name string
	Conn []net.Conn
	Hash int64
}

func NewTopic(name string) *Topic {
	t :=  &Topic{
		Name: name,
		Conn: make([]net.Conn, 0),
	}
	t.CalcHash()

	return t
}

func (t *Topic) Less(b btree.Item) bool {
	return t.Hash < b.(*Topic).Hash
}

func (t *Topic) CalcHash()  {
	b := []byte(t.Name)
	for _, v := range b {
		t.Hash += int64(v)
	}
}

测试代码

package opmq

import (
	"flag"
	"fmt"
	"github.com/google/btree"
	"strconv"
	"testing"
)

// all extracts all items from a tree in order as a slice.
func all(t *btree.BTree) (out []btree.Item) {
	t.Ascend(func(a btree.Item) bool {
		out = append(out, a)
		return true
	})
	return
}

func hashHere(s string) int64 {
	h := int64(0)
	b := []byte(s)
	for _, v := range b {
		h += int64(v)
	}

	return h
}

func TestTopic_CalcHash(t *testing.T) {
	var btreeDegree = flag.Int("degree", 32, "B-Tree degree")
	tree := btree.New(*btreeDegree)

	for i := 0; i < 10; i++ {
		to := &Topic{
			Name: strconv.Itoa(i),
		}
		to.CalcHash()

		tree.ReplaceOrInsert(to)
	}

	for _, v := range all(tree) {
		fmt.Println(v)
	}

	fmt.Println("查找")
	to2 := &Topic{
		Name: "1",
	}
	to2.CalcHash()
	fmt.Println(tree.Get(to2))
}

# 输出代码
=== RUN   TestTopic_CalcHash
&{0 [] 48}
&{1 [] 49}
&{2 [] 50}
&{3 [] 51}
&{4 [] 52}
&{5 [] 53}
&{6 [] 54}
&{7 [] 55}
&{8 [] 56}
&{9 [] 57}
查找
&{1 [] 49}
--- PASS: TestTopic_CalcHash (0.00s)
PASS

Process finished with the exit code 0

报文

直接用开源包的报文解析

测试报文

server.go

package opmq

import (
	"fmt"
	"gitee.com/maomaomaoge/opmq/packets"
	"log"
	"net"
)

func Serve() {
	ln, err := net.Listen("tcp", ":7000")
	if err != nil {
		return
	}
	for {
		conn, err := ln.Accept()
		if err != nil {
			log.Fatalln(err)
		}

		go read(conn)
	}
}

func read(conn net.Conn)  {
	for {
		packet, err := packets.ReadPacket(conn)
		if err != nil {
			return
		}

		fmt.Println("收到的报文: ", packet.String())

		ack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
		ack.Write(conn)
	}
}

开源包的最简单代码

/*
 * Copyright (c) 2021 IBM Corp and others.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    https://www.eclipse.org/legal/epl-2.0/
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Seth Hoenig
 *    Allan Stockdill-Mander
 *    Mike Robertson
 */

package main

import (
	"fmt"
	"log"
	"os"
	"time"

	"github.com/eclipse/paho.mqtt.golang"
)

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Printf("TOPIC: %s
", msg.Topic())
	fmt.Printf("MSG: %s
", msg.Payload())
}

func main() {
	mqtt.DEBUG = log.New(os.Stdout, "", 0)
	mqtt.ERROR = log.New(os.Stdout, "", 0)
	opts := mqtt.NewClientOptions().AddBroker("tcp://127.0.0.1:7000")

	c := mqtt.NewClient(opts)
	token := c.Connect()
	if token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	if token := c.Subscribe("go-mqtt/sample", 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}

	for i := 0; i < 5; i++ {
		text := fmt.Sprintf("this is msg #%d!", i)
		token := c.Publish("go-mqtt/sample", 0, false, text)
		token.Wait()
	}

	time.Sleep(6 * time.Second)

	if token := c.Unsubscribe("go-mqtt/sample"); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}

	c.Disconnect(250)

	time.Sleep(1 * time.Second)
}

serve收到的报文

=== RUN   TestServe
收到的报文:  CONNECT: dup: false qos: 0 retain: false rLength: 12 protocolversion: 4 protocolname: MQTT cleansession: true willflag: false WillQos: 0 WillRetain: false Usernameflag: false Passwordflag: false keepalive: 30 clientId:  willtopic:  willmessage:  Username:  Password: 

nice, 初步完成