使用SignalR实现用户自动加入新对话并实时收发消息的问题排查

发起新对话时，对方需刷新页面并触发Join方法后才能接收消息；希望实现创建对话时自动创建对话组，让双方自动加入组，从而即时收发消息。当前CreateConversation方法控制台可输出用户加入日志，但对方仍无法收到消息。

Angular端SignalR服务代码 #

export class SignalService {
    private hubConnection: signalR.HubConnection;
    private messageSubject = new Subject<ConversationSignalMessage>();
    private conversationCreatedSubject = new Subject<string>();
    public messageReceived$ = this.messageSubject.asObservable();
    public conversationCreated$ = this.conversationCreatedSubject.asObservable();
    
    constructor(private authService: AuthService) {
        // 修正：从authService获取当前用户，避免直接使用未定义的user变量
        const user = this.authService.getCurrentUser();
        this.hubConnection = new signalR.HubConnectionBuilder()
        .withUrl(`https://localhost:5001/sampleHub?userId=${user.id}`, { withCredentials: false})
        .build();

        this.hubConnection.on('MessageReceived', (chatId: string, message: Message) => {
            if (message.userSender.id == user.id)
                return;

            this.messageSubject.next(new ConversationSignalMessage(chatId, message));
        });

        // 新增：监听服务器推送的对话创建事件
        this.hubConnection.on('ConversationCreated', (chatId: string) => {
            this.conversationCreatedSubject.next(chatId);
        });
    }

    startConnection(): Promise<any> {
        return this.hubConnection.start();
    }

    joinChat(chatId: string, participant: Participant): void {
        this.hubConnection.invoke('Join', chatId, participant)
        .catch(err => console.error(`Error joining chat: ${err}`));
    }

    sendMessage(chatId: string, message: Message): void {
        this.hubConnection.invoke('SendMessage', chatId, message)
        .catch(err => console.error(`Error sending message: ${err}`));
    }

    createConversation(users: string[], chatId: string, message: Message): void {
        this.hubConnection.invoke('CreateConversation', users, chatId, message)
        .catch(err => console.error(`Error creating conversation and joining users: ${err}`));
    }

    getMessageStream(): Observable<ConversationSignalMessage> {
        return this.messageSubject.asObservable();
    }

    stopConnection(): Promise<any> {
        return this.hubConnection.stop();
    }
}

C#端SignalR Hub修复后代码 #

private static readonly Dictionary<string, string> _connectedUsers = new(); // Key: ConnectionId, Value: UserId
private readonly Dictionary<Guid, List<string>> _conversationParticipants = new();
private readonly Dictionary<Guid, List<Message>> _conversationMessages = new();

public async Task SendMessage(Guid chatId, Message message)
{
    // 修正：替换未定义的conversationId为参数chatId
    Console.WriteLine($"{chatId}: {message.Text}");
    await Clients.Groups(chatId.ToString()).SendAsync("MessageReceived", chatId, message).ConfigureAwait(false);
}

public async Task Join(Guid chatId, Participant participant)
{
    Console.WriteLine($"{participant.UserName} joined {chatId}.");
    await Groups.AddToGroupAsync(Context.ConnectionId, chatId.ToString()).ConfigureAwait(false);
}

public async Task LeaveConversation(Guid chatId) => await Groups.RemoveFromGroupAsync(Context.ConnectionId, chatId.ToString()).ConfigureAwait(false);

public override async Task OnConnectedAsync()
{
    string userId = Context.GetHttpContext().Request.Query["userId"];
    _connectedUsers[Context.ConnectionId] = userId;
    await base.OnConnectedAsync().ConfigureAwait(false);
    Console.WriteLine($"Client `{userId}` connected.");
}

public override async Task OnDisconnectedAsync(Exception exception)
{
    if (_connectedUsers.TryGetValue(Context.ConnectionId, out string userId))
    {
        _ = _connectedUsers.Remove(Context.ConnectionId);
        Console.WriteLine($"Client `{userId}` disconnected.");
    }

    await base.OnDisconnectedAsync(exception).ConfigureAwait(false);
}

public async Task CreateConversation(List<string> users, Guid chatId, Message message)
{
    string creatorConnectionId = Context.ConnectionId;

    // 修正：替换未定义的conversationId和userIds为参数chatId和users
    _conversationParticipants[chatId] = users;

    if (!_conversationMessages.ContainsKey(chatId))
    {
        _conversationMessages[chatId] = new List<Message>();
    }
    _conversationMessages[chatId].Add(message);

    // 核心修复：找到目标用户的ConnectionId并加入组，而非重复添加创建者的ConnectionId
    foreach (var userId in users)
    {
        // 支持多端登录：获取该用户所有在线的ConnectionId
        var targetConnectionIds = _connectedUsers
            .Where(kv => kv.Value == userId)
            .Select(kv => kv.Key);
        
        foreach (var connId in targetConnectionIds)
        {
            await Groups.AddToGroupAsync(connId, chatId.ToString()).ConfigureAwait(false);
            Console.WriteLine($"{userId} (ConnectionId: {connId}) joined {chatId}.");
        }
    }

    await Clients.Groups(chatId.ToString()).SendAsync("ConversationCreated", chatId).ConfigureAwait(false);
    await Clients.Groups(chatId.ToString()).SendAsync("MessageReceived", chatId, message).ConfigureAwait(false);
}

错误分析与修复说明 #

1. 变量名不匹配：原CreateConversation方法中使用了未定义的conversationId和userIds，已替换为方法参数chatId和users。
2. 组成员添加错误：原代码循环中重复将创建者的creatorConnectionId加入组，而非目标用户的ConnectionId。现在通过_connectedUsers字典匹配userId对应的ConnectionId，将目标用户加入对话组。
3. 前端未监听对话创建事件：Angular端新增了ConversationCreated事件监听，确保前端能实时感知新对话，无需刷新页面。
4. 多端登录支持：优化了ConnectionId查询逻辑，支持同一用户多设备在线时全部加入对话组。

内容的提问来源于stack exchange，提问作者User1899289003