增加分区参数

.gitignore

@@ -1 +1 @@
-./proxy
+proxy

serv.go

@@ -36,6 +36,8 @@ var (
 	space            = []byte{' '}
 	addr             = flag.String("addr", ":9000", "http service address")
 	bootstrapServers = flag.String("bootstrap-servers", "localhost:9092", "kafka bootstrap servers")
+	partition        = flag.Int("partition", 0, "kafka tipic partion")
+	wait             = flag.Duration("graceful-timeout", time.Second*15, "the duration for which the server gracefully wait for existing connections to finish - e.g. 15s or 1m")
 	upgrader         = websocket.Upgrader{
 		ReadBufferSize:  1024,
 		WriteBufferSize: 1024,
@@ -90,18 +92,10 @@ func (c *Client) readPump(topic string) {
 	})
 	reader.SetOffset(-1) //lastoffset
 
-	// if err != nil {
-	// 	// fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
-	// 	log.WithError(err).Fatal("Failed to create consumer")
-	// 	os.Exit(1)
-	// }
-
 	log.WithFields(log.Fields{
 		"reader": reader,
 	}).Info("Created ")
 
-	// consumer.SubscribeTopics([]string{topic}, nil)
-
 	run := true
 
 	for run == true {
@@ -111,37 +105,17 @@ func (c *Client) readPump(topic string) {
 			run = false
 			log.Fatal("break ")
 		default:
-			// ev := consumer.Poll(100)
-			// if ev == nil {
-			// 	continue
-			// }
-
-			// switch e := ev.(type) {
-			// case *kafka.Message:
-			// 	log.WithFields(log.Fields{"partition": e.TopicPartition, "message": string(e.Value)}).Info("got message")
-			// 	c.hub.broadcast <- e.Value
-			// 	if e.Headers != nil {
-			// 		log.WithFields(log.Fields{"header": e.Headers}).Info(" with header")
-			// 	}
-			// case kafka.Error:
-			// 	// Errors should generally be considered as informational, the client will try to automatically recover
-			// 	log.WithFields(log.Fields{"error": e}).Error("got error")
-			// default:
-			// 	log.WithFields(log.Fields{"event": e}).Info("ignored")
-			// }
 			for {
 				m, err := reader.ReadMessage(context.Background())
 				if err != nil {
 					log.WithFields(log.Fields{"error": err}).Error("read error")
 				}
-				// fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
 				log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message")
 				c.hub.broadcast <- m.Value
 			}
 		}
 	}
 
-	// consumer.Close()
 }
 
 // writePump pumps messages from the hub to the websocket connection.
@@ -258,8 +232,6 @@ func (h *Hub) run() {
 }
 
 func main() {
-	var wait time.Duration
-	flag.DurationVar(&wait, "graceful-timeout", time.Second*15, "the duration for which the server gracefully wait for existing connections to finish - e.g. 15s or 1m")
 	flag.Parse()
 	r := mux.NewRouter()
 	loggedRouter := handlers.LoggingHandler(os.Stdout, r)
@@ -295,7 +267,7 @@ func main() {
 	<-c
 
 	// Create a deadline to wait for.
-	ctx, cancel := context.WithTimeout(context.Background(), wait)
+	ctx, cancel := context.WithTimeout(context.Background(), *wait)
 	defer cancel()
 	// Doesn't block if no connections, but will otherwise wait
 	// until the timeout deadline.