Optimizing internal rpc

dobyte · dobyte · commit e6dff3bb82bb · 2025-12-17T20:38:28.000+08:00
diff --git a/cache/memcache/go.mod b/cache/memcache/go.mod
@@ -4,7 +4,7 @@ go 1.23.0
 
 require (
 	github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	golang.org/x/sync v0.13.0
 )
 
diff --git a/cache/redis/go.mod b/cache/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/cache/redis/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/go-redis/redis/v8 v8.11.5
 	golang.org/x/sync v0.13.0
 )
diff --git a/component/http/go.mod b/component/http/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/component/http/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/go-openapi/runtime v0.28.0
 	github.com/gofiber/fiber/v3 v3.0.0-beta.4
 )
diff --git a/config/consul/go.mod b/config/consul/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/config/consul/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/hashicorp/consul/api v1.32.0
 )
 
diff --git a/config/etcd/go.mod b/config/etcd/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/config/etcd/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	go.etcd.io/etcd/api/v3 v3.5.21
 	go.etcd.io/etcd/client/v3 v3.5.21
 )
diff --git a/config/nacos/go.mod b/config/nacos/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/config/nacos/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/nacos-group/nacos-sdk-go/v2 v2.2.7
 )
 
diff --git a/core/info/info.go b/core/info/info.go
@@ -27,7 +27,7 @@ const (
 	leftBottomBorder  = "└"
 	rightBottomBorder = "┘"
 	website           = "https://github.com/dobyte/due"
-	version           = "v2.4.2"
+	version           = "v2.4.3"
 	global            = "Global"
 )
 
diff --git a/crypto/ecc/go.mod b/crypto/ecc/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/crypto/ecc/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/ethereum/go-ethereum v1.12.0
 )
 
diff --git a/crypto/rsa/go.mod b/crypto/rsa/go.mod
@@ -2,7 +2,7 @@ module github.com/dobyte/due/crypto/rsa/v2
 
 go 1.23.0
 
-require github.com/dobyte/due/v2 v2.4.2
+require github.com/dobyte/due/v2 v2.4.3
 
 require (
 	dario.cat/mergo v1.0.1 // indirect
diff --git a/eventbus/kafka/go.mod b/eventbus/kafka/go.mod
@@ -4,7 +4,7 @@ go 1.23.0
 
 require (
 	github.com/IBM/sarama v1.43.3
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 )
 
 require (
diff --git a/eventbus/nats/go.mod b/eventbus/nats/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/eventbus/nats/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/nats-io/nats.go v1.37.0
 )
 
diff --git a/eventbus/redis/go.mod b/eventbus/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/eventbus/redis/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/go-redis/redis/v8 v8.11.5
 )
 
