go-micro broker能力

结论

micro.RegisterSubscriber 和 micro.NewPublisher 配对使用, pub.Publish 第二参数传递对应 proto的 struct broker.Subscribe 和 broker.Publish配对使用, broker.Publish 第二参数传递broker.Message

micro 封装实例

创建go.micro.hello.micro.srv.log srv

package main

import (
	"context"

	logProto "github.com/cnbattle/hello-micro/proto/log"
	"github.com/micro/go-micro"
	"github.com/micro/go-micro/util/log"
)

type Sub struct {}

func (s Sub) Process(ctx context.Context, evt *logProto.LogEvt) error {
	// 业务逻辑
	log.Logf("[sub] 收到日志: %v", evt.Msg)
	return nil
}

func main() {
	service := micro.NewService(
		micro.Name("go.micro.hello.micro.srv.log"),
	)
	service.Init()
	_ = micro.RegisterSubscriber("go.micro.hello.micro.topic.log", service.Server(), &Sub{})
	if err := service.Run(); err != nil {
		panic(err)
	}
}

调用

pub := micro.NewPublisher("go.micro.hello.micro.topic.log", cli)
			_ = pub.Publish(ctx, &logProto.LogEvt{
				Msg: "Hello Friends",
			})

broker 封装实例

创建

package main

import (
	"github.com/micro/go-micro"
	"github.com/micro/go-micro/broker"
	"github.com/micro/go-micro/util/log"
)

func Handler(event broker.Event) error {
	log.Logf("[sub] 收到消息，请查收: %v, %v, %v", event.Message().Header, event.Message().Header["serviceName"], string(event.Message().Body))
	return nil
}

func main() {
	// New Service
	service := micro.NewService(
		micro.Name("go.micro.hello.micro.srv.log2"),
		micro.Version("latest"),
	)

	// Initialise service
	service.Init()

	broker := service.Server().Options().Broker

	if err := broker.Init(); err != nil {
		log.Fatalf("Broker Init error: %v", err)
	}
	if err := broker.Connect(); err != nil {
		log.Fatalf("Broker Connect error: %v", err)
	}

	if _, err := broker.Subscribe("go.micro.hello.micro.topic.event", Handler); err != nil {
		log.Fatalf("broker.Subscribe error: %v", err)
	}

	// Run service
	if err := service.Run(); err != nil {
		log.Fatal(err)
	}
}

调用

event := log2.Event{
				Id:        uuid.New().String(),
				Timestamp: time.Now().Unix(),
				Message:   fmt.Sprintf("如果你看到了消息 %s, '那是因为我一直爱着你", "hello"),
			}
			body, _ := json.Marshal(event)
			_ = broker.Publish("go.micro.hello.micro.topic.event", &broker.Message{
				Header: map[string]string{
					"serviceName": "hello-srv",
				},
				Body: body,
			})