edgar1992 opened a new issue #496:
URL: https://github.com/apache/rocketmq-client-go/issues/496
Here is my consumer code, it will hang after a few minutes.
func main() {
count := 0
c, err := rocketmq.NewPushConsumer(
consumer.WithNameServer([]string{"0.0.0.0:9876"}),
consumer.WithConsumerModel(consumer.Clustering),
consumer.WithConsumeFromWhere(consumer.ConsumeFromLastOffset),
consumer.WithGroupName("GID_XXXXXX"),
)
if err != nil {
fmt.Println(err)
}
err = c.Subscribe("test", consumer.MessageSelector{}, func(ctx
context.Context,
msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
count = count + 1
fmt.Printf("count: %d \n", count)
for _, msg := range msgs {
fmt.Printf("Received on topic: %v, content: %v\n",
msg.Topic, string(msg.Body))
}
return consumer.ConsumeSuccess, nil
})
if err != nil {
fmt.Println(err)
}
err = c.Start()
if err != nil {
fmt.Println(err)
}
for {
}
}
Here is my producer code, it send msg to the test topic every second.
func main() {
p, err := rocketmq.NewProducer(
producer.WithNameServer([]string{"0.0.0.0:9876"}),
producer.WithRetry(2),
producer.WithGroupName("GID_xxxxxx"),
producer.WithCreateTopicKey("test"),
)
if err != nil {
fmt.Println(err)
}
err = p.Start()
if err != nil {
fmt.Println(err)
}
for {
_, err = p.SendSync(context.Background(), &primitive.Message{
Topic: "test",
Body: []byte("Hello RocketMQ Go Client!"),
})
if err != nil {
fmt.Println(err)
}
time.Sleep(time.Second)
}
}
go.mod
go 1.13
require (
github.com/apache/rocketmq-client-go v1.2.4
github.com/apache/rocketmq-client-go/v2 v2.0.0
)
rocketmq server 4.5.0
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
