Merge pull request #5 from luohuixi/main

包装了sarama包的生产消费者

Author: sunjunnan79

example/kafka/kafka.go

@@ -0,0 +1,116 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/IBM/sarama"
+	"github.com/muxi-Infra/muxi-micro/pkg/kafka"
+)
+
+// 同步生产者示例
+func SyncProducer() {
+	p := kafka.NewSyncConfig([]string{"localhost:9092"})
+	//可自行改配置如：（可选）
+	//p.GetConfig().Producer.RequiredAcks = sarama.WaitForAll
+
+	err := p.CreateProducer()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	partition, offset, err := p.SendMessage("study2", "testkey8", "testvalue8")
+	if err != nil {
+		log.Fatal(err)
+	}
+	log.Printf("partition: %d, offset: %d\n", partition, offset)
+
+	err = p.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+// 异步生产者示例
+func AsyncProducer() {
+	p := kafka.NewAsyncConfig([]string{"localhost:9092"})
+	//可自行改配置如：（可选）
+	p.GetConfig().Producer.Return.Successes = true
+
+	err := p.CreateProducer()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	//异步获取成功消息
+	ch := p.GetSuccess()
+	//或 ch:= p.GetError()
+	go func() {
+		for msg := range ch {
+			log.Printf("partition: %d, offset: %d, key: %s, value: %s\n", msg.Partition, msg.Offset, msg.Key, msg.Value)
+		}
+	}()
+
+	p.SendMessage("study2", "testkey5", "testvalue5")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	time.Sleep(time.Second * 5)
+
+	err = p.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+}
+
+// 消费者示例
+func Consumer() {
+	// 先构造 handler 实现 ConsumerGroupHandler 接口
+	// setup, cleanup, consumeclaim 都要用户自行实现
+	fn := func(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+		for msg := range claim.Messages() {
+			log.Println("partition: ", msg.Partition, "offset: ", msg.Offset, "key: ", string(msg.Key), "value: ", string(msg.Value))
+			session.MarkMessage(msg, "")
+		}
+		return nil // 正常退出
+	}
+	handler := kafka.NewHandler(nil, nil, fn)
+
+	// 构造消费者组
+	c := kafka.NewConsumerConfig([]string{"localhost:9092"}, "test-group", []string{"study2"}, handler)
+	//可自行改配置如：（可选）
+	//c.GetConfig().Consumer.Offsets.Initial = sarama.OffsetOldest
+
+	err := c.CreateConsumerGroup()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// 消费消息
+	ctx, cancel := context.WithCancel(context.Background())
+
+	// 取消消费
+	go func() {
+		time.Sleep(time.Second * 10)
+		cancel()
+	}()
+
+	err = c.Consume(ctx)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = c.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func main() {
+	SyncProducer()
+	//AsyncProducer()
+	Consumer()
+}

example/tracer/tracer.go

@@ -0,0 +1,146 @@
+package main
+
+import (
+	"context"
+	"github.com/gin-gonic/gin"
+	"github.com/muxi-Infra/muxi-micro/pkg/tracer"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/examples/helloworld/helloworld"
+	"log"
+	"net"
+)
+
+// 简易的 HelloWorld 实现
+type server struct {
+	helloworld.UnimplementedGreeterServer
+}
+
+func (s *server) SayHello(_ context.Context, in *helloworld.HelloRequest) (*helloworld.HelloReply, error) {
+	log.Printf("Received: %v", in.GetName())
+	return &helloworld.HelloReply{Message: "Hello " + in.GetName()}, nil
+}
+
+func GrpcServer() {
+	// Zipkin
+	config, err := tracer.NewZipkin(
+		"http://localhost:9411/api/v2/spans",
+		"demo_service",
+		"localhost:50051",
+		1,
+	)
+
+	// Jaeger
+	//config , err := tracer.NewJaeger(
+	//	"http://localhost:14268/api/traces",
+	//	"demo_service",
+	//	1,
+	//	)
+
+	// SkyWalking
+	//config , err := tracer.NewSkyWalking(
+	//	"localhost:11800",
+	//	"demo_service",
+	//	"demo_instance", //实例名
+	//	1
+	//	)
+
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer func() {
+		if err := config.Close(); err != nil {
+			log.Println(err)
+		}
+	}()
+
+	s := grpc.NewServer(
+		grpc.ChainUnaryInterceptor(
+			config.ServerInterceptor(),
+		),
+	)
+
+	helloworld.RegisterGreeterServer(s, &server{})
+
+	lis, err := net.Listen("tcp", "0.0.0.0:50051")
+	if err != nil {
+		log.Fatalf("failed to listen: %v", err)
+	}
+
+	if err := s.Serve(lis); err != nil {
+		log.Fatalf("failed to serve: %v", err)
+	}
+}
+
+func GrpcClient() {
+	// 同上
+	config, err := tracer.NewZipkin(
+		"http://localhost:9411/api/v2/spans",
+		"demo_client",
+		"localhost:50051",
+		1,
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer func() {
+		if err := config.Close(); err != nil {
+			log.Println(err)
+		}
+	}()
+
+	conn, err := grpc.NewClient(
+		"localhost:50051",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithChainUnaryInterceptor(
+			config.ClientInterceptor(),
+		),
+	)
+	if err != nil {
+		log.Fatalf("did not connect: %v", err)
+	}
+	defer conn.Close()
+
+	c := helloworld.NewGreeterClient(conn)
+
+	resp, err := c.SayHello(context.Background(), &helloworld.HelloRequest{Name: "MuXi"})
+	if err != nil {
+		log.Fatalf("could not greet: %v", err)
+	}
+	log.Printf("Greeting: %s", resp.Message)
+}
+
+func GinService() {
+	// 同上
+	config, err := tracer.NewZipkin(
+		"http://localhost:9411/api/v2/spans",
+		"demo_gin",
+		"localhost:8081",
+		1,
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer func() {
+		if err := config.Close(); err != nil {
+			log.Println(err)
+		}
+	}()
+
+	r := gin.Default()
+	r.Use(config.GinMiddleware(r))
+
+	r.GET("/hello", func(c *gin.Context) {
+		c.JSON(200, gin.H{"message": "hello world"})
+	})
+	r.Run("0.0.0.0:8081")
+}
+
+func main() {
+	// grpc服务端
+	//GrpcServer()
+	// grpc客户端
+	GrpcClient()
+	// gin
+	//GinService()
+}