use rabbitmq replaced zeromq

internal/miner/miner.go

@@ -26,7 +26,7 @@ func NewMiner(coin string, poolAddress string, minerConn net.Conn) (*Miner, erro
 	if err != nil {
 		return nil, fmt.Errorf("pool连接失败: %v", err)
 	}
-
+	fmt.Println("有新矿工接入")
 	return &Miner{
 		Coin:        coin,
 		PoolConn:    poolConn,
@@ -68,6 +68,7 @@ func (m *Miner) HandleMinerMsg(ch chan string) {
 
 	for {
 		msgStr, err := reader.ReadString('\n')
+		fmt.Println(topic + msgStr)
 		if err != nil {
 			fmt.Println("miner消息读取失败：", err)
 			return
@@ -113,6 +114,7 @@ func (m *Miner) HandlePoolMsg() {
 	reader := bufio.NewReader(m.PoolConn)
 	for {
 		poolMsg, err := reader.ReadString('\n')
+		fmt.Println("[pool]:", poolMsg)
 		if err != nil {
 			fmt.Println("pool消息读取失败：", err)
 			return

internal/proxy.go

@@ -64,7 +64,7 @@ func (p *ProxyCtx) handleMinerConnect() {
 			fmt.Println(TOPIC+"接收连接失败:", err)
 			continue
 		}
-
+		fmt.Println("有新矿工接入")
 		go func(conn net.Conn) {
 			ch := make(chan string, 1)
 			minerObj, err := miner.NewMiner(p.Coin, p.Cfg.DefaultAddr, conn)
@@ -101,7 +101,7 @@ func (p *ProxyCtx) handleMinerConnect() {
 }
 
 func (p *ProxyCtx) startZMQ() {
-	zmq.StartZMQ(p.Cfg.ZmqAddr, p.ProxyList, &p.mu)
+	zmq.StartZMQ(p.Cfg.ZmqAddr, "testQueue", p.ProxyList, &p.mu)
 }
 
 func StartProxy() {

internal/zmq/zmq.go

@@ -1,49 +1,134 @@
 package zmq
 
+// import (
+// 	"encoding/json"
+// 	"fmt"
+// 	"proxy/internal/msg"
+// 	"sync"
+// 	"time"
+
+// 	"github.com/zeromq/goczmq"
+// )
+
+// func initZmqPull(sub_to string) *goczmq.Sock {
+// 	pull_ch, err := goczmq.NewPull(sub_to)
+// 	if err != nil {
+// 		fmt.Println("[zmq]:", err)
+// 	}
+// 	//pull_ch.SetMaxmsgsize(1024 * 1024 * 8)
+// 	return pull_ch
+// }
+
+// func handleZmqMsg(zmqMsg []byte, proxyList map[string]string, proxyListLock *sync.RWMutex) {
+// 	var data msg.ZmqMsg
+// 	if err := json.Unmarshal(zmqMsg, &data); err != nil {
+// 		fmt.Println("[zmq]:", err)
+// 		return
+// 	}
+// 	proxyListLock.Lock()
+// 	// MethodID: 0(add), 1(delete)
+// 	if data.MethodID == 0 {
+// 		proxyList[data.ID] = data.Address
+// 	} else {
+// 		delete(proxyList, data.ID)
+// 	}
+// 	proxyListLock.Unlock()
+// }
+
+// func StartZMQ(zmqAddr string, proxyList map[string]string, proxyListLock *sync.RWMutex) {
+// 	conn := initZmqPull(zmqAddr)
+// 	for {
+// 		zmqMsg, _, err := conn.RecvFrame()
+// 		fmt.Println("[zmq]:", zmqMsg)
+// 		if err != nil {
+// 			fmt.Println("[zmq recv]:", err)
+// 			time.Sleep(time.Second) // 防止CPU空转
+// 			continue
+// 		}
+// 		handleZmqMsg(zmqMsg, proxyList, proxyListLock)
+// 	}
+// }
+
 import (
 	"encoding/json"
 	"fmt"
 	"proxy/internal/msg"
 	"sync"
-	"time"
 
-	"github.com/zeromq/goczmq"
+	amqp "github.com/rabbitmq/amqp091-go"
 )
 
-func initZmqPull(sub_to string) *goczmq.Sock {
-	pull_ch, err := goczmq.NewPull(sub_to)
+func initRabbitConsumer(rabbitURL, queueName string) (*amqp.Connection, *amqp.Channel, <-chan amqp.Delivery, error) {
+	conn, err := amqp.Dial(rabbitURL)
 	if err != nil {
-		fmt.Println("[zmq]:", err)
+		return nil, nil, nil, fmt.Errorf("failed to connect to rabbitmq: %v", err)
 	}
-	//pull_ch.SetMaxmsgsize(1024 * 1024 * 8)
-	return pull_ch
+
+	ch, err := conn.Channel()
+	if err != nil {
+		conn.Close()
+		return nil, nil, nil, fmt.Errorf("failed to open a channel: %v", err)
+	}
+
+	// 确保队列存在
+	_, err = ch.QueueDeclare(
+		queueName, // name
+		true,      // durable
+		false,     // delete when unused
+		false,     // exclusive
+		false,     // no-wait
+		nil,       // arguments
+	)
+	if err != nil {
+		conn.Close()
+		return nil, nil, nil, fmt.Errorf("failed to declare a queue: %v", err)
+	}
+
+	msgs, err := ch.Consume(
+		queueName, // queue
+		"",        // consumer
+		true,      // auto-ack
+		false,     // exclusive
+		false,     // no-local
+		false,     // no-wait
+		nil,       // args
+	)
+	if err != nil {
+		conn.Close()
+		return nil, nil, nil, fmt.Errorf("failed to register a consumer: %v", err)
+	}
+
+	return conn, ch, msgs, nil
 }
 
-func handleZmqMsg(zmqMsg []byte, proxyList map[string]string, proxyListLock *sync.RWMutex) {
+func handleRabbitMsg(body []byte, proxyList map[string]string, proxyListLock *sync.RWMutex) {
 	var data msg.ZmqMsg
-	if err := json.Unmarshal(zmqMsg, &data); err != nil {
-		fmt.Println("[zmq]:", err)
+	if err := json.Unmarshal(body, &data); err != nil {
+		fmt.Println("[rabbitmq]: json unmarshal error:", err)
 		return
 	}
 	proxyListLock.Lock()
-	// MethodID: 0(add), 1(delete)
+	defer proxyListLock.Unlock()
 	if data.MethodID == 0 {
 		proxyList[data.ID] = data.Address
 	} else {
 		delete(proxyList, data.ID)
 	}
-	proxyListLock.Unlock()
 }
 
-func StartZMQ(zmqAddr string, proxyList map[string]string, proxyListLock *sync.RWMutex) {
-	conn := initZmqPull(zmqAddr)
-	for {
-		zmqMsg, _, err := conn.RecvFrame()
-		if err != nil {
-			fmt.Println("[zmq recv]:", err)
-			time.Sleep(time.Second) // 防止CPU空转
-			continue
-		}
-		handleZmqMsg(zmqMsg, proxyList, proxyListLock)
+func StartZMQ(rabbitURL, queueName string, proxyList map[string]string, proxyListLock *sync.RWMutex) {
+	conn, ch, msgs, err := initRabbitConsumer(rabbitURL, queueName)
+	if err != nil {
+		fmt.Println("[rabbitmq]:", err)
+		return
+	}
+	defer conn.Close()
+	defer ch.Close()
+
+	fmt.Println("[rabbitmq]: waiting for messages...")
+
+	for d := range msgs {
+		// fmt.Printf("[rabbitmq] received: %s\n", d.Body)
+		handleRabbitMsg(d.Body, proxyList, proxyListLock)
 	}
 }