diff --git a/internal/transporter/gate/builder.go b/internal/transporter/gate/builder.go
@@ -1,6 +1,7 @@
 package gate
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -51,6 +52,7 @@ func (b *Builder) Build(addr string) (*Client, error) {
 			InsID:   b.opts.InsID,
 			InsKind: b.opts.InsKind,
 			CloseHandler: func() {
+				fmt.Printf("client %s closed----------------------------------------\n", addr)
 				b.faults.Store(addr, xtime.Now())
 				b.clients.Delete(addr)
 			},
diff --git a/internal/transporter/internal/client/client.go b/internal/transporter/internal/client/client.go
@@ -24,19 +24,19 @@ type chWrite struct {
 }
 
 type Client struct {
-	opts            *Options       // 配置
-	connections     []*Conn        // 连接
-	disorderlyQueue chan *chWrite  // 无序队列
-	wg              sync.WaitGroup // 等待组
-	closed          atomic.Bool    // 已关闭
-	pool            sync.Pool      // 连接池
+	opts   *Options       // 配置
+	conns  []*Conn        // 连接
+	queue  chan *chWrite  // 无序队列
+	wg     sync.WaitGroup // 等待组
+	closed atomic.Bool    // 已关闭
+	pool   sync.Pool      // 连接池
 }
 
 func NewClient(opts *Options) *Client {
 	c := &Client{}
 	c.opts = opts
-	c.connections = make([]*Conn, 0, defaultConnNum)
-	c.disorderlyQueue = make(chan *chWrite, 10240)
+	c.conns = make([]*Conn, 0, defaultConnNum)
+	c.queue = make(chan *chWrite, 10240)
 	c.closed.Store(true)
 	c.pool = sync.Pool{New: func() any { return &chWrite{} }}
 
@@ -56,22 +56,22 @@ func (c *Client) Establish() error {
 
 	for range defaultConnNum {
 		eg.Go(func() error {
-			conn := newConn(c, c.disorderlyQueue)
+			conn := newConn(c)
 
 			if err := conn.dial(); err != nil {
 				log.Warnf("conn dial failed: %v", err)
 				return err
 			}
 
 			mu.Lock()
-			c.connections = append(c.connections, conn)
+			c.conns = append(c.conns, conn)
 			mu.Unlock()
 
 			return nil
 		})
 	}
 
-	if err := eg.Wait(); err != nil && len(c.connections) == 0 {
+	if err := eg.Wait(); err != nil && len(c.conns) == 0 {
 		return err
 	}
 
@@ -139,9 +139,9 @@ func (c *Client) Send(ctx context.Context, buf *buffer.NocopyBuffer, idx ...int6
 // 获取连接
 func (c *Client) load(idx ...int64) *Conn {
 	if len(idx) > 0 {
-		return c.connections[idx[0]%int64(len(c.connections))]
+		return c.conns[idx[0]%int64(len(c.conns))]
 	} else {
-		return c.connections[0]
+		return c.conns[0]
 	}
 }
 
@@ -170,7 +170,7 @@ func (c *Client) wait() {
 	c.closed.Store(true)
 
 	time.AfterFunc(time.Second, func() {
-		close(c.disorderlyQueue)
+		close(c.queue)
 	})
 
 	if c.opts.CloseHandler != nil {
diff --git a/internal/transporter/internal/client/conn.go b/internal/transporter/internal/client/conn.go
@@ -23,20 +23,18 @@ type Conn struct {
 	cli               *Client            // 客户端
 	state             atomic.Int32       // 连接状态
 	pending           *pending           // 等待队列
-	orderlyQueue      chan *chWrite      // 有序队列
-	disorderlyQueue   chan *chWrite      // 无序队列
+	queue             chan *chWrite      // 有序队列
 	ctx               context.Context    // 上下文
 	cancel            context.CancelFunc // 取消函数
 	lastHeartbeatTime atomic.Int64       // 上次心跳时间
 }
 
-func newConn(cli *Client, queue chan *chWrite) *Conn {
+func newConn(cli *Client) *Conn {
 	c := &Conn{}
 	c.cli = cli
 	c.state.Store(def.ConnHanged)
+	c.queue = make(chan *chWrite, 4096)
 	c.pending = newPending()
-	c.orderlyQueue = make(chan *chWrite, 4096)
-	c.disorderlyQueue = queue
 
 	return c
 }
@@ -88,9 +86,9 @@ func (c *Conn) send(ch *chWrite, isOrderly ...bool) error {
 	}
 
 	if len(isOrderly) > 0 && isOrderly[0] {
-		c.orderlyQueue <- ch
+		c.queue <- ch
 	} else {
-		c.disorderlyQueue <- ch
+		c.cli.queue <- ch
 	}
 
 	return nil
@@ -202,15 +200,15 @@ func (c *Conn) write(conn net.Conn) {
 					return
 				}
 			}
-		case ch, ok := <-c.orderlyQueue: // 有序队列
+		case ch, ok := <-c.queue: // 有序队列
 			if !ok {
 				return
 			}
 
 			if ok = c.doWrite(conn, ch); !ok {
 				return
 			}
-		case ch, ok := <-c.disorderlyQueue: // 无序队列
+		case ch, ok := <-c.cli.queue: // 无序队列
 			if !ok {
 				return
 			}
@@ -275,7 +273,7 @@ func (c *Conn) close() {
 	}
 
 	time.AfterFunc(time.Second, func() {
-		close(c.orderlyQueue)
+		close(c.queue)
 	})
 }
 
diff --git a/locate/redis/go.mod b/locate/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/locate/redis/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/go-redis/redis/v8 v8.11.5
 	golang.org/x/sync v0.13.0
 )
diff --git a/lock/memcache/go.mod b/lock/memcache/go.mod
@@ -4,7 +4,7 @@ go 1.23.0
 
 require (
 	github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 )
 
 require (
diff --git a/lock/redis/go.mod b/lock/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/lock/redis/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/go-redis/redis/v8 v8.11.5
 )
 
diff --git a/log/aliyun/go.mod b/log/aliyun/go.mod
@@ -4,7 +4,7 @@ go 1.23.0
 
 require (
 	github.com/aliyun/aliyun-log-go-sdk v0.1.100
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 )
 
 require (
diff --git a/log/tencent/go.mod b/log/tencent/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/log/tencent/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/tencentcloud/tencentcloud-cls-sdk-go v1.0.11
 )
 
diff --git a/network/kcp/go.mod b/network/kcp/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/network/kcp/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/xtaci/kcp-go/v5 v5.6.20
 )
 
diff --git a/network/tcp/go.mod b/network/tcp/go.mod
@@ -2,7 +2,7 @@ module github.com/dobyte/due/network/tcp/v2
 
 go 1.23.0
 
-require github.com/dobyte/due/v2 v2.4.2
+require github.com/dobyte/due/v2 v2.4.3
 
 require (
 	dario.cat/mergo v1.0.1 // indirect
diff --git a/network/ws/go.mod b/network/ws/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/network/ws/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/gorilla/websocket v1.5.3
 )
 
diff --git a/registry/consul/go.mod b/registry/consul/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/consul/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/hashicorp/consul/api v1.32.0
 )
 
diff --git a/registry/etcd/go.mod b/registry/etcd/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/etcd/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	go.etcd.io/etcd/api/v3 v3.5.21
 	go.etcd.io/etcd/client/v3 v3.5.21
 )
diff --git a/registry/nacos/go.mod b/registry/nacos/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/nacos/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/nacos-group/nacos-sdk-go/v2 v2.2.7
 	golang.org/x/sync v0.13.0
 )
diff --git a/transport/grpc/go.mod b/transport/grpc/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/transport/grpc/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	golang.org/x/sync v0.13.0
 	google.golang.org/grpc v1.71.1
 )
