服务器参数,字符串转换为数组
This commit is contained in:
@@ -6,6 +6,7 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -33,7 +34,6 @@ const (
|
||||
|
||||
var (
|
||||
newline = []byte{'\n'}
|
||||
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")
|
||||
@@ -76,21 +76,17 @@ func (c *Client) readPump(topic string) {
|
||||
log.WithFields(log.Fields{
|
||||
"topic": topic,
|
||||
"bootstrap servers": *bootstrapServers,
|
||||
"partition": partition,
|
||||
"group": group}).Info("create kafka client")
|
||||
|
||||
// consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
|
||||
// "bootstrap.servers": *bootstrapServers,
|
||||
// "group.id": group,
|
||||
// "session.timeout.ms": 6000,
|
||||
// "auto.offset.reset": "latest"})
|
||||
reader := kafka.NewReader(kafka.ReaderConfig{
|
||||
Brokers: []string{*bootstrapServers},
|
||||
Brokers: strings.Split(*bootstrapServers, ","),
|
||||
Topic: topic,
|
||||
Partition: 0,
|
||||
Partition: *partition,
|
||||
MinBytes: 10e1,
|
||||
MaxBytes: 10e6, // 10MB
|
||||
})
|
||||
reader.SetOffset(-1) //lastoffset
|
||||
reader.SetOffset(-1) //latest offset
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"reader": reader,
|
||||
|
||||
Reference in New Issue
Block a user