基于PubNub的聊天应用实时未读消息计数功能实现求助

API 探路者

2026-7-8

问题：实现实时未读消息计数与UI展示

我正在用PubNub SDK开发聊天应用，技术栈如下：

服务端：Express.js + Prisma ORM + MySQL
客户端：React.js

应用流程：

用户登录后，根据用户ID获取会话列表；
用户点击会话时，根据会话ID调用API获取消息，并通过PubNub订阅对应频道、注册事件监听器，实现实时消息收发。

当前需求：

用户A给用户B发消息时，若用户B未订阅该会话频道，消息标记为未读；
用户B订阅该频道时，自动将该会话的所有未读消息标记为已读；
实现实时未读消息计数并在UI展示。

数据库模型

model User {
  id            String         @id @default(uuid())
  email         String         @unique
  name          String?
  password      String
  createdAt     DateTime       @default(now())
  conversations Conversation[]
  messages      Message[]
  participants  Participant[]
}

model Conversation {
  id               String        @id @default(uuid())
  conversationName String?
  type             String        @default("1vs1")
  creatorUserId    String
  creatorUser      User          @relation(fields: [creatorUserId], references: [id], onDelete: Cascade)
  timestamp        DateTime      @default(now())
  messages         Message[]
  participants     Participant[]
}

model Message {
  id             String        @id @default(uuid())
  conversationId String
  conversation   Conversation  @relation(fields: [conversationId], references: [id], onDelete: Cascade)
  senderUserId   String
  senderUser     User          @relation(fields: [senderUserId], references: [id], onDelete: Cascade)
  content        String?
  messageType    String
  fileUrl        String?
  emojiCode      String?
  timestamp      DateTime      @default(now())
  isRead         Boolean       @default(false)
}

model Participant {
  id                String       @id @default(uuid())
  conversationId    String
  conversation      Conversation @relation(fields: [conversationId], references: [id], onDelete: Cascade)
  userId            String
  user              User         @relation(fields: [userId], references: [id], onDelete: Cascade)
}

解决方案

一、服务端修改

1. 新增会话列表的未读消息计数逻辑

修改获取用户会话列表的API，计算每个会话中当前用户的未读消息数：

// conversationAPI.js
async function getConversationsForUser(userId) {
  const conversations = await prisma.conversation.findMany({
    where: {
      participants: { some: { userId } }
    },
    include: {
      participants: { include: { user: true } },
      messages: { take: 1, orderBy: { timestamp: 'desc' } }
    },
    orderBy: { timestamp: 'desc' }
  });

  return Promise.all(conversations.map(async (conversation) => {
    // 统计当前用户的未读消息数（排除自己发送的）
    const unreadCount = await prisma.message.count({
      where: {
        conversationId: conversation.id,
        senderUserId: { not: userId },
        isRead: false
      }
    });
    // 提取会话对方的信息（排除当前用户）
    const participantInfo = conversation.participants.find(p => p.userId !== userId)?.user;
    
    return {
      ...conversation,
      participantInfo,
      unreadMessageCount: unreadCount
    };
  }));
}

2. 新增批量标记消息为已读的API

创建接口用于用户订阅会话时，批量更新未读状态：

// messageAPI.js
async function markMessagesAsRead(conversationId, userId) {
  return await prisma.message.updateMany({
    where: {
      conversationId,
      senderUserId: { not: userId },
      isRead: false
    },
    data: { isRead: true }
  });
}

二、客户端修改

1. 订阅会话时标记消息为已读

修改MessageListComponent，在订阅频道后调用标记已读接口，并触发会话列表更新：

// MessageListComponent.js
useEffect(() => {
  const markMessagesRead = async () => {
    try {
      await messageAPIs.markMessagesAsRead(conversationId, user.id);
      // 触发自定义事件通知Chat组件更新会话列表
      window.dispatchEvent(new CustomEvent('updateConversations'));
    } catch (error) {
      console.error('标记消息已读失败:', error);
    }
  };

  addListeners(pn, {
    message: handleNewMessage,
    presence: handlePresence
  });
  subscribeToChannels(pn, [conversationId]);
  markMessagesRead(); // 订阅频道时立即标记已读

  return () => {
    pn.removeListener({ message: handleNewMessage, presence: handlePresence });
    unsubscribeAll(pn);
  };
}, [conversationId, pn]);

2. 监听会话更新事件，刷新会话列表

修改Chat组件，添加自定义事件监听，实时更新会话的未读计数：

// Chat.js
useEffect(() => {
  const updateConversations = async () => {
    try {
      const response = await conversationAPIs.getConversationsForUser(user.id);
      setConversations(response.data.conversations);
    } catch (error) {
      setError(error);
    }
  };

  // 监听自定义事件，更新会话列表
  window.addEventListener('updateConversations', updateConversations);

  // 初始加载会话
  const fetchConversations = async () => {
    try {
      const response = await conversationAPIs.getConversationsForUser(user.id);
      setConversations(response.data.conversations);
    } catch (error) {
      setError(error);
    } finally {
      setLoading(false);
    }
  };
  fetchConversations();

  return () => window.removeEventListener('updateConversations', updateConversations);
}, [user.id]);

3. 实时接收消息时更新未读计数

修改MessageListComponent的消息监听逻辑，收到新消息后触发会话列表更新：

// MessageListComponent.js
const handleNewMessage = useCallback((envelope) => {
  setMessages(prevMessages => [...prevMessages, envelope.message]);
  // 如果当前用户是消息接收者，更新会话未读计数
  if (envelope.message.senderUserId !== user.id) {
    window.dispatchEvent(new CustomEvent('updateConversations'));
  }
}, [user.id]);

4. 完善未读计数的UI展示

确保Conversation组件正确渲染未读计数（示例）：

// Conversation.js
function Conversation({ name, convoId, date, unread, onClick }) {
  return (
    <div onClick={onClick} className="p-3 border-b hover:bg-gray-50 cursor-pointer">
      <div className="flex justify-between items-start">
        <h3 className="font-medium">{name}</h3>
        {unread > 0 && (
          <span className="bg-blue-500 text-white text-xs rounded-full w-5 h-5 flex items-center justify-center">
            {unread}
          </span>
        )}
      </div>
      <p className="text-sm text-gray-500">{date}</p>
    </div>
  );
}

三、可选：利用PubNub Presence优化在线状态检测

通过Presence事件精准检测用户加入频道的动作，触发标记已读：

// MessageListComponent.js
const handlePresence = useCallback((e) => {
  const { action, uuid } = e;
  // 当前用户加入频道时标记消息已读
  if (action === 'join' && uuid === user.id) {
    messageAPIs.markMessagesAsRead(conversationId, user.id)
      .then(() => window.dispatchEvent(new CustomEvent('updateConversations')))
      .catch(err => console.error(err));
  }
}, [user.id, conversationId]);

内容的提问来源于stack exchange，提问作者Nguyễn Văn Tiến