无需Firebase Cloud Functions：Flutter聊天应用消息推送方案咨询

方案一：在你的Python服务器上实现Firestore实时监听 #

你已经搭建了Python服务器，完全可以利用Firebase Admin SDK的Firestore实时监听功能，监听聊天消息集合的新增文档，一旦检测到新消息，就提取聊天成员的FCM令牌，调用你已有的sendPush函数发送通知。

具体实现步骤 #

• 监听目标消息集合的快照变化，过滤出新增的文档
• 从新消息文档中获取聊天会话ID，再从对应的会话集合中获取参与成员的FCM令牌（注意排除发送消息的用户自己）
• 调用sendPush发送通知

示例代码 #

import firebase_admin
from firebase_admin import credentials, messaging, firestore
import time

cred = credentials.Certificate("C:/Users/MASTER/PycharmProjects/pythonFCMTest/serviceAccountKey.json")
firebase_admin.initialize_app(cred)
db = firestore.client()

def sendPush(title, msg, registration_token, dataObject=None):
    message = messaging.MulticastMessage(
        notification=messaging.Notification(
            title=title,
            body=msg
        ),
        data=dataObject,
        tokens=registration_token,
    )
    response = messaging.send_multicast(message)
    print('Successfully sent message:', response)

def listen_to_new_messages():
    # 监听messages集合的新增文档
    messages_ref = db.collection('messages')
    
    def on_snapshot(doc_snapshot, changes, read_time):
        for change in changes:
            if change.type.name == 'ADDED':
                # 提取新消息数据
                msg_data = change.document.to_dict()
                sender_id = msg_data['sender_id']
                chat_id = msg_data['chat_id']
                message_content = msg_data['content']
                
                # 获取聊天会话的所有成员（排除发送者）
                chat_doc = db.collection('chats').document(chat_id).get()
                if chat_doc.exists:
                    members = chat_doc.to_dict()['members']
                    member_tokens = []
                    
                    # 批量获取成员的FCM令牌（假设users集合存储了token字段）
                    for member_id in members:
                        if member_id != sender_id:
                            user_doc = db.collection('users').document(member_id).get()
                            if user_doc.exists and 'fcm_token' in user_doc.to_dict():
                                member_tokens.append(user_doc.to_dict()['fcm_token'])
                    
                    # 发送通知
                    if member_tokens:
                        sendPush(
                            title="新消息",
                            msg=message_content,
                            registration_token=member_tokens,
                            dataObject={'chat_id': chat_id}
                        )
    
    # 启动实时监听
    messages_ref.on_snapshot(on_snapshot)
    print("已启动新消息监听...")

# 保持服务器持续运行
if __name__ == '__main__':
    listen_to_new_messages()
    while True:
        time.sleep(60)

注意事项 #

• 确保Python服务器保持持续运行，否则监听会中断
• 可优化令牌查询逻辑，比如缓存用户令牌，减少Firestore查询次数
• 增加异常处理，避免网络波动、令牌失效等情况导致监听崩溃

方案二：Flutter客户端本地监听（仅作补充） #

你可以在Flutter客户端使用cloud_firestore包监听消息集合变化，检测到新消息时触发本地通知。但该方案有明显局限性：

• 仅在APP前台或后台活跃状态下生效，APP关闭后无法触发通知
• 无法确保所有聊天成员都能收到通知（比如成员未启动APP）

示例Flutter代码片段 #

FirebaseFirestore.instance
    .collection('messages')
    .where('chat_id', isEqualTo: currentChatId)
    .snapshots()
    .listen((snapshot) {
      for (var change in snapshot.docChanges) {
        if (change.type == DocumentChangeType.added) {
          // 触发本地通知
          showLocalNotification(
            title: "新消息",
            body: change.doc['content'],
          );
        }
      }
    });

总结 #

最可靠的替代方案是方案一，依托你已有的Python服务器实现Firestore实时监听，无需依赖付费的Firebase Cloud Functions，且能确保新消息通知推送给所有在线/离线的聊天成员。

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