You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "majintao (via GitHub)" <gi...@apache.org> on 2024/02/06 11:41:28 UTC
[I] User pushConsumer，Reach max retry it auto consumed... [rocketmq-client-go]
majintao opened a new issue, #1129:
URL: https://github.com/apache/rocketmq-client-go/issues/1129
`package rocket_mq
import (
"context"
"fmt"
"github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/consumer"
"github.com/apache/rocketmq-client-go/v2/primitive"
"github.com/apache/skywalking-go/toolkit/trace"
)
type ProcFunc func(context.Context, []byte) error
type ConsumerProcessor struct {
opts []consumer.Option
selector consumer.MessageSelector
topic string
procFunc ProcFunc
consumer rocketmq.PushConsumer
}
func NewConsumerProcessor(topic string, procFunc ProcFunc, opts ...consumer.Option) *ConsumerProcessor {
return &ConsumerProcessor{
opts: opts,
topic: topic,
procFunc: procFunc,
}
}
func (cp *ConsumerProcessor) Start() error {
// 启动消费
c, err := rocketmq.NewPushConsumer(cp.opts...)
if err != nil {
return err
}
cp.consumer = c
c.Subscribe(cp.topic, cp.selector, func(ctx context.Context, messages ...*primitive.MessageExt) (
consumer.ConsumeResult, error) {
// 处理消息
for _, message := range messages {
err := cp.process(ctx, message)
if err != nil {
fmt.Printf("messageId:%s", message.MsgId)
return consumer.ConsumeRetryLater, nil
}
}
return consumer.ConsumeSuccess, nil
})
err = c.Start()
if err != nil {
return err
}
return nil
}
func (cp *ConsumerProcessor) process(ctx context.Context, message *primitive.MessageExt) error {
// 添加链路追踪支持
_, err := trace.CreateLocalSpan("")
if err != nil {
}
defer trace.StopSpan()
err = cp.procFunc(ctx, message.Body)
if err != nil {
return err
}
return nil
}
func (cp *ConsumerProcessor) Stop() {
// 关闭消费
cp.consumer.Shutdown()
}
`
`
topic := "test_topic"
rocket_mq.NewConsumerProcessor(topic, func(c context.Context, msg []byte) error {
xflog.Infof("msg:%v", string(msg))
return errors.New("111")
}, consumer.WithRetry(-1), consumer.WithAutoCommit(false), consumer.WithGroupName("abc"),
consumer.WithNameServer([]string{"10.34.22.164:9876", "10.34.22.167:9876"})).Start()
`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [BUG] 使用 pushconsumer 客户端，默认重试两次后就commit，消息状态变成consumed [rocketmq-client-go]
Posted by "ShannonDing (via GitHub)" <gi...@apache.org>.
ShannonDing closed issue #1129: [BUG] 使用 pushconsumer 客户端，默认重试两次后就commit，消息状态变成consumed
URL: https://github.com/apache/rocketmq-client-go/issues/1129
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [BUG] 使用 pushconsumer 客户端，默认重试两次后就commit，消息状态变成consumed [rocketmq-client-go]
Posted by "majintao (via GitHub)" <gi...@apache.org>.
majintao commented on issue #1129:
URL: https://github.com/apache/rocketmq-client-go/issues/1129#issuecomment-1951146922
@twz915
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [BUG] 使用 pushconsumer 客户端，默认重试两次后就commit，消息状态变成consumed [rocketmq-client-go]
Posted by "caichunlong (via GitHub)" <gi...@apache.org>.
caichunlong commented on issue #1129:
URL: https://github.com/apache/rocketmq-client-go/issues/1129#issuecomment-1987731391
@twz915
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [BUG] 使用 pushconsumer 客户端，默认重试两次后就commit，消息状态变成consumed [rocketmq-client-go]
Posted by "twz915 (via GitHub)" <gi...@apache.org>.
twz915 commented on issue #1129:
URL: https://github.com/apache/rocketmq-client-go/issues/1129#issuecomment-1991018271
我测试了，工作正常，和预期一致。你可以检查一下 metaq 服务端有没有什么设置只允许重试两次。
我使用的版本是 master 分支
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org