diff --git a/transport/rpcx/go.mod b/transport/rpcx/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/transport/rpcx/v2
 go 1.23.0
 
 require (
-	github.com/dobyte/due/v2 v2.4.2
+	github.com/dobyte/due/v2 v2.4.3
 	github.com/smallnest/rpcx v1.9.1
 	golang.org/x/sync v0.13.0
 )

Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ go 1.23.0`
`4`	`4`
`5`	`5`	`require (`
`6`	`6`	`github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf`
`7`		`- github.com/dobyte/due/v2 v2.4.2`
	`7`	`+ github.com/dobyte/due/v2 v2.4.3`
`8`	`8`	`golang.org/x/sync v0.13.0`
`9`	`9`	`)`
`10`	`10`
Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@ module github.com/dobyte/due/cache/redis/v2`
`3`	`3`	`go 1.23.0`
`4`	`4`
`5`	`5`	`require (`
`6`		`- github.com/dobyte/due/v2 v2.4.2`
	`6`	`+ github.com/dobyte/due/v2 v2.4.3`
`7`	`7`	`github.com/go-redis/redis/v8 v8.11.5`
`8`	`8`	`golang.org/x/sync v0.13.0`
`9`	`9`	`)`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ const (`
`27`	`27`	`leftBottomBorder = "└"`
`28`	`28`	`rightBottomBorder = "┘"`
`29`	`29`	`website = "https://github.com/dobyte/due"`
`30`		`- version = "v2.4.2"`
	`30`	`+ version = "v2.4.3"`
`31`	`31`	`global = "Global"`
`32`	`32`	`)`
`33`	`33`
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ go 1.23.0`
`4`	`4`
`5`	`5`	`require (`
`6`	`6`	`github.com/IBM/sarama v1.43.3`
`7`		`- github.com/dobyte/due/v2 v2.4.2`
	`7`	`+ github.com/dobyte/due/v2 v2.4.3`
`8`	`8`	`)`
`9`	`9`
`10`	`10`	`require (`