fix: #72 cpu占用率100%

.gitignore

@@ -21,4 +21,7 @@ trace.out
 cmd/tc/cmd
 cmd/tc/tc
 cmd/tc/trace.out
-dist/tc
+dist/tc
+dist/cmd
+dist/seata.log
+examples/

pkg/client/client.go

@@ -1,7 +1,9 @@
 package client
 
 import (
-	"log"
+	"context"
+	"fmt"
+	"time"
 
 	"google.golang.org/grpc"
 
@@ -15,11 +17,15 @@ import (
 // Init init resource manager，init transaction manager, expose a port to listen tc
 // call back request.
 func Init(config *config.Configuration) {
-	conn, err := grpc.Dial(config.ServerAddressing,
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+	defer cancel()
+
+	conn, err := grpc.DialContext(ctx, config.ServerAddressing,
 		grpc.WithInsecure(),
+		grpc.WithBlock(),
 		grpc.WithKeepaliveParams(config.GetClientParameters()))
 	if err != nil {
-		log.Fatalf("did not connect: %v", err)
+		panic(fmt.Errorf("did not connect: %v", err))
 	}
 
 	resourceManagerClient := apis.NewResourceManagerServiceClient(conn)

pkg/client/rm/resource_manager.go

@@ -72,13 +72,21 @@ func GetResourceManager() *ResourceManager {
 }
 
 func (manager *ResourceManager) branchCommunicate() {
+	var count int
+
 	for {
 		ctx := metadata.AppendToOutgoingContext(context.Background(), "addressing", manager.addressing)
 		stream, err := manager.rpcClient.BranchCommunicate(ctx)
-		if err != nil {
+		if err != nil && count < 3 {
+			count++
+			log.Info(err)
 			continue
 		}
 
+		if count == 3 {
+			panic(err)
+		}
+
 		done := make(chan bool)
 		runtime.GoWithRecover(func() {
 			for {
@@ -92,8 +100,6 @@ func (manager *ResourceManager) branchCommunicate() {
 					if err != nil {
 						return
 					}
-				default:
-					continue
 				}
 			}
 		}, nil)

pkg/tc/server/callback_message_queue.go

@@ -10,27 +10,35 @@ type CallbackMessageQueue struct {
 	lock *sync.Mutex
 
 	queue []*apis.BranchMessage
+
+	notify chan struct{}
 }
 
 func NewCallbackMessageQueue() *CallbackMessageQueue {
 	return &CallbackMessageQueue{
-		queue: make([]*apis.BranchMessage, 0),
-		lock:  &sync.Mutex{},
+		queue:  make([]*apis.BranchMessage, 0),
+		lock:   &sync.Mutex{},
+		notify: make(chan struct{}, 3),
 	}
 }
 
 func (p *CallbackMessageQueue) Enqueue(msg *apis.BranchMessage) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 	p.queue = append(p.queue, msg)
+
+	p.notify <- struct{}{}
 }
 
 func (p *CallbackMessageQueue) Dequeue() *apis.BranchMessage {
 	p.lock.Lock()
 	defer p.lock.Unlock()
+
+	<-p.notify
 	if len(p.queue) == 0 {
 		return nil
 	}
+
 	var msg *apis.BranchMessage
 	msg, p.queue = p.queue[0], p.queue[1:]
 	return msg