每个主题使用一个单独的hub转发消息

.vscode/launch.json

@@ -0,0 +1,17 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [   
+        {
+            "name": "Launch",
+            "type": "go",
+            "request": "launch",
+            "mode": "auto",
+            "program": "${fileDirname}",
+            "env": {},
+            "args": []
+        }
+    ]
+}

serv.go

@@ -42,6 +42,8 @@ var (
 		ReadBufferSize:  1024,
 		WriteBufferSize: 1024,
 	}
+	hubMap = make(map[string]Hub)
+	run    = true
 )
 
 // Client is a middleman between the websocket connection and the hub.
@@ -55,25 +57,8 @@ type Client struct {
 	send chan []byte
 }
 
-// readPump pumps messages from the websocket connection to the hub.
-//
-// The application runs readPump in a per-connection goroutine. The application
-// ensures that there is at most one reader on a connection by executing all
-// reads from this goroutine.
-func (c *Client) readPump(topic string) {
-	defer func() {
-		c.hub.unregister <- c
-		c.conn.Close()
-	}()
-	c.conn.SetReadLimit(maxMessageSize)
-	c.conn.SetReadDeadline(time.Now().Add(pongWait))
-	c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
-
-	sigchan := make(chan os.Signal, 1)
-	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
-
-	// group, err := uuid.Must(uuid.NewV4())
-	group := uuid.NewV4()
+func create(topic string) *kafka.Reader {
+	group, _ := uuid.NewV4()
 
 	log.WithFields(log.Fields{
 		"topic":             topic,
@@ -90,11 +75,26 @@ func (c *Client) readPump(topic string) {
 	})
 	reader.SetOffset(-1) //latest offset
 
-	log.WithFields(log.Fields{
-		"reader": reader,
-	}).Info("Created ")
+	return reader
+}
 
-	run := true
+// readPump pumps messages from the websocket connection to the hub.
+//
+// The application runs readPump in a per-connection goroutine. The application
+// ensures that there is at most one reader on a connection by executing all
+// reads from this goroutine.
+func (c *Client) readPump() {
+	defer func() {
+		c.hub.unregister <- c
+		c.conn.Close()
+		log.WithField("client", c).Warn("exit")
+	}()
+	c.conn.SetReadLimit(maxMessageSize)
+	c.conn.SetReadDeadline(time.Now().Add(pongWait))
+	c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
+
+	sigchan := make(chan os.Signal, 1)
+	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
 
 	for run == true {
 		select {
@@ -102,15 +102,15 @@ func (c *Client) readPump(topic string) {
 			log.WithFields(log.Fields{"signal": sig}).Info(" terminating")
 			run = false
 			log.Fatal("break ")
-		default:
-			for {
-				m, err := reader.ReadMessage(context.Background())
-				if err != nil {
-					log.WithFields(log.Fields{"error": err}).Error("read error")
-				}
-				log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message")
-				c.hub.broadcast <- m.Value
-			}
+			// default:
+			// for {
+			// 	m, err := c.hub.reader.ReadMessage(context.Background())
+			// 	if err != nil {
+			// 		log.WithFields(log.Fields{"error": err}).Error("read error")
+			// 	}
+			// 	log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message")
+			// 	c.hub.broadcast <- m.Value
+			// }
 		}
 	}
 
@@ -163,7 +163,7 @@ func (c *Client) writePump() {
 }
 
 // serveWs handles websocket requests from the peer.
-func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
+func serveWs(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	topic := vars["topic"]
 
@@ -172,18 +172,29 @@ func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
 		log.Error("upgrade connection failed ", err)
 		return
 	}
-	client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
+	h, ok := hubMap[topic]
+	if !ok {
+		log.WithField("topic", topic).Info("create new hub ")
+		h = *newHub(topic)
+		go h.run()
+		hubMap[topic] = h
+	} else {
+		log.Info("join hub")
+	}
+	client := &Client{hub: &h, conn: conn, send: make(chan []byte, 256)}
 	client.hub.register <- client
 
 	// Allow collection of memory referenced by the caller by doing all work in
 	// new goroutines.
 	go client.writePump()
-	go client.readPump(topic)
+	go client.readPump()
 }
 
 // Hub maintains the set of active clients and broadcasts messages to the
 // clients.
 type Hub struct {
+	reader *kafka.Reader
+
 	// Registered clients.
 	clients map[*Client]bool
 
@@ -197,17 +208,30 @@ type Hub struct {
 	unregister chan *Client
 }
 
-func newHub() *Hub {
+func newHub(topic string) *Hub {
 	return &Hub{
+		reader:     create(topic),
 		broadcast:  make(chan []byte),
 		register:   make(chan *Client),
 		unregister: make(chan *Client),
 		clients:    make(map[*Client]bool),
 	}
 }
+func (h *Hub) read() {
+	log.WithField("reader", h.reader).Info("start to read")
+	for run == true {
+		m, err := h.reader.ReadMessage(context.Background())
+		if err != nil {
+			log.WithFields(log.Fields{"error": err}).Error("read error")
+		}
+		log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message")
+		h.broadcast <- m.Value
+	}
+}
 
 func (h *Hub) run() {
-	for {
+	go h.read()
+	for run == true {
 		select {
 		case client := <-h.register:
 			h.clients[client] = true
@@ -242,17 +266,17 @@ func main() {
 		Handler:      loggedRouter, // Pass our instance of gorilla/mux in.
 	}
 
-	hub := newHub()
-	go hub.run()
+	// hub := newHub()
+	// go hub.run()
 	r.HandleFunc("/ws/{topic}", func(w http.ResponseWriter, r *http.Request) {
-		serveWs(hub, w, r)
+		serveWs(w, r)
 	})
 
 	// Run our server in a goroutine so that it doesn't block.
 	go func() {
 		log.WithFields(log.Fields{"address": *addr}).Info("starting server")
 		if err := srv.ListenAndServe(); err != nil {
-			log.WithFields(log.Fields{"error": err}).Error("startup error")
+			log.WithFields(log.Fields{"error": err}).Error("error")
 		}
 	}()