新增监控机制

client/chat/chat.pb.go

@@ -21,6 +21,94 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+type PingRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *PingRequest) Reset() {
+	*x = PingRequest{}
+	mi := &file_proto_chat_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *PingRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PingRequest) ProtoMessage() {}
+
+func (x *PingRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_chat_proto_msgTypes[0]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PingRequest.ProtoReflect.Descriptor instead.
+func (*PingRequest) Descriptor() ([]byte, []int) {
+	return file_proto_chat_proto_rawDescGZIP(), []int{0}
+}
+
+type PingResponse struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Status        string                 `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
+	Timestamp     int64                  `protobuf:"varint,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *PingResponse) Reset() {
+	*x = PingResponse{}
+	mi := &file_proto_chat_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *PingResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PingResponse) ProtoMessage() {}
+
+func (x *PingResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_chat_proto_msgTypes[1]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PingResponse.ProtoReflect.Descriptor instead.
+func (*PingResponse) Descriptor() ([]byte, []int) {
+	return file_proto_chat_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *PingResponse) GetStatus() string {
+	if x != nil {
+		return x.Status
+	}
+	return ""
+}
+
+func (x *PingResponse) GetTimestamp() int64 {
+	if x != nil {
+		return x.Timestamp
+	}
+	return 0
+}
+
 type JoinRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	UserId        string                 `protobuf:"bytes,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
@@ -30,7 +118,7 @@ type JoinRequest struct {
 
 func (x *JoinRequest) Reset() {
 	*x = JoinRequest{}
-	mi := &file_proto_chat_proto_msgTypes[0]
+	mi := &file_proto_chat_proto_msgTypes[2]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -42,7 +130,7 @@ func (x *JoinRequest) String() string {
 func (*JoinRequest) ProtoMessage() {}
 
 func (x *JoinRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_chat_proto_msgTypes[0]
+	mi := &file_proto_chat_proto_msgTypes[2]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -55,7 +143,7 @@ func (x *JoinRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use JoinRequest.ProtoReflect.Descriptor instead.
 func (*JoinRequest) Descriptor() ([]byte, []int) {
-	return file_proto_chat_proto_rawDescGZIP(), []int{0}
+	return file_proto_chat_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *JoinRequest) GetUserId() string {
@@ -77,7 +165,7 @@ type Message struct {
 
 func (x *Message) Reset() {
 	*x = Message{}
-	mi := &file_proto_chat_proto_msgTypes[1]
+	mi := &file_proto_chat_proto_msgTypes[3]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -89,7 +177,7 @@ func (x *Message) String() string {
 func (*Message) ProtoMessage() {}
 
 func (x *Message) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_chat_proto_msgTypes[1]
+	mi := &file_proto_chat_proto_msgTypes[3]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -102,7 +190,7 @@ func (x *Message) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Message.ProtoReflect.Descriptor instead.
 func (*Message) Descriptor() ([]byte, []int) {
-	return file_proto_chat_proto_rawDescGZIP(), []int{1}
+	return file_proto_chat_proto_rawDescGZIP(), []int{3}
 }
 
 func (x *Message) GetUserId() string {
@@ -143,7 +231,7 @@ type MessageAck struct {
 
 func (x *MessageAck) Reset() {
 	*x = MessageAck{}
-	mi := &file_proto_chat_proto_msgTypes[2]
+	mi := &file_proto_chat_proto_msgTypes[4]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -155,7 +243,7 @@ func (x *MessageAck) String() string {
 func (*MessageAck) ProtoMessage() {}
 
 func (x *MessageAck) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_chat_proto_msgTypes[2]
+	mi := &file_proto_chat_proto_msgTypes[4]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -168,7 +256,7 @@ func (x *MessageAck) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MessageAck.ProtoReflect.Descriptor instead.
 func (*MessageAck) Descriptor() ([]byte, []int) {
-	return file_proto_chat_proto_rawDescGZIP(), []int{2}
+	return file_proto_chat_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *MessageAck) GetSuccess() bool {
@@ -189,7 +277,11 @@ var File_proto_chat_proto protoreflect.FileDescriptor
 
 const file_proto_chat_proto_rawDesc = "" +
 	"\n" +
-	"\x10proto/chat.proto\x12\x04chat\"&\n" +
+	"\x10proto/chat.proto\x12\x04chat\"\r\n" +
+	"\vPingRequest\"D\n" +
+	"\fPingResponse\x12\x16\n" +
+	"\x06status\x18\x01 \x01(\tR\x06status\x12\x1c\n" +
+	"\ttimestamp\x18\x02 \x01(\x03R\ttimestamp\"&\n" +
 	"\vJoinRequest\x12\x17\n" +
 	"\auser_id\x18\x01 \x01(\tR\x06userId\"l\n" +
 	"\aMessage\x12\x17\n" +
@@ -201,10 +293,11 @@ const file_proto_chat_proto_rawDesc = "" +
 	"MessageAck\x12\x18\n" +
 	"\asuccess\x18\x01 \x01(\bR\asuccess\x12\x1d\n" +
 	"\n" +
-	"message_id\x18\x02 \x01(\tR\tmessageId2m\n" +
+	"message_id\x18\x02 \x01(\tR\tmessageId2\x9c\x01\n" +
 	"\vChatService\x12.\n" +
 	"\bJoinChat\x12\x11.chat.JoinRequest\x1a\r.chat.Message0\x01\x12.\n" +
-	"\vSendMessage\x12\r.chat.Message\x1a\x10.chat.MessageAckB\rZ\v./chat;chatb\x06proto3"
+	"\vSendMessage\x12\r.chat.Message\x1a\x10.chat.MessageAck\x12-\n" +
+	"\x04Ping\x12\x11.chat.PingRequest\x1a\x12.chat.PingResponseB\rZ\v./chat;chatb\x06proto3"
 
 var (
 	file_proto_chat_proto_rawDescOnce sync.Once
@@ -218,19 +311,23 @@ func file_proto_chat_proto_rawDescGZIP() []byte {
 	return file_proto_chat_proto_rawDescData
 }
 
-var file_proto_chat_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_proto_chat_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
 var file_proto_chat_proto_goTypes = []any{
-	(*JoinRequest)(nil), // 0: chat.JoinRequest
-	(*Message)(nil),     // 1: chat.Message
-	(*MessageAck)(nil),  // 2: chat.MessageAck
+	(*PingRequest)(nil),  // 0: chat.PingRequest
+	(*PingResponse)(nil), // 1: chat.PingResponse
+	(*JoinRequest)(nil),  // 2: chat.JoinRequest
+	(*Message)(nil),      // 3: chat.Message
+	(*MessageAck)(nil),   // 4: chat.MessageAck
 }
 var file_proto_chat_proto_depIdxs = []int32{
-	0, // 0: chat.ChatService.JoinChat:input_type -> chat.JoinRequest
-	1, // 1: chat.ChatService.SendMessage:input_type -> chat.Message
-	1, // 2: chat.ChatService.JoinChat:output_type -> chat.Message
-	2, // 3: chat.ChatService.SendMessage:output_type -> chat.MessageAck
-	2, // [2:4] is the sub-list for method output_type
-	0, // [0:2] is the sub-list for method input_type
+	2, // 0: chat.ChatService.JoinChat:input_type -> chat.JoinRequest
+	3, // 1: chat.ChatService.SendMessage:input_type -> chat.Message
+	0, // 2: chat.ChatService.Ping:input_type -> chat.PingRequest
+	3, // 3: chat.ChatService.JoinChat:output_type -> chat.Message
+	4, // 4: chat.ChatService.SendMessage:output_type -> chat.MessageAck
+	1, // 5: chat.ChatService.Ping:output_type -> chat.PingResponse
+	3, // [3:6] is the sub-list for method output_type
+	0, // [0:3] is the sub-list for method input_type
 	0, // [0:0] is the sub-list for extension type_name
 	0, // [0:0] is the sub-list for extension extendee
 	0, // [0:0] is the sub-list for field type_name
@@ -247,7 +344,7 @@ func file_proto_chat_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_chat_proto_rawDesc), len(file_proto_chat_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   3,
+			NumMessages:   5,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

client/chat/chat_grpc.pb.go

@@ -21,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9
 const (
 	ChatService_JoinChat_FullMethodName    = "/chat.ChatService/JoinChat"
 	ChatService_SendMessage_FullMethodName = "/chat.ChatService/SendMessage"
+	ChatService_Ping_FullMethodName        = "/chat.ChatService/Ping"
 )
 
 // ChatServiceClient is the client API for ChatService service.
@@ -29,6 +30,7 @@ const (
 type ChatServiceClient interface {
 	JoinChat(ctx context.Context, in *JoinRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Message], error)
 	SendMessage(ctx context.Context, in *Message, opts ...grpc.CallOption) (*MessageAck, error)
+	Ping(ctx context.Context, in *PingRequest, opts ...grpc.CallOption) (*PingResponse, error)
 }
 
 type chatServiceClient struct {
@@ -68,12 +70,23 @@ func (c *chatServiceClient) SendMessage(ctx context.Context, in *Message, opts .
 	return out, nil
 }
 
+func (c *chatServiceClient) Ping(ctx context.Context, in *PingRequest, opts ...grpc.CallOption) (*PingResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(PingResponse)
+	err := c.cc.Invoke(ctx, ChatService_Ping_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // ChatServiceServer is the server API for ChatService service.
 // All implementations must embed UnimplementedChatServiceServer
 // for forward compatibility.
 type ChatServiceServer interface {
 	JoinChat(*JoinRequest, grpc.ServerStreamingServer[Message]) error
 	SendMessage(context.Context, *Message) (*MessageAck, error)
+	Ping(context.Context, *PingRequest) (*PingResponse, error)
 	mustEmbedUnimplementedChatServiceServer()
 }
 
@@ -90,6 +103,9 @@ func (UnimplementedChatServiceServer) JoinChat(*JoinRequest, grpc.ServerStreamin
 func (UnimplementedChatServiceServer) SendMessage(context.Context, *Message) (*MessageAck, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SendMessage not implemented")
 }
+func (UnimplementedChatServiceServer) Ping(context.Context, *PingRequest) (*PingResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
+}
 func (UnimplementedChatServiceServer) mustEmbedUnimplementedChatServiceServer() {}
 func (UnimplementedChatServiceServer) testEmbeddedByValue()                     {}
 
@@ -140,6 +156,24 @@ func _ChatService_SendMessage_Handler(srv interface{}, ctx context.Context, dec
 	return interceptor(ctx, in, info, handler)
 }
 
+func _ChatService_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PingRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ChatServiceServer).Ping(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: ChatService_Ping_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ChatServiceServer).Ping(ctx, req.(*PingRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // ChatService_ServiceDesc is the grpc.ServiceDesc for ChatService service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -151,6 +185,10 @@ var ChatService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "SendMessage",
 			Handler:    _ChatService_SendMessage_Handler,
 		},
+		{
+			MethodName: "Ping",
+			Handler:    _ChatService_Ping_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{

client/config.json

@@ -1,4 +1,4 @@
 {
   "serviceAddress": "localhost:50051",
-  "informationDelay": 5
+  "informationDelay": 20
 }

client/log.go

@@ -37,7 +37,7 @@ func initLog(saveDay int) {
 	go logfile()
 	task := cron.New()
 	task.Start()
-	task.AddFunc("0 0 0/10 * * ?", func() {
+	task.AddFunc("0 0 0/24 * * ?", func() {
 		go logfile()
 		time.Sleep(50 * time.Second)
 		if saveDay > 0 {

client/proto/chat.proto

@@ -6,8 +6,15 @@ option go_package = "./chat;chat";
 
 service ChatService {
   rpc JoinChat(JoinRequest) returns (stream Message);
-  rpc SendMessage(Message) returns (MessageAck);}
+  rpc SendMessage(Message) returns (MessageAck);
+  rpc Ping(PingRequest) returns (PingResponse);
 
+}
+message PingRequest {}
+message PingResponse {
+  string status = 1;
+  int64 timestamp = 2;
+}
 message JoinRequest {
   string user_id = 1;
 }

client/service/chatClient.go

@@ -0,0 +1,433 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"google.golang.org/grpc/codes"
+	"log"
+	"math"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	. "client/chat"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	initialReconnectInterval = 1 * time.Second
+	keepaliveTime            = 60 * time.Second
+	keepaliveTimeout         = 20 * time.Second
+	maxRetryCount            = 60
+	connectionTimeout        = 3 * time.Second
+	maxReconnectInterval     = 60 * time.Second
+	healthCheckInterval      = 30 * time.Second
+)
+
+var client = &ChatClient{}
+
+type ChatClient struct {
+	conn              *grpc.ClientConn
+	client            ChatServiceClient
+	ctx               context.Context
+	cancel            context.CancelFunc
+	userID            string
+	mu                sync.RWMutex
+	retryCount        int
+	isConnected       bool
+	wg                sync.WaitGroup
+	reconnecting      bool
+	serviceAddress    string
+	stream            ChatService_JoinChatClient
+	streamMutex       sync.Mutex
+	healthCheckTicker *time.Ticker
+	lastPingTime      time.Time
+}
+
+func NewChatClient(userID, address string) *ChatClient {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &ChatClient{
+		userID:         userID,
+		ctx:            ctx,
+		cancel:         cancel,
+		serviceAddress: address,
+	}
+}
+
+// 连接服务器
+func (c *ChatClient) connect() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.isConnected && c.conn.GetState() == connectivity.Ready {
+		return nil
+	}
+
+	log.Println("[连接] 尝试连接服务器...", c.serviceAddress)
+	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
+	defer cancel()
+
+	conn, err := grpc.DialContext(ctx, c.serviceAddress,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock(),
+		grpc.WithDefaultCallOptions(
+			grpc.MaxCallRecvMsgSize(20*1024*1024),
+			grpc.MaxCallSendMsgSize(20*1024*1024),
+		),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			Time:                keepaliveTime,
+			Timeout:             keepaliveTimeout,
+			PermitWithoutStream: true,
+		}),
+	)
+	if err != nil {
+		return fmt.Errorf("连接失败: %v", err)
+	}
+
+	// 检查连接状态
+	state := conn.GetState()
+	if state != connectivity.Ready {
+		if conn.WaitForStateChange(ctx, connectivity.Connecting) {
+			state := conn.GetState()
+			if state != connectivity.Ready {
+				_ = conn.Close()
+				return fmt.Errorf("连接未就绪，状态: %v", state)
+			}
+		}
+	}
+	client := NewChatServiceClient(conn)
+	_, err = client.JoinChat(context.Background(), &JoinRequest{UserId: c.userID})
+	if err != nil {
+		_ = conn.Close()
+		return fmt.Errorf("连接测试失败: %v", err)
+	}
+
+	// 关闭旧连接
+	if c.conn != nil {
+		_ = c.conn.Close()
+	}
+
+	c.conn = conn
+	c.client = client
+	c.isConnected = true
+	c.retryCount = 0
+	c.lastPingTime = time.Now()
+	// 启动健康检查
+	c.startHealthCheck()
+	log.Printf("[连接][用户:%s] 服务器连接成功", c.userID)
+	return nil
+}
+
+// 断开连接
+func (c *ChatClient) disconnect() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if !c.isConnected {
+		return
+	}
+
+	// 停止健康检查
+	if c.healthCheckTicker != nil {
+		c.healthCheckTicker.Stop()
+		c.healthCheckTicker = nil
+	}
+
+	c.closeStream()
+	if c.conn != nil {
+		if err := c.conn.Close(); err != nil {
+			log.Printf("[连接][用户:%s] 关闭连接出错: %v", c.userID, err)
+		}
+		c.conn = nil
+	}
+	c.isConnected = false
+	log.Printf("[连接][用户:%s] 已断开连接", c.userID)
+}
+
+// 关闭流
+func (c *ChatClient) closeStream() {
+	c.streamMutex.Lock()
+	defer c.streamMutex.Unlock()
+	if c.stream != nil {
+		if err := c.stream.CloseSend(); err != nil {
+			log.Printf("[流] 关闭流错误: %v", err)
+		}
+		c.stream = nil
+	}
+}
+
+// 重连逻辑
+func (c *ChatClient) reconnect() {
+	c.mu.Lock()
+	if c.reconnecting {
+		c.mu.Unlock()
+		return
+	}
+	c.reconnecting = true
+	currentRetry := c.retryCount
+	c.mu.Unlock()
+
+	defer func() {
+		c.mu.Lock()
+		c.reconnecting = false
+		c.mu.Unlock()
+	}()
+
+	c.wg.Add(1)
+	defer c.wg.Done()
+
+	log.Printf("[重连] 开始重连流程，当前重试计数: %d", currentRetry)
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			log.Println("[重连] 上下文取消，停止重连")
+			return
+		default:
+			c.mu.Lock()
+			if c.retryCount >= maxRetryCount {
+				log.Printf("[重连] 达到最大重试次数(%d)，停止重连", maxRetryCount)
+				c.mu.Unlock()
+				return
+			}
+			c.mu.Unlock()
+
+			log.Printf("[重连] 尝试第%d次连接...", currentRetry+1)
+			if err := c.connect(); err != nil {
+				log.Printf("[重连][用户:%s] 连接失败: %v", c.userID, err)
+
+				c.mu.Lock()
+				c.retryCount++
+				currentRetry = c.retryCount
+				c.mu.Unlock()
+
+				// 指数退避算法
+				backoff := initialReconnectInterval
+				if currentRetry > 0 {
+					backoff = time.Duration(math.Min(
+						float64(initialReconnectInterval)*math.Pow(1.5, float64(currentRetry)),
+						float64(maxReconnectInterval),
+					))
+				}
+
+				select {
+				case <-time.After(backoff):
+					continue
+				case <-c.ctx.Done():
+					log.Printf("[重连][用户:%s] 等待期间上下文取消", c.userID)
+					return
+				}
+			} else {
+				log.Printf("[重连][用户:%s] 连接成功！", c.userID)
+				go c.establishStream()
+				return
+			}
+		}
+	}
+}
+
+// 建立流
+func (c *ChatClient) establishStream() {
+	c.wg.Add(1)
+	defer c.wg.Done()
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		default:
+			c.mu.RLock()
+			connected := c.isConnected
+			client := c.client
+			c.mu.RUnlock()
+
+			if !connected {
+				time.Sleep(1 * time.Second)
+				continue
+			}
+
+			c.streamMutex.Lock()
+			stream, err := client.JoinChat(c.ctx, &JoinRequest{UserId: c.userID})
+			if err != nil {
+				c.streamMutex.Unlock()
+				log.Printf("[流] 建立流失败: %v", err)
+				c.disconnect()
+				go c.reconnect()
+				return
+			}
+			c.stream = stream
+			c.streamMutex.Unlock()
+
+			if err := c.receiveMessages(stream); err != nil {
+				log.Printf("[流] 接收消息错误: %v", err)
+				c.disconnect()
+				go c.reconnect()
+				return
+			}
+		}
+	}
+}
+
+// 接收消息
+func (c *ChatClient) receiveMessages(stream ChatService_JoinChatClient) error {
+	for {
+		msg, err := stream.Recv()
+		if err != nil {
+			if status.Code(err) == codes.Canceled {
+				return nil // 正常关闭
+			}
+			return fmt.Errorf("接收消息错误: %v", err)
+		}
+
+		log.Printf("[接收] 收到消息: %+v", msg)
+		if msg.UserId == "系统" {
+			switch msg.Action {
+			case "sendTalk":
+				go SendTalk(msg.Text)
+			case "getContacts":
+				go GetContacts()
+			case "reject":
+				go Reject(msg.Text)
+			default:
+				log.Printf("[系统通知]: %s", msg.Text)
+			}
+		}
+	}
+}
+
+// 发送消息
+func (c *ChatClient) SendMessage(text, action string) error {
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if !c.isReady() {
+		return fmt.Errorf("未连接服务器")
+	}
+
+	msg := &Message{
+		UserId: c.userID,
+		Text:   text,
+		Action: action,
+	}
+	log.Printf("[发送] 发送消息: %+v", msg)
+
+	ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
+	defer cancel()
+
+	_, err := c.client.SendMessage(ctx, msg)
+	if err != nil {
+		log.Printf("[发送] 发送失败: %v", err)
+		c.disconnect()
+		go c.reconnect()
+		return err
+	}
+	return nil
+}
+
+// 启动客户端
+func ConnectGRPC(userId, address string) {
+	log.Println("[主程序] 启动GRPC连接")
+	client = NewChatClient(userId, address)
+	defer client.Shutdown()
+
+	if err := client.connect(); err != nil {
+		log.Printf("[主程序] 初始连接失败: %v", err)
+		go client.reconnect()
+	} else {
+		go client.establishStream()
+	}
+	// 保持主线程运行
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+	<-quit
+}
+func (c *ChatClient) isReady() bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.isConnected && c.conn != nil && c.conn.GetState() == connectivity.Ready
+}
+
+// startHealthCheck 启动健康检查
+func (c *ChatClient) startHealthCheck() {
+	if c.healthCheckTicker != nil {
+		c.healthCheckTicker.Stop()
+	}
+
+	c.healthCheckTicker = time.NewTicker(healthCheckInterval)
+	c.wg.Add(1)
+
+	go func() {
+		defer c.wg.Done()
+		defer c.healthCheckTicker.Stop()
+		for {
+			select {
+			case <-c.healthCheckTicker.C:
+				if !c.isReady() {
+					log.Printf("[健康检查][用户:%s] 连接不可用，触发重连", c.userID)
+					go c.reconnect()
+					continue
+				}
+
+				// 执行Ping检查
+				ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
+				_, err := c.client.Ping(ctx, &PingRequest{})
+				cancel()
+
+				if err != nil {
+					log.Printf("[健康检查][用户:%s] Ping失败: %v", c.userID, err)
+					c.disconnect()
+					go c.reconnect()
+				} else {
+					c.mu.Lock()
+					c.lastPingTime = time.Now()
+					c.mu.Unlock()
+				}
+
+			case <-c.ctx.Done():
+				log.Printf("[健康检查][用户:%s] 停止健康检查", c.userID)
+				return
+			}
+		}
+	}()
+}
+
+// 修改健康检查实现
+func (c *ChatClient) checkHealth() error {
+	ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
+	defer cancel()
+
+	// 优先使用JoinChat作为健康检查
+	_, err := c.client.JoinChat(ctx, &JoinRequest{UserId: c.userID})
+	if err != nil {
+		st, ok := status.FromError(err)
+		if ok && st.Code() == codes.Unimplemented {
+			// 如果JoinChat未实现，尝试其他方法
+			return c.fallbackHealthCheck()
+		}
+		return err
+	}
+	return nil
+}
+func (c *ChatClient) fallbackHealthCheck() error {
+	// 实现其他健康检查方式
+	return nil
+}
+func (c *ChatClient) Shutdown() {
+	// 1. 取消上下文
+	c.cancel()
+
+	// 2. 断开连接
+	c.disconnect()
+
+	// 3. 等待所有goroutine结束
+	c.wg.Wait()
+}
+func (c *ChatClient) log() *log.Logger {
+	return log.New(os.Stdout, fmt.Sprintf("[用户:%s] ", c.userID), log.LstdFlags)
+}

client/service/commandHandle.go

@@ -1,286 +0,0 @@
-package service
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"sync"
-	"time"
-
-	. "client/chat"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/keepalive"
-)
-
-const (
-	reconnectInterval = 60 * time.Second
-	timeInt64         = 60 * time.Second
-	TimeoutInt64      = 20 * time.Second
-
-	maxRetryCount = 60
-)
-
-type Config struct {
-	ServerAddress     string
-	ReconnectInterval time.Duration
-	MaxRetryCount     int
-}
-
-var client = &ChatClient{}
-
-type ChatClient struct {
-	conn           *grpc.ClientConn
-	client         ChatServiceClient
-	ctx            context.Context
-	cancel         context.CancelFunc
-	userID         string
-	mu             sync.Mutex
-	retryCount     int
-	isConnected    bool
-	wg             sync.WaitGroup
-	reconnecting   bool
-	ServiceAddress string
-}
-
-func NewChatClient(userID, address string) *ChatClient {
-	ctx, cancel := context.WithCancel(context.Background())
-	return &ChatClient{
-		userID:         userID,
-		ctx:            ctx,
-		cancel:         cancel,
-		ServiceAddress: address,
-	}
-}
-
-// grpc连接
-func (c *ChatClient) connect() error {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if c.isConnected {
-		return nil
-	}
-
-	log.Println("[连接] 尝试连接服务器...")
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	defer cancel()
-
-	conn, err := grpc.DialContext(ctx, "localhost:50051",
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithBlock(),
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:                timeInt64,
-			Timeout:             TimeoutInt64,
-			PermitWithoutStream: true,
-		}))
-	if err != nil {
-		return fmt.Errorf("连接失败: %v", err)
-	}
-
-	// 测试连接
-	testClient := NewChatServiceClient(conn)
-	_, err = testClient.JoinChat(context.Background(), &JoinRequest{UserId: c.userID})
-	if err != nil {
-		_ = conn.Close()
-		return fmt.Errorf("连接测试失败: %v", err)
-	}
-
-	c.conn = conn
-	c.client = testClient
-	c.isConnected = true
-	c.retryCount = 0
-	log.Println("[连接] 服务器连接成功")
-	return nil
-}
-
-func (c *ChatClient) disconnect() {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if !c.isConnected {
-		return
-	}
-
-	if c.conn != nil {
-		if err := c.conn.Close(); err != nil {
-			log.Println("[连接] 关闭连接出错: %v", err)
-		}
-		c.conn = nil
-	}
-	c.isConnected = false
-	log.Println("[连接] 已断开连接")
-}
-
-func (c *ChatClient) reconnect() {
-	// 初始加锁检查重连状态
-	c.mu.Lock()
-	if c.reconnecting {
-		c.mu.Unlock()
-		return
-	}
-	c.reconnecting = true
-	currentRetry := c.retryCount
-	c.mu.Unlock()
-
-	// 确保最终重置重连状态
-	defer func() {
-		c.mu.Lock()
-		c.reconnecting = false
-		c.mu.Unlock()
-	}()
-
-	// 等待组管理
-	c.wg.Add(1)
-	defer c.wg.Done()
-
-	log.Println("[重连] 开始重连流程，当前重试计数: %d", currentRetry)
-
-	for {
-		select {
-		case <-c.ctx.Done():
-			log.Println("[重连] 上下文取消，停止重连")
-			return
-		default:
-			// 检查最大重试次数
-			c.mu.Lock()
-			if c.retryCount >= maxRetryCount {
-				log.Println("[重连] 达到最大重试次数(%d)，停止重连", maxRetryCount)
-				c.mu.Unlock()
-				return
-			}
-			c.mu.Unlock()
-			// 尝试连接
-			log.Println("[重连] 尝试第%d次连接...", currentRetry+1)
-			if err := c.connect(); err != nil {
-				log.Println("[重连] 连接失败: %v", err)
-
-				// 更新重试计数
-				c.mu.Lock()
-				c.retryCount++
-				currentRetry = c.retryCount
-				c.mu.Unlock()
-				log.Println("[重连] 等待%d秒后重试...", int(reconnectInterval.Seconds()))
-				select {
-				case <-time.After(reconnectInterval):
-					continue
-				case <-c.ctx.Done():
-					log.Println("[重连] 等待期间上下文取消")
-					return
-				}
-			} else {
-				log.Println("[重连] 连接成功！")
-				go c.receiveMessages()
-				return
-			}
-		}
-	}
-}
-
-// 接收信息处理
-// 修改后的 receiveMessages 方法
-func (c *ChatClient) receiveMessages() {
-	c.wg.Add(1)
-	defer c.wg.Done()
-
-	for {
-		select {
-		case <-c.ctx.Done():
-			return
-		default:
-			c.mu.Lock()
-			connected := c.isConnected
-			c.mu.Unlock()
-
-			if !connected {
-				time.Sleep(1 * time.Second)
-				continue
-			}
-
-			// 加锁保护整个连接过程
-
-			c.mu.Lock()
-			stream, err := c.client.JoinChat(c.ctx, &JoinRequest{UserId: c.userID})
-			c.mu.Unlock()
-
-			if err != nil {
-				log.Println(fmt.Sprintf("[接收] 加入聊天室失败: %v", err))
-				c.disconnect()
-				go c.reconnect()
-				return
-			}
-
-			// 消息接收循环
-			for {
-				msg, err := stream.Recv()
-				if err != nil {
-					log.Println(fmt.Sprintf("[接收] 接收消息错误: %v\n", err))
-					c.disconnect()
-					go c.reconnect()
-					return
-				}
-				// ... 处理消息逻辑 ...
-				log.Println("[接收] 收到消息: ", msg)
-				if msg.UserId == "系统" {
-					switch msg.Action {
-					case "sendTalk":
-						go SendTalk(msg.Text)
-					case "getContacts":
-						go GetContacts()
-					case "reject":
-						go Reject(msg.Text)
-					default:
-						//后期删除掉
-						if msg.Text == "欢迎加入聊天室！" {
-							log.Println("发送固定信息", client.SendMessage("", "sendTalk"))
-						}
-						log.Println(fmt.Sprintf("[系统通知]: %s ", msg.Text))
-					}
-				}
-			}
-		}
-	}
-}
-
-// SendMessage 发送单条消息
-func (c *ChatClient) SendMessage(text string, action string) error {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	if !c.isConnected {
-		return fmt.Errorf("未连接服务器")
-	}
-	msg := &Message{
-		UserId: c.userID,
-		Text:   text,
-		Action: action,
-	}
-	log.Println(fmt.Sprintf("[发送] 发送消息 : %+v\n", msg))
-	_, err := c.client.SendMessage(c.ctx, msg)
-	if err != nil {
-		log.Println(fmt.Sprintf("[发送] 发送失败: %v", err))
-		c.disconnect()
-		go c.reconnect()
-		return err
-	}
-	return nil
-}
-
-func ConnectGRPC(userId string, address string) {
-	log.Println("[主程序] 启动GRPC连接")
-	client = NewChatClient(userId, address)
-	defer func() {
-		client.cancel()
-		client.disconnect()
-		client.wg.Wait()
-		log.Println("[主程序] 客户端安全退出")
-	}()
-
-	// 初始连接
-	if err := client.connect(); err != nil {
-		log.Println(fmt.Sprintf("[主程序] 初始连接失败: %v", err))
-		go client.reconnect()
-	} else {
-		go client.receiveMessages()
-	}
-	select {}
-}

client/service/wx.go

@@ -75,7 +75,7 @@ func OnMsg(ctx context.Context) {
 // msg: 包含微信消息所有信息的结构体
 func processMsg(msg *wcf.WxMsg) {
 	// 结构化日志输出消息详情
-	fmt.Printf(`
+	log.Println(fmt.Sprintf(`
 		[消息详情]
 		是否来自自己: %v
 		是否是群消息: %v
@@ -90,7 +90,7 @@ func processMsg(msg *wcf.WxMsg) {
 		msg.Type, msg.Ts,
 		msg.Id, msg.Roomid, msg.Sender,
 		msg.Content,
-	)
+	))
 	if !msg.IsGroup && msg.Type == 1 {
 		returnData := map[string]interface{}{
 			"roomid":  msg.Roomid,
@@ -110,7 +110,6 @@ func WxHandle() {
 	// 注册Ctrl+C信号处理函数
 	signalChan := make(chan os.Signal, 1)
 	signal.Notify(signalChan, os.Interrupt)
-	// 开启微信消息接收功能
 	go func() {
 		<-signalChan
 		cancleFn()
@@ -150,6 +149,7 @@ func GetContacts() {
 			"remark": c.Remark, // 备注名
 			"phone":  phone,
 		})
+
 	}
 	// 将联系人列表发送给客户端
 	if err := client.SendMessage(gconv.String(returnData), "getContacts"); err != nil {
@@ -339,6 +339,7 @@ func doSendTalk(dataStr string) error {
 			}
 		}
 	}
+	time.Sleep(3 * time.Second)
 	if replyLanguage != "" {
 		ok1, err := sendText(replyLanguage, wxId)
 		if !ok1 {
@@ -346,7 +347,7 @@ func doSendTalk(dataStr string) error {
 			log.Println(err)
 		}
 	}
-	//
+
 	// 发送回执
 	returnData := map[string]interface{}{
 		"taskId":       taskId,
@@ -359,6 +360,7 @@ func doSendTalk(dataStr string) error {
 
 // 拒绝接受回执
 func Reject(text string) {
+	time.Sleep(10 * time.Second)
 	// 1. 解析输入数据
 	dataMap := gconv.Map(text)
 	if dataMap == nil {

rpc/chat/chat.pb.go

@@ -21,6 +21,86 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+type PingRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *PingRequest) Reset() {
+	*x = PingRequest{}
+	mi := &file_proto_chat_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *PingRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PingRequest) ProtoMessage() {}
+
+func (x *PingRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_chat_proto_msgTypes[0]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PingRequest.ProtoReflect.Descriptor instead.
+func (*PingRequest) Descriptor() ([]byte, []int) {
+	return file_proto_chat_proto_rawDescGZIP(), []int{0}
+}
+
+type PingResponse struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Status        string                 `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"` // 可以返回服务器状态
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *PingResponse) Reset() {
+	*x = PingResponse{}
+	mi := &file_proto_chat_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *PingResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PingResponse) ProtoMessage() {}
+
+func (x *PingResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_chat_proto_msgTypes[1]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PingResponse.ProtoReflect.Descriptor instead.
+func (*PingResponse) Descriptor() ([]byte, []int) {
+	return file_proto_chat_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *PingResponse) GetStatus() string {
+	if x != nil {
+		return x.Status
+	}
+	return ""
+}
+
 type JoinRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	UserId        string                 `protobuf:"bytes,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
@@ -30,7 +110,7 @@ type JoinRequest struct {
 
 func (x *JoinRequest) Reset() {
 	*x = JoinRequest{}
-	mi := &file_proto_chat_proto_msgTypes[0]
+	mi := &file_proto_chat_proto_msgTypes[2]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -42,7 +122,7 @@ func (x *JoinRequest) String() string {
 func (*JoinRequest) ProtoMessage() {}
 
 func (x *JoinRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_chat_proto_msgTypes[0]
+	mi := &file_proto_chat_proto_msgTypes[2]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -55,7 +135,7 @@ func (x *JoinRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use JoinRequest.ProtoReflect.Descriptor instead.
 func (*JoinRequest) Descriptor() ([]byte, []int) {
-	return file_proto_chat_proto_rawDescGZIP(), []int{0}
+	return file_proto_chat_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *JoinRequest) GetUserId() string {
@@ -77,7 +157,7 @@ type Message struct {
 
 func (x *Message) Reset() {
 	*x = Message{}
-	mi := &file_proto_chat_proto_msgTypes[1]
+	mi := &file_proto_chat_proto_msgTypes[3]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -89,7 +169,7 @@ func (x *Message) String() string {
 func (*Message) ProtoMessage() {}
 
 func (x *Message) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_chat_proto_msgTypes[1]
+	mi := &file_proto_chat_proto_msgTypes[3]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -102,7 +182,7 @@ func (x *Message) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Message.ProtoReflect.Descriptor instead.
 func (*Message) Descriptor() ([]byte, []int) {
-	return file_proto_chat_proto_rawDescGZIP(), []int{1}
+	return file_proto_chat_proto_rawDescGZIP(), []int{3}
 }
 
 func (x *Message) GetUserId() string {
@@ -143,7 +223,7 @@ type MessageAck struct {
 
 func (x *MessageAck) Reset() {
 	*x = MessageAck{}
-	mi := &file_proto_chat_proto_msgTypes[2]
+	mi := &file_proto_chat_proto_msgTypes[4]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -155,7 +235,7 @@ func (x *MessageAck) String() string {
 func (*MessageAck) ProtoMessage() {}
 
 func (x *MessageAck) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_chat_proto_msgTypes[2]
+	mi := &file_proto_chat_proto_msgTypes[4]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -168,7 +248,7 @@ func (x *MessageAck) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MessageAck.ProtoReflect.Descriptor instead.
 func (*MessageAck) Descriptor() ([]byte, []int) {
-	return file_proto_chat_proto_rawDescGZIP(), []int{2}
+	return file_proto_chat_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *MessageAck) GetSuccess() bool {
@@ -189,7 +269,10 @@ var File_proto_chat_proto protoreflect.FileDescriptor
 
 const file_proto_chat_proto_rawDesc = "" +
 	"\n" +
-	"\x10proto/chat.proto\x12\x04chat\"&\n" +
+	"\x10proto/chat.proto\x12\x04chat\"\r\n" +
+	"\vPingRequest\"&\n" +
+	"\fPingResponse\x12\x16\n" +
+	"\x06status\x18\x01 \x01(\tR\x06status\"&\n" +
 	"\vJoinRequest\x12\x17\n" +
 	"\auser_id\x18\x01 \x01(\tR\x06userId\"l\n" +
 	"\aMessage\x12\x17\n" +
@@ -201,10 +284,11 @@ const file_proto_chat_proto_rawDesc = "" +
 	"MessageAck\x12\x18\n" +
 	"\asuccess\x18\x01 \x01(\bR\asuccess\x12\x1d\n" +
 	"\n" +
-	"message_id\x18\x02 \x01(\tR\tmessageId2m\n" +
+	"message_id\x18\x02 \x01(\tR\tmessageId2\x9c\x01\n" +
 	"\vChatService\x12.\n" +
 	"\bJoinChat\x12\x11.chat.JoinRequest\x1a\r.chat.Message0\x01\x12.\n" +
-	"\vSendMessage\x12\r.chat.Message\x1a\x10.chat.MessageAckB\rZ\v./chat;chatb\x06proto3"
+	"\vSendMessage\x12\r.chat.Message\x1a\x10.chat.MessageAck\x12-\n" +
+	"\x04Ping\x12\x11.chat.PingRequest\x1a\x12.chat.PingResponseB\rZ\v./chat;chatb\x06proto3"
 
 var (
 	file_proto_chat_proto_rawDescOnce sync.Once
@@ -218,19 +302,23 @@ func file_proto_chat_proto_rawDescGZIP() []byte {
 	return file_proto_chat_proto_rawDescData
 }
 
-var file_proto_chat_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_proto_chat_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
 var file_proto_chat_proto_goTypes = []any{
-	(*JoinRequest)(nil), // 0: chat.JoinRequest
-	(*Message)(nil),     // 1: chat.Message
-	(*MessageAck)(nil),  // 2: chat.MessageAck
+	(*PingRequest)(nil),  // 0: chat.PingRequest
+	(*PingResponse)(nil), // 1: chat.PingResponse
+	(*JoinRequest)(nil),  // 2: chat.JoinRequest
+	(*Message)(nil),      // 3: chat.Message
+	(*MessageAck)(nil),   // 4: chat.MessageAck
 }
 var file_proto_chat_proto_depIdxs = []int32{
-	0, // 0: chat.ChatService.JoinChat:input_type -> chat.JoinRequest
-	1, // 1: chat.ChatService.SendMessage:input_type -> chat.Message
-	1, // 2: chat.ChatService.JoinChat:output_type -> chat.Message
-	2, // 3: chat.ChatService.SendMessage:output_type -> chat.MessageAck
-	2, // [2:4] is the sub-list for method output_type
-	0, // [0:2] is the sub-list for method input_type
+	2, // 0: chat.ChatService.JoinChat:input_type -> chat.JoinRequest
+	3, // 1: chat.ChatService.SendMessage:input_type -> chat.Message
+	0, // 2: chat.ChatService.Ping:input_type -> chat.PingRequest
+	3, // 3: chat.ChatService.JoinChat:output_type -> chat.Message
+	4, // 4: chat.ChatService.SendMessage:output_type -> chat.MessageAck
+	1, // 5: chat.ChatService.Ping:output_type -> chat.PingResponse
+	3, // [3:6] is the sub-list for method output_type
+	0, // [0:3] is the sub-list for method input_type
 	0, // [0:0] is the sub-list for extension type_name
 	0, // [0:0] is the sub-list for extension extendee
 	0, // [0:0] is the sub-list for field type_name
@@ -247,7 +335,7 @@ func file_proto_chat_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_chat_proto_rawDesc), len(file_proto_chat_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   3,
+			NumMessages:   5,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

rpc/chat/chat_grpc.pb.go

@@ -21,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9
 const (
 	ChatService_JoinChat_FullMethodName    = "/chat.ChatService/JoinChat"
 	ChatService_SendMessage_FullMethodName = "/chat.ChatService/SendMessage"
+	ChatService_Ping_FullMethodName        = "/chat.ChatService/Ping"
 )
 
 // ChatServiceClient is the client API for ChatService service.
@@ -29,6 +30,7 @@ const (
 type ChatServiceClient interface {
 	JoinChat(ctx context.Context, in *JoinRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Message], error)
 	SendMessage(ctx context.Context, in *Message, opts ...grpc.CallOption) (*MessageAck, error)
+	Ping(ctx context.Context, in *PingRequest, opts ...grpc.CallOption) (*PingResponse, error)
 }
 
 type chatServiceClient struct {
@@ -68,12 +70,23 @@ func (c *chatServiceClient) SendMessage(ctx context.Context, in *Message, opts .
 	return out, nil
 }
 
+func (c *chatServiceClient) Ping(ctx context.Context, in *PingRequest, opts ...grpc.CallOption) (*PingResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(PingResponse)
+	err := c.cc.Invoke(ctx, ChatService_Ping_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // ChatServiceServer is the server API for ChatService service.
 // All implementations must embed UnimplementedChatServiceServer
 // for forward compatibility.
 type ChatServiceServer interface {
 	JoinChat(*JoinRequest, grpc.ServerStreamingServer[Message]) error
 	SendMessage(context.Context, *Message) (*MessageAck, error)
+	Ping(context.Context, *PingRequest) (*PingResponse, error)
 	mustEmbedUnimplementedChatServiceServer()
 }
 
@@ -90,6 +103,9 @@ func (UnimplementedChatServiceServer) JoinChat(*JoinRequest, grpc.ServerStreamin
 func (UnimplementedChatServiceServer) SendMessage(context.Context, *Message) (*MessageAck, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SendMessage not implemented")
 }
+func (UnimplementedChatServiceServer) Ping(context.Context, *PingRequest) (*PingResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
+}
 func (UnimplementedChatServiceServer) mustEmbedUnimplementedChatServiceServer() {}
 func (UnimplementedChatServiceServer) testEmbeddedByValue()                     {}
 
@@ -140,6 +156,24 @@ func _ChatService_SendMessage_Handler(srv interface{}, ctx context.Context, dec
 	return interceptor(ctx, in, info, handler)
 }
 
+func _ChatService_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PingRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ChatServiceServer).Ping(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: ChatService_Ping_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ChatServiceServer).Ping(ctx, req.(*PingRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // ChatService_ServiceDesc is the grpc.ServiceDesc for ChatService service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -151,6 +185,10 @@ var ChatService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "SendMessage",
 			Handler:    _ChatService_SendMessage_Handler,
 		},
+		{
+			MethodName: "Ping",
+			Handler:    _ChatService_Ping_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{

rpc/config/config.go

@@ -58,6 +58,7 @@ var (
 )
 
 func init() {
+
 	util.ReadConfig(&DbConf)
 	TiDbMysql = &mysql.Mysql{
 		Address:      DbConf.TidbMysql.Address,

rpc/log.go

@@ -37,7 +37,7 @@ func initLog(saveDay int) {
 	go logfile()
 	task := cron.New()
 	task.Start()
-	task.AddFunc("0 0 0/10 * * ?", func() {
+	task.AddFunc("0 0 0/24 * * ?", func() {
 		go logfile()
 		time.Sleep(50 * time.Second)
 		if saveDay > 0 {

rpc/main.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"app.yhyue.com/moapp/jybase/go-xweb/xweb"
 	"context"
 	"fmt"
 	"log"
@@ -11,87 +10,179 @@ import (
 	"os/signal"
 	"rpc/config"
 	"rpc/service"
+	"sync"
 	"syscall"
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/keepalive"
 	_ "rpc/a"
 	. "rpc/chat"
 	_ "rpc/filter"
 )
 
+type server struct {
+	httpServer   *http.Server
+	grpcServer   *grpc.Server
+	healthServer *health.Server
+	wg           sync.WaitGroup
+	startTime    time.Time
+}
+
 func main() {
-	// 检查配置
+
+	// 初始化配置检查
 	if config.DbConf == nil {
 		log.Fatal("配置未初始化")
 	}
 
-	// 打印启动信息
-	startTime := time.Now()
-	log.Printf("服务启动中... [HTTP:%s, gRPC:%s]",
-		config.DbConf.WebPort, config.DbConf.GrpcWebPort)
+	// 创建服务器实例
+	srv := &server{
+		startTime: time.Now(),
+	}
 
-	// 启动HTTP服务
-	mux1 := http.NewServeMux()
-	go func() {
-		log.Printf("HTTP服务监听 :%s", config.DbConf.WebPort)
-		xweb.RunBase(":"+config.DbConf.WebPort, mux1)
-	}()
+	// 设置上下文
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-	// 启动gRPC服务
-	lis, err := net.Listen("tcp", fmt.Sprintf(":%s", config.DbConf.GrpcWebPort))
-	if err != nil {
-		log.Fatalf("无法监听gRPC端口 %s: %v", config.DbConf.GrpcWebPort, err)
+	// 初始化服务
+	srv.initHTTPServer()
+	srv.initGRPCServer()
+
+	// 启动服务
+	srv.startServices(ctx)
+
+	// 设置定时任务
+	srv.setupTimedTasks(ctx)
+
+	// 等待终止信号
+	srv.waitForShutdown()
+
+	// 优雅关闭
+	srv.gracefulShutdown()
+}
+
+func (s *server) initHTTPServer() {
+	mux := http.NewServeMux()
+	// 可以在这里添加HTTP路由
+	// mux.HandleFunc("/health", healthHandler)
+
+	s.httpServer = &http.Server{
+		Addr:    ":" + config.DbConf.WebPort,
+		Handler: mux,
 	}
+}
 
-	server := grpc.NewServer(
+func (s *server) initGRPCServer() {
+	// 创建健康检查服务
+	s.healthServer = health.NewServer()
+
+	s.grpcServer = grpc.NewServer(
 		grpc.KeepaliveParams(keepalive.ServerParameters{
 			Time:    60 * time.Second,
 			Timeout: 120 * time.Second,
 		}),
 		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
-			MinTime:             30 * time.Second, // 调整为30秒
-			PermitWithoutStream: true,             // 允许无流连接
+			MinTime:             30 * time.Second,
+			PermitWithoutStream: true,
 		}),
+		grpc.MaxRecvMsgSize(20*1024*1024),
+		grpc.MaxSendMsgSize(20*1024*1024),
 	)
 
-	RegisterChatServiceServer(server, service.ChatSrv)
+	// 注册服务
+
+	RegisterChatServiceServer(s.grpcServer, service.Chatserver)
+	grpc_health_v1.RegisterHealthServer(s.grpcServer, s.healthServer)
+	s.healthServer.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING)
+}
 
-	// 启动定时任务
-	go service.ChatSrv.StartTimedMessages(2*time.Hour, "getContacts")
-	go service.ChatSrv.StartTimedMessages(10*time.Minute, "listenIn")
+func (s *server) startServices(ctx context.Context) {
+	// 启动HTTP服务
+	go func() {
+		log.Printf("HTTP服务监听 :%s", config.DbConf.WebPort)
+		if err := s.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			log.Fatalf("HTTP服务启动失败: %v", err)
+		}
+	}()
 
 	// 启动gRPC服务
+	lis, err := net.Listen("tcp", fmt.Sprintf(":%s", config.DbConf.GrpcWebPort))
+	if err != nil {
+		log.Fatalf("无法监听gRPC端口 %s: %v", config.DbConf.GrpcWebPort, err)
+	}
+
 	go func() {
 		log.Printf("gRPC服务已启动，监听端口 :%s [PID: %d]",
 			config.DbConf.GrpcWebPort, os.Getpid())
-		if err := server.Serve(lis); err != nil {
+		if err := s.grpcServer.Serve(lis); err != nil {
 			log.Fatalf("gRPC服务启动失败: %v", err)
 		}
 	}()
+}
+
+func (s *server) setupTimedTasks(ctx context.Context) {
+	tasks := []struct {
+		interval time.Duration
+		taskName string
+	}{
+		{1 * time.Minute, "sendTalk"},
+		{1 * time.Hour, "getContacts"},
+		{120 * time.Second, "heartbeat"},
+	}
 
-	// 优雅关闭处理
+	for _, task := range tasks {
+		task := task // 创建局部变量
+		s.wg.Add(1)
+		go func() {
+			defer s.wg.Done()
+			service.Chatserver.StartTimedMessages(ctx, task.interval, task.taskName)
+		}()
+	}
+}
+func (s *server) waitForShutdown() {
 	done := make(chan os.Signal, 1)
 	signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
 
 	<-done
-	log.Printf("\n接收到关闭信号 [运行时间: %s]", time.Since(startTime).Round(time.Second))
+	log.Printf("\n接收到关闭信号 [运行时间: %s]", time.Since(s.startTime).Round(time.Second))
+}
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
+func (s *server) gracefulShutdown() {
+	// 标记服务为不健康
+	s.healthServer.SetServingStatus("", grpc_health_v1.HealthCheckResponse_NOT_SERVING)
+
+	// 创建关闭上下文
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer shutdownCancel()
 
-	stopped := make(chan struct{})
+	// 关闭HTTP服务器
 	go func() {
-		server.GracefulStop()
-		close(stopped)
+		if err := s.httpServer.Shutdown(shutdownCtx); err != nil {
+			log.Printf("HTTP服务关闭错误: %v", err)
+		}
+	}()
+
+	// 关闭gRPC服务器
+	grpcStopped := make(chan struct{})
+	go func() {
+		s.grpcServer.GracefulStop()
+		close(grpcStopped)
 	}()
 
+	// 等待所有服务关闭或超时
 	select {
-	case <-ctx.Done():
-		log.Println("警告: 优雅关闭超时，强制终止")
-		server.Stop()
-	case <-stopped:
-		log.Println("服务已正常关闭")
+	case <-grpcStopped:
+		log.Println("gRPC服务已正常关闭")
+	case <-shutdownCtx.Done():
+		log.Println("警告: 优雅关闭超时，强制终止gRPC服务")
+		s.grpcServer.Stop()
 	}
+
+	// 等待所有定时任务结束
+	s.wg.Wait()
+
+	log.Println("服务已完全关闭")
 }

rpc/proto/chat.proto

@@ -6,7 +6,13 @@ option go_package = "./chat;chat";
 
 service ChatService {
   rpc JoinChat(JoinRequest) returns (stream Message); // 服务端流
-  rpc SendMessage(Message) returns (MessageAck);      // 一元调用
+  rpc SendMessage(Message) returns (MessageAck);
+  rpc Ping (PingRequest) returns (PingResponse);
+}
+message PingRequest {}  // 空请求
+
+message PingResponse {
+  string status = 1;  // 可以返回服务器状态
 }
 message JoinRequest {
   string user_id = 1;

rpc/service/chatServer.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/gogf/gf/v2/util/gconv"
-	"google.golang.org/grpc"
 	"log"
 	. "rpc/chat"
 	"rpc/config"
@@ -12,25 +11,39 @@ import (
 	"time"
 )
 
-type chatServer struct {
+var Chatserver *ChatServer
+
+func init() {
+	Chatserver = NewChatServer()
+}
+
+type ChatServer struct {
 	UnimplementedChatServiceServer
-	clients  map[string]chan *Message
-	adminMsg chan *Message
-	mu       sync.RWMutex
+	clients      map[string]chan *Message
+	adminMsg     chan *Message
+	mu           sync.RWMutex
+	shutdownChan chan struct{} // 关闭信号通道
 }
 
-var ChatSrv = &chatServer{
-	clients:  make(map[string]chan *Message),
-	adminMsg: make(chan *Message, 100),
+func NewChatServer() *ChatServer {
+	return &ChatServer{
+		clients:      make(map[string]chan *Message),
+		adminMsg:     make(chan *Message, 100),
+		shutdownChan: make(chan struct{}),
+	}
 }
 
 // 建立连接
-func (s *chatServer) JoinChat(req *JoinRequest, stream ChatService_JoinChatServer) error {
+func (s *ChatServer) JoinChat(req *JoinRequest, stream ChatService_JoinChatServer) error {
 	// 创建新通道
 	msgChan := make(chan *Message, 100)
 
 	// 注册客户端
 	s.mu.Lock()
+	if _, exists := s.clients[req.UserId]; exists {
+		s.mu.Unlock()
+		return fmt.Errorf("用户 %s 已连接", req.UserId)
+	}
 	s.clients[req.UserId] = msgChan
 	s.mu.Unlock()
 
@@ -41,46 +54,39 @@ func (s *chatServer) JoinChat(req *JoinRequest, stream ChatService_JoinChatServe
 		Timestamp: time.Now().Unix(),
 	}
 	if err := stream.Send(welcomeMsg); err != nil {
-		s.mu.Lock()
-		delete(s.clients, req.UserId)
-		s.mu.Unlock()
+		s.removeClient(req.UserId)
 		return err
 	}
-
 	// 清理处理
-	defer func() {
-		s.mu.Lock()
-		if ch, exists := s.clients[req.UserId]; exists {
-			delete(s.clients, req.UserId)
-			close(ch)
-		}
-		s.mu.Unlock()
-	}()
+	defer s.removeClient(req.UserId)
+	// 消息循环
 	for {
 		select {
 		case msg := <-msgChan:
-			if err := stream.Send(msg); err != nil {
+			if err := s.sendWithTimeout(stream, msg, 5*time.Second); err != nil {
 				return err
 			}
 		case adminMsg := <-s.adminMsg:
-			if err := stream.Send(adminMsg); err != nil {
+			if err := s.sendWithTimeout(stream, adminMsg, 5*time.Second); err != nil {
 				return err
 			}
 		case <-stream.Context().Done():
 			return nil
+		case <-s.shutdownChan:
+			return nil
 		}
 	}
 }
 
 // 接收消息处理
-func (s *chatServer) SendMessage(ctx context.Context, msg *Message) (*MessageAck, error) {
+func (s *ChatServer) SendMessage(ctx context.Context, msg *Message) (*MessageAck, error) {
 	msg.Timestamp = time.Now().Unix()
-	log.Printf("收到来自 %s 的 %s 消息: %s", msg.UserId, msg.Action, msg.Text)
+	log.Printf("收到来自 %s 的 %s 消息: %s\n", msg.UserId, msg.Action, msg.Text)
 
 	// 先处理业务逻辑
 	switch msg.Action {
 	case "getContacts":
-		log.Printf("接收%s通讯录信息", msg.UserId)
+		log.Printf("接收%s通讯录信息\n", msg.UserId)
 		//go SynchronousContacts(msg.UserId, msg.Text)
 	case "chatHistory":
 		go AddChatRecord(msg.UserId, msg.Text) // 异步处理
@@ -99,7 +105,7 @@ func (s *chatServer) SendMessage(ctx context.Context, msg *Message) (*MessageAck
 		select {
 		case ch <- msg:
 		default:
-			log.Printf("客户端 %s 的消息通道已满", userId)
+			log.Printf("客户端 %s 的消息通道已满\n", userId)
 		}
 	}
 
@@ -107,7 +113,7 @@ func (s *chatServer) SendMessage(ctx context.Context, msg *Message) (*MessageAck
 }
 
 // SendAdminMessage 向指定用户发送系统消息
-func (s *chatServer) SendAdminMessage(userId string, text string, action string) error {
+func (s *ChatServer) SendAdminMessage(userId string, text string, action string) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	// 检查目标用户是否存在
@@ -125,41 +131,96 @@ func (s *chatServer) SendAdminMessage(userId string, text string, action string)
 	// 发送消息
 	select {
 	case msgChan <- msg:
-		log.Printf("已向用户 %s 发送系统消息: %s", userId, text)
+		log.Printf("已向用户 %s 发送系统消息: %s\n", userId, text)
 		return nil
 	default:
 		return fmt.Errorf("用户 %s 的消息通道已满", userId)
 	}
 }
+func (s *ChatServer) StartTimedMessages(ctx context.Context, interval time.Duration, action string) {
+	// 立即执行一次任务
+	s.executeTimedAction(ctx, action)
 
-// StartTimedMessages 启动定时消息发送
-func (s *chatServer) StartTimedMessages(interval time.Duration, action string) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
-	for range ticker.C {
-		message := fmt.Sprintf("系统定时消息: 当前时间 %v", time.Now().Format("2006-01-02 15:04:05"))
-		// 快速获取客户端列表
-		s.mu.RLock()
-		clients := make([]string, 0, len(s.clients))
-		for userId := range s.clients {
-			clients = append(clients, userId)
+
+	for {
+		select {
+		case <-ticker.C:
+			s.executeTimedAction(ctx, action)
+		case <-ctx.Done():
+			log.Printf("定时任务[%s]已停止", action)
+			return
+		case <-s.shutdownChan:
+			log.Printf("服务关闭，停止定时任务[%s]", action)
+			return
 		}
-		s.mu.RUnlock()
-
-		// 处理发送（无需在锁内）
-		switch action {
-		case "getContacts":
-			s.BroadcastAdminMessage(message, "getContacts")
-		case "sendTalk":
-			Task()
-		case "listenIn":
-			s.BroadcastAdminMessage(message, "isnline")
+	}
+}
+
+func (s *ChatServer) executeTimedAction(ctx context.Context, action string) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("定时任务[%s]执行出错: %v\n", action, r)
 		}
+	}()
+
+	startTime := time.Now()
+	log.Printf("开始执行定时任务[%s]\n", action)
+	message := fmt.Sprintf("系统定时消息: 当前时间 %v", startTime.Format("2006-01-02 15:04:05"))
+	// 使用更安全的方式获取客户端列表
+	clients := s.getClientsSnapshot()
+	if len(clients) > 0 {
+		log.Printf("当前在线客户端数: %d\n", len(clients))
+	}
+
+	// 根据action执行不同操作
+	switch action {
+	case "getContacts":
+		s.BroadcastAdminMessage(message, "getContacts")
+	case "sendTalk":
+		s.executeTask(ctx)
+	case "heartbeat":
+		s.BroadcastAdminMessage(message, "heartbeat")
+	default:
+		log.Printf("未知的定时任务类型: %s\n", action)
+	}
+
+	log.Printf("完成定时任务[%s], 耗时: %v \n", action, time.Since(startTime))
+}
+
+func (s *ChatServer) getClientsSnapshot() []string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	clients := make([]string, 0, len(s.clients))
+	for userId := range s.clients {
+		clients = append(clients, userId)
+	}
+	return clients
+}
+
+func (s *ChatServer) executeTask(ctx context.Context) {
+	// 为Task操作添加超时控制
+	taskCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		Task() // 假设Task是您定义的任务函数
+	}()
+
+	select {
+	case <-done:
+		log.Println("Task执行完成")
+	case <-taskCtx.Done():
+		log.Println("Task执行超时")
 	}
 }
 
 // BroadcastAdminMessage 向所有客户端广播系统消息
-func (s *chatServer) BroadcastAdminMessage(text string, action string) {
+func (s *ChatServer) BroadcastAdminMessage(text string, action string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	msg := &Message{
@@ -171,15 +232,15 @@ func (s *chatServer) BroadcastAdminMessage(text string, action string) {
 	for userId, ch := range s.clients {
 		select {
 		case ch <- msg:
-			log.Printf("已广播系统消息到用户 %s: %s", userId, text)
+			log.Printf("已广播系统消息到用户 %s: %s\n", userId, text)
 		default:
-			log.Printf("用户 %s 的消息通道已满，无法广播", userId)
+			log.Printf("用户 %s 的消息通道已满，无法广播\n", userId)
 		}
 	}
 }
 
 // SpecifyAdminMessage 向制定客户端广播系统消息
-func (s *chatServer) SpecifyAdminMessage(taskId int64, userMap map[string]interface{}, contentData *[]map[string]interface{}, action, batchCode string) error {
+func (s *ChatServer) SpecifyAdminMessage(taskId int64, userMap map[string]interface{}, contentData *[]map[string]interface{}, action, batchCode string) error {
 	userId := gconv.String(userMap["userId"])
 	isRefuse := gconv.Int64(userMap["isRefuse"])
 	if isRefuse == 1 {
@@ -221,22 +282,22 @@ func (s *chatServer) SpecifyAdminMessage(taskId int64, userMap map[string]interf
 	}
 	select {
 	case ch <- msg:
-		log.Printf("系统消息已发送到用户 %s: %s (Action: %s)", userId, text, action)
+		log.Printf("系统消息已发送到用户 %s: %s (Action: %s)\n", userId, text, action)
 		return nil
 	default:
-		log.Printf("用户 %s 的消息通道已满，丢弃消息", userId)
+		log.Printf("用户 %s 的消息通道已满，丢弃消息\n", userId)
 		return fmt.Errorf("用户 %s 的消息通道阻塞", userId)
 	}
 }
 
 // SpecifysystemMessage 向制定客户端广播系统消息(拒绝也发，不保存发送记录)
-func (s *chatServer) SpecifysystemMessage(userId, wxId string, contentData map[string]interface{}, action string) error {
+func (s *ChatServer) SpecifysystemMessage(userId, wxId string, contentData map[string]interface{}, action string) error {
 	// 1. 加锁并获取用户channel
 	s.mu.Lock()
 	ch, exists := s.clients[userId]
 	if !exists {
 		s.mu.Unlock()
-		log.Printf("用户 %s 不存在或已离线 (wxId: %s)", userId, wxId)
+		log.Printf("用户 %s 不存在或已离线 (wxId: %s)\n", userId, wxId)
 		return fmt.Errorf("user %s not found", userId)
 	}
 
@@ -268,15 +329,72 @@ func buildMessageText(contentData map[string]interface{}, wxId string) string {
 func trySendMessage(ch chan<- *Message, msg *Message, userId, action string) error {
 	select {
 	case ch <- msg:
-		log.Printf("系统消息发送成功 | 用户: %s | 动作: %s", userId, action)
+		log.Printf("系统消息发送成功 | 用户: %s | 动作: %s\n", userId, action)
 		return nil
 	default:
-		log.Printf("消息通道已满 | 用户: %s | 动作: %s", userId, action)
+		log.Printf("消息通道已满 | 用户: %s | 动作: %s\n", userId, action)
 		return fmt.Errorf("message queue full for user %s", userId)
 	}
 }
-func logPings(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
-	// 只有客户端调用 RPC 方法时才会执行这里！
-	log.Println("RPC called:", info.FullMethod)
-	return handler(ctx, req)
+
+func (s *ChatServer) Ping(ctx context.Context, req *PingRequest) (*PingResponse, error) {
+	return &PingResponse{Status: "OK"}, nil // 确认返回值类型匹配
+}
+
+// Shutdown 优雅关闭服务
+func (s *ChatServer) Shutdown() {
+	close(s.shutdownChan)
+
+	// 关闭所有客户端连接
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for userId, ch := range s.clients {
+		close(ch)
+		delete(s.clients, userId)
+	}
+}
+
+// removeClient 安全地移除客户端连接
+func (s *ChatServer) removeClient(userId string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if ch, exists := s.clients[userId]; exists {
+		// 关闭通道前先检查是否已关闭
+		select {
+		case _, ok := <-ch:
+			if ok {
+				close(ch) // 只有通道未关闭时才关闭它
+			}
+		default:
+			close(ch)
+		}
+		delete(s.clients, userId)
+		log.Printf("客户端 %s 已断开连接", userId)
+	}
+}
+
+// sendWithTimeout 带超时的消息发送
+func (s *ChatServer) sendWithTimeout(stream ChatService_JoinChatServer, msg *Message, timeout time.Duration) error {
+	ctx, cancel := context.WithTimeout(stream.Context(), timeout)
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		done <- stream.Send(msg)
+	}()
+
+	select {
+	case err := <-done:
+		return err
+	case <-ctx.Done():
+		// 超时后检查原始上下文是否已取消
+		select {
+		case <-stream.Context().Done():
+			return stream.Context().Err()
+		default:
+			return fmt.Errorf("消息发送超时")
+		}
+	}
 }

rpc/service/service.go

@@ -20,9 +20,9 @@ var (
 )
 
 func Task() {
+	log.Println("定时任务查询")
 	allUserMap := InitUser()
-	go ScheduledTasks(allUserMap, 0)
-	log.Println(allUserMap)
+	//go ScheduledTasks(allUserMap, 0)
 	go RecurringTask(allUserMap)
 }
 
@@ -40,14 +40,16 @@ func ScheduledTasks(allUserMap map[int64]map[string]interface{}, newtaskId int64
 		return
 	}
 	for _, v := range *taskData {
-		batchCode := FindBatch()
+
 		sendGroupId := gconv.String(v["send_group_id"])
 		taskType := gconv.Int64(v["task_type"])
 		taskId := gconv.Int64(v["id"])
 		sendTimeType := gconv.Int64(v["send_time_type"])
+		log.Println(taskId, "一次性任务处理", sendTimeType, gconv.String(v["send_time"]), IsSameMinute(now, gconv.String(v["send_time"]), "2006-01-02 15:04:05"))
 		if sendTimeType == 1 && !IsSameMinute(now, gconv.String(v["send_time"]), "2006-01-02 15:04:05") && newtaskId == 0 {
 			continue
 		}
+		batchCode := FindBatch()
 		go UserHandle(sendGroupId, batchCode, taskType, taskId, allUserMap)
 	}
 }
@@ -61,14 +63,19 @@ func RecurringTask(allUserMap map[int64]map[string]interface{}) {
 		return
 	}
 	for _, v := range *taskData {
-		batchCode := FindBatch()
 		sendGroupId := gconv.String(v["send_group_id"])
 		taskType := gconv.Int64(v["task_type"])
 		taskId := gconv.Int64(v["id"])
-		sendTimeType := gconv.Int64(v["send_time_type"])
-		if sendTimeType == 1 && !IsSameMinute(now, gconv.String(v["send_time"]), "2006-01-02 15:04:05") {
+		if gconv.String(v["send_time"]) == "" {
 			continue
 		}
+		log.Println(taskId, "循环任务处理", gconv.String(v["send_time"]), IsSameTime(now, gconv.String(v["send_time"]), "15:04:05"))
+
+		if !IsSameTime(now, gconv.String(v["send_time"]), "15:04:05") {
+			continue
+		}
+		log.Println(1111)
+		batchCode := FindBatch()
 		go UserHandle(sendGroupId, batchCode, taskType, taskId, allUserMap)
 
 	}
@@ -80,19 +87,27 @@ func UserHandle(sendGroupId, batchCode string, taskType int64, taskId int64, all
 	if taskType == 1 {
 		//循环任务
 		//查询历史发送记录
-		recorData := config.WxRobot.SelectBySql("select  * from  send_record  where   task_id =? and  send_status in  (0,3)", taskId)
+		recorData := config.WxRobot.SelectBySql("select  * from  send_record  where   task_id =? and  send_status in  (0)", taskId)
 		for _, m := range *recorData {
 			baseUserId := gconv.Int64(m["base_user_id"])
 			sendsuccessMap[baseUserId] = true
 		}
 	}
 	//查询分组人员信息
+	log.Println(sendGroupId, "分组人员查询")
 	uic := NewUserIdConstructor(strings.Split(sendGroupId, ","), 0)
 	baseUserIdList := uic.QueryBaseUserIDList()
 	if len(baseUserIdList) == 0 {
 		log.Println("QueryBaseUserIdList 未获取到有效用户base_user_id")
+		config.WxRobot.Update("task", map[string]interface{}{
+			"id": taskId,
+		}, map[string]interface{}{
+			"task_status": util.If(taskType == 1, 1, 3),
+		})
 		return
 	}
+	//分组人员数量
+	log.Println(sendGroupId, "分组人员数量", len(baseUserIdList))
 	//排除掉未建联的人
 	newUserArr := []map[string]interface{}{}
 	for _, v := range baseUserIdList {
@@ -109,6 +124,11 @@ func UserHandle(sendGroupId, batchCode string, taskType int64, taskId int64, all
 
 	}
 	if len(newUserArr) == 0 {
+		config.WxRobot.Update("task", map[string]interface{}{
+			"id": taskId,
+		}, map[string]interface{}{
+			"task_status": util.If(taskType == 1, 1, 3),
+		})
 		return
 	}
 	//获取发送内容
@@ -119,11 +139,11 @@ func UserHandle(sendGroupId, batchCode string, taskType int64, taskId int64, all
 		log.Println("任务内容获取失败：", taskId)
 		return
 	}
-
+	log.Println(taskId, "需要分发的数量", len(newUserArr))
 	for _, v := range newUserArr {
 		log.Println(v)
 		//发送消息
-		go ChatSrv.SpecifyAdminMessage(taskId, v, contentData, "sendTalk", batchCode)
+		go Chatserver.SpecifyAdminMessage(taskId, v, contentData, "sendTalk", batchCode)
 
 	}
 	config.WxRobot.Update("task", map[string]interface{}{
@@ -146,6 +166,15 @@ func IsSameMinute(timeStr1, timeStr2, layout string) bool {
 	return t1.Truncate(time.Minute).Equal(t2.Truncate(time.Minute))
 }
 
+func IsSameTime(timeStr1, timeStr2, layout string) bool {
+	t1Time, err := time.Parse(time.DateTime, timeStr1)
+	if err != nil {
+		return false
+	}
+	t1Str := t1Time.Truncate(time.Minute).Format(layout)
+	return t1Str == timeStr2
+}
+
 // 通讯录同步
 func SynchronousContacts(robotCode, contactsStr string) {
 	config.WxRobot.Update("user_address_book", map[string]interface{}{
@@ -229,7 +258,7 @@ func AddChatRecord(robotCode string, data string) {
 					"update_time": time.Now().Format(date.Date_Full_Layout),
 				})
 				//发送一条 拒绝自动回复
-				ChatSrv.SpecifysystemMessage(robotCode, wxId, map[string]interface{}{
+				Chatserver.SpecifysystemMessage(robotCode, wxId, map[string]interface{}{
 					"content":      config.DbConf.ReplyAutomatically,
 					"content_type": 0,
 				}, "reject")
@@ -250,8 +279,7 @@ func AddChatRecord(robotCode string, data string) {
 
 // 用户信息初始化
 func InitUser() map[int64]map[string]interface{} {
-	allUserMap :=
-		make(map[int64]map[string]interface{})
+	allUserMap := make(map[int64]map[string]interface{})
 	config.WxRobot.SelectByBath(1000, func(l *[]map[string]interface{}) bool {
 		for _, v := range *l {
 			baseUserId := gconv.Int64(v["base_user_id"])
@@ -306,7 +334,7 @@ func FindBatch() string {
 	yesterdayKey := fmt.Sprintf("wxRobot_%s", yesterday)
 	// 2. 异步清理昨日数据（不影响主流程）
 	if exists, _ := redis.Exists("newother", yesterdayKey); exists {
-		redis.Del(yesterdayKey) // 不等待结果
+		redis.Del("newother", yesterdayKey) // 不等待结果
 	}
 	// 3. 获取/创建今日批次号
 	incrResult := redis.Incr("newother", todayKey)

rpc/service/wxRobot.go

@@ -20,8 +20,8 @@ func init() {
 }
 
 type TaskRequest struct {
-	TaskID        int64                    `json:"task_id,omitempty"`
-	TaskName      string                   `json:"task_name"`
+	TaskID        int64                    `json:"taskId,omitempty"`
+	TaskName      string                   `json:"taskName"`
 	SendGroupID   string                   `json:"sendGroupId"`
 	SendGroupName string                   `json:"sendGroupName"`
 	TaskType      int64                    `json:"taskType"`
@@ -53,7 +53,7 @@ func (a *WXroBot) AddTaskHandler() {
 			"create_person_id": userid,
 			"send_group_name":  req.SendGroupName,
 		})
-		if taskId == 0 {
+		if taskId <= 0 {
 			msg = "保存失败"
 		} else {
 			for _, v := range content {
@@ -130,7 +130,6 @@ func (a *WXroBot) UpdateTaskHandler() {
 					allUserMap := InitUser()
 					ScheduledTasks(allUserMap, taskId)
 				}()
-
 			}
 		} else {
 			msg = "修改失败"