自动创建Rocket topic

v2/brokers/rocketmq.go

@@ -2,6 +2,7 @@ package brokers
 
 import (
 	"fmt"
+
 	"github.com/gojuukaze/YTask/v2/drive"
 	"github.com/gojuukaze/YTask/v2/message"
 	"github.com/gojuukaze/YTask/v2/util/yjson"
@@ -11,31 +12,35 @@ import (
 
 type RocketMqBroker struct {
 	client   *drive.RocketMqClient
-	host     string
-	port     string
+	namesrvAddr []string
+	brokerAddr []string
+	auto bool
 }
 
-func NewRocketMqBroker(host, port string) RocketMqBroker {
+func NewRocketMqBroker(namesrvAddr []string,brokerAddr... []string) RocketMqBroker {
 	 /*
-	    1、目前不能自动创建topic (mqadmin手动创建，并设置读写队列数为1)
-	    2、rocketmq topic名称不允许存在 ‘:’ ,
-	    (所以在生产、消费前先做了名称转换topic RocketMqClient.topicChecker 将非法字符全部转换为 ‘_’)
-		3、为提供pullConsumer实现，所以添加了在worker和consumer之间添加了 RocketMqClient.MsgChan
-	    4、consumerOffset不能同步更新，所以任务执行时间更长
+	    FIX：1、目前不能自动创建topic (mqadmin手动创建，并设置读写队列数为1)
+	    2、consumerOffset不能同步更新，所以任务执行时间更长
 	    (需要将队列中多余的message消费掉才能消费到当前taskId对应的消息)
-	    5、未支持RocketMqBroker.LSend
-	    6、rockemq日志级别设置,设置环境变量"ROCKETMQ_GO_LOG_LEVEL"="error"||"info"||“debug”||...
+	    3、未支持RocketMqBroker.LSend
 	 */
+	var auto bool
+	if len(brokerAddr)>0 {
+		auto=true
+	}
 	return RocketMqBroker{
-		host:     host,
-		port:     port,
-		//poolSize: 0,
+		namesrvAddr: namesrvAddr,
+		brokerAddr: brokerAddr[0],
+		auto: auto,
 	}
 }
 func (r *RocketMqBroker) Activate() {
-
-	client := drive.NewRocketMqClient(r.host, r.port)
+	client := drive.NewRocketMqClient(
+		drive.WithNameSrvAddr(r.namesrvAddr),
+		drive.WithBrokerAddr(r.brokerAddr),
+		drive.WithAutoCreateTopic(r.auto))
 	r.client = &client
+
 }
 
 func (r *RocketMqBroker) SetPoolSize(n int) {
@@ -58,12 +63,11 @@ func (r *RocketMqBroker) Next(topic string) (message.Message, error) {
 	select {
 	case value=<-queue:
 
-	case <-time.After(2 * time.Second):
+	case <-time.After(5 * time.Second):
 		return msg,yerrors.ErrEmptyQuery{}
 	}
 
 	err = yjson.YJson.UnmarshalFromString(value, &msg)
-
 	return msg, err
 }
 
@@ -92,28 +96,36 @@ func (r *RocketMqBroker) LSend(queueName string, msg message.Message) error {
 func (r RocketMqBroker) Clone() BrokerInterface {
 
 	return &RocketMqBroker{
-		host:     r.host,
-		port:     r.port,
-
-		//poolSize: 0,
+		namesrvAddr: r.namesrvAddr,
+		brokerAddr: r.brokerAddr,
+		auto: r.auto,
 	}
 }
+//目前不做使用
 func (r RocketMqBroker)Shutdown(){
-	for topic,producer:=range r.client.RocketMqProducerMap{
-		TRY1:
-		err:=producer.Shutdown()
-		if err !=nil{
-			fmt.Println(topic,"producer shutdown err",err)
-			goto TRY1
-		}
+	TRY1:
+	err:=r.client.Producer.Shutdown()
+	if err !=nil{
+		fmt.Println("YTask[RocketMQ]: producer shutdown err:",err)
+		goto TRY1
 	}
-	for topic,consumer:=range r.client.RocketMqConsumerMap{
-		close(r.client.MsgChanMap[topic])
+	for topic,consumer:=range r.client.ConsumerMap{
+		err:=consumer.Unsubscribe(topic)
+		if err!=nil {
+			fmt.Println("YTask[RocketMQ]: Unsubscribe err: ",err)
+		}
 		TRY2:
-		err:=consumer.Shutdown()
+		err=consumer.Shutdown()
 		if err !=nil{
-			fmt.Println(topic,"consumer shutdown err",err)
+			fmt.Println(topic,"YTask[RocketMQ]: consumer shutdown err: ",err)
 			goto TRY2
 		}
+		close(r.client.MsgChanMap[topic])
+
+		r.client.TopicDeleter(topic)
+		//consumer.Shutdown()方法没法及时同步，所以在异步任务结束后删除topic
+		//重新开启任务时创建topic,代理点位和消费点位重置为0
+		//不得已为之，待改善
 	}
+	r.client.Admin.Close()
 }

v2/drive/rocketmq.go

@@ -4,61 +4,137 @@ import (
 	"context"
 	"fmt"
 	"github.com/apache/rocketmq-client-go/v2"
-	"github.com/apache/rocketmq-client-go/v2/producer"
-	"regexp"
+	"github.com/apache/rocketmq-client-go/v2/admin"
 	"github.com/apache/rocketmq-client-go/v2/consumer"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/apache/rocketmq-client-go/v2/producer"
+	"regexp"
+	"sync"
 )
 
 type RocketMqClient struct {
-	addr primitive.NamesrvAddr
-	group string
-	RocketMqProducerMap map[string]rocketmq.Producer
-	RocketMqConsumerMap map[string]rocketmq.PushConsumer
-	MsgChanMap map[string]chan string
+	options *clientOptions
+	Producer rocketmq.Producer
+	ConsumerMap map[string]rocketmq.PushConsumer
+ 	MsgChanMap map[string]chan string
+	Admin admin.Admin
+	topicMap sync.Map
+
+}
+type clientOptions struct {
+	NamesrvAddr primitive.NamesrvAddr
+	AutoCreateTopic bool
+	BrokerAddr []string
+	auto bool
+}
+type ClientOption func(options *clientOptions)
+
+func defaultAdminOptions() *clientOptions {
+	return &clientOptions{}
+}
+func WithNameSrvAddr(addr []string) ClientOption{
+	return func(opts *clientOptions) {
+		opts.NamesrvAddr=addr
+	}
+}
+func WithBrokerAddr(addr []string) ClientOption{
+	return func(opts *clientOptions) {
+		opts.BrokerAddr=addr
+	}
+}
+func WithAutoCreateTopic(auto bool) ClientOption {
+	return func(opts *clientOptions) {
+		opts.auto=auto
+	}
 }
 
-func NewRocketMqClient(host,port string) RocketMqClient{
 
+func NewRocketMqClient(opts... ClientOption) RocketMqClient{
+	defaultOpts := defaultAdminOptions()
+	for _, opt := range opts {
+		opt(defaultOpts)
+	}
+	var adm admin.Admin
 	var err error
-	addr,err:=primitive.NewNamesrvAddr(host+":"+port)
+
+	adm, err = admin.NewAdmin(admin.WithResolver(
+		primitive.NewPassthroughResolver(defaultOpts.NamesrvAddr)))
 	if err!=nil {
-		panic("YTask: rocketMq error : " + err.Error())
+		panic("YTask: admin create error : "+err.Error())
+	}
+
+	input, err := rocketmq.NewProducer(
+		producer.WithNameServer(defaultOpts.NamesrvAddr),
+		//producer.WithCreateTopicKey(topic),
+	)
+	if err!=nil {
+		panic("YTask[RockerMQ]: Producer create error : " + err.Error())
+	}
+	err=input.Start()
+	if err!=nil {
+		panic("YTask[RockerMQ]: Producer start error : " +err.Error())
 	}
 	return RocketMqClient{
-		addr:addr,
-		RocketMqProducerMap: make(map[string]rocketmq.Producer) ,
-		RocketMqConsumerMap: make(map[string]rocketmq.PushConsumer),
+		options: defaultOpts,
+		Producer: input,
+		ConsumerMap: make(map[string]rocketmq.PushConsumer),
 		MsgChanMap: make(map[string]chan string),
+		Admin: adm,
+	}
+}
+func (c *RocketMqClient) topicCreator(topic string) {
+	if c.options.BrokerAddr == nil {
+		return
+	}
+	//create topic
+	for _, addr := range c.options.BrokerAddr {
+		err := c.Admin.CreateTopic(
+			context.Background(),
+			admin.WithTopicCreate(topic),
+			admin.WithBrokerAddrCreate(addr),
+			admin.WithReadQueueNums(1),
+			admin.WithWriteQueueNums(1),
+			admin.WithPerm(6),
+		)
+		if err != nil {
+			fmt.Println("YTask[RocketMQ]: create topic error:", err.Error())
+		}
+	}
+}
+func (c *RocketMqClient) TopicDeleter(topic string) {
+	//delete topic
+	err:=c.Admin.DeleteTopic(
+		context.Background(),
+		admin.WithTopicDelete(topic),
+	)
+	if err != nil {
+		fmt.Println("YTask[RocketMQ]: delete topic error:", err.Error())
 	}
 }
-
 
 func (c *RocketMqClient) topicChecker(topic string)(string) {
 	//rocketmq topic 只能包含%数字大小写字母及下划线和中划线
 	re := regexp.MustCompile("[^A-z0-9_-]")
 	//所以用下划线替换非法字符
 	return re.ReplaceAllString(topic, "_")
 }
-func (c *RocketMqClient) Register(topic string) (<-chan string,error){
-	topic=c.topicChecker(topic)
-
 
-	if _,ok:=c.MsgChanMap[topic];!ok{
+func (c *RocketMqClient) Register(topic string)(<-chan string,error){
+	topic=c.topicChecker(topic)
+	if _,ok:=c.ConsumerMap[topic];!ok{
+		if _,ok:=c.topicMap.LoadOrStore(topic,1);!ok {
+			if c.options.auto {
+				c.topicCreator(topic)
+			}
+		}
 		c.MsgChanMap[topic]=make(chan string,0)
 		output,err:=rocketmq.NewPushConsumer(
-			consumer.WithNameServer(c.addr),
+			consumer.WithNameServer(c.options.NamesrvAddr),
 			consumer.WithGroupName(topic),
 		)
-		c.RocketMqConsumerMap[topic]=output
-		/*addr,_:=internal.NewNamesrv(c.addr)
-		options:=internal.ClientOptions{
-			GroupName: topic,
-			Namesrv: addr,
+		if err!=nil {
+			panic("YTask[RockerMQ]: Consumer create error : " + err.Error())
 		}
-		callBackChan:=make(chan interface{})
-		client:=internal.GetOrNewRocketMQClient(options,callBackChan)
-		offset:=consumer.NewRemoteOffsetStore(topic,client,addr)*/
 		output.Subscribe(topic, consumer.MessageSelector{}, func(ctx context.Context,
 			msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
 
@@ -69,40 +145,26 @@ func (c *RocketMqClient) Register(topic string) (<-chan string,error){
 			return consumer.ConsumeSuccess, nil
 		})
 		err=output.Start()
-		if err!=nil{
-			fmt.Println("consumer start error ",err.Error())
+		if err!=nil {
+			panic("YTask[RockerMQ]: Consumer start error : " +err.Error())
 		}
-		return c.MsgChanMap[topic],err
+		c.ConsumerMap[topic]=output
+		return c.MsgChanMap[topic],nil
 	}
-	//pull方式未实现
-	//ref:=reflect.ValueOf(c.rocketMqConsumer)
-	//method:=ref.MethodByName("Pull")
-	//args:=[]reflect.Value{reflect.ValueOf(context.Background()),
-	//	reflect.ValueOf(topic),
-	//	reflect.ValueOf(consumer.MessageSelector{}),
-	//	reflect.ValueOf(1)}
-	//result:=method.Call(args)
-	//res,err:=result[0].Interface().((*primitive.PullResult)),result[1].Interface().(error)
+
 	return c.MsgChanMap[topic],nil
 }
-func (c *RocketMqClient) Publish(topic string,value interface{}, Priority uint8) error {
 
-	topic=c.topicChecker(topic)
-	if _,ok:=c.RocketMqProducerMap[topic];!ok{
-		input, err := rocketmq.NewProducer(
-			producer.WithNameServer(c.addr),
-			producer.WithCreateTopicKey(topic),
-			producer.WithGroupName(topic),
-		)
-		err=input.Start()
-		if err!=nil {
-			panic("YTask: rocketMq error : " + err.Error())
-			return err
+
+func (c *RocketMqClient) Publish(topic string,value interface{}, Priority uint8) error {
+	if _,ok:=c.topicMap.LoadOrStore(topic,1);!ok {
+		if c.options.auto {
+			c.topicCreator(topic)
 		}
-		c.RocketMqProducerMap[topic]=input
 	}
-	fmt.Println("product:",string(value.([]byte)))
-	_, err := c.RocketMqProducerMap[topic].SendSync(context.Background(),
+	topic=c.topicChecker(topic)
+	fmt.Println("produce:",string(value.([]byte)))
+	_, err := c.Producer.SendSync(context.Background(),
 		primitive.NewMessage(topic,value.([]byte)))
 	if err != nil {
 		return err

v2/go.mod

@@ -3,7 +3,7 @@ module github.com/gojuukaze/YTask/v2
 go 1.12
 
 require (
-	github.com/apache/rocketmq-client-go/v2 v2.0.0
+	github.com/apache/rocketmq-client-go/v2 v2.1.0-rc5.0.20201102074636-e1d9be806c18
 	github.com/bradfitz/gomemcache v0.0.0-20190913173617-a41fca850d0b
 	github.com/go-redis/redis/v7 v7.4.0
 	github.com/golang/protobuf v1.3.3 // indirect

v2/test/rockemqBroker_test.go

@@ -10,7 +10,9 @@ import (
 
 func TestRocketMqBroker(t *testing.T) {
 
-	broker := brokers.NewRocketMqBroker("127.0.0.1", "9876")
+
+	broker := brokers.NewRocketMqBroker([]string{"127.0.0.1:9876"},[]string{"127.0.0.1:10911"})
+
 	broker.Activate()
 	//broker.Shutdown()主要是为了关闭consumer,同步offset到broker
 	//BUG：会出现同步失败
@@ -47,7 +49,10 @@ func TestRocketMqBroker(t *testing.T) {
 }
 
 func TestRocketMqBrokerLSend(t *testing.T) {
-	broker := brokers.NewRocketMqBroker("127.0.0.1", "9876")
+	broker := brokers.NewRocketMqBroker(
+		[]string{"127.0.0.1:9876"},
+	[]string{"127.0.0.1:10911"})
+
 	broker.Activate()
 	defer broker.Shutdown()
 	msg := message.NewMessage(controller.NewTaskCtl())

v2/test/ytask_test.go

@@ -101,6 +101,7 @@ func testWorker1(ser server.Server, t *testing.T) {
 	if !result.IsSuccess() {
 		t.Fatal("result is not success")
 	}
+
 }
 
 func testWorker2(ser server.Server, t *testing.T) {

v2/ytask.go

@@ -37,8 +37,8 @@ func (i iBroker) NewRabbitMqBroker(host, port, user, password, vhost string) bro
 	return brokers.NewRabbitMqBroker(host, port, user, password, vhost)
 }
 
-func (i iBroker) NewRocketMqBroker(host, port string) brokers.RocketMqBroker {
-	return brokers.NewRocketMqBroker(host, port)
+func (i iBroker) NewRocketMqBroker(namesrvAddr []string,brokerAddr... []string) brokers.RocketMqBroker {
+	return brokers.NewRocketMqBroker(namesrvAddr,brokerAddr...)
 }
 
 type iConfig struct {