如何用Python或API接收Firebase消息？含设备注册与监听

一、基于Firebase Cloud Messaging (FCM)的实现（推荐） #

1. 设备端注册获取InstanceID令牌 #

Python设备可借助第三方库pyfcm完成FCM注册，获取专属令牌（需提前在Firebase控制台创建项目并获取服务器密钥）。

先安装依赖：

pip install pyfcm

注册获取令牌的代码：

from pyfcm import FCMNotification

# 替换为你的Firebase服务器密钥
push_service = FCMNotification(api_key="YOUR_FCM_SERVER_KEY")

# 首次运行会生成并返回设备令牌，需将令牌同步到你的后端存储，用于后续消息定向发送
registration_id = push_service.get_registration_id()
print(f"当前设备令牌: {registration_id}")

2. 设备端监听消息 #

通过pyfcm的长连接能力实现消息监听，类似你期望的阻塞式获取逻辑：

from pyfcm import FCMClient
import time

def handle_received_message(message):
    print(f"收到消息: {message}")

# 初始化客户端，传入服务器密钥和当前设备令牌
client = FCMClient(api_key="YOUR_FCM_SERVER_KEY", registration_id="YOUR_DEVICE_TOKEN")
client.set_message_listener(handle_received_message)

# 持续监听消息
while True:
    client.check_for_messages()
    time.sleep(1)  # 控制请求频率，避免资源浪费

3. 设备间单发消息 #

借助后端存储的设备令牌列表，任意设备可向目标设备发送定向消息：

from pyfcm import FCMNotification

push_service = FCMNotification(api_key="YOUR_FCM_SERVER_KEY")

# 目标设备的令牌
target_token = "TARGET_DEVICE_REGISTRATION_ID"

# 构造并发送消息
message_content = {"title": "设备通知", "content": "这是来自其他设备的单发消息"}
send_result = push_service.single_device_data_message(
    registration_id=target_token,
    data=message_content
)

print(f"消息发送结果: {send_result}")

二、替代方案：基于MQTT的点对点通信 #

如果FCM的依赖过重，MQTT是更轻量的双向通信选择，通过Broker中转实现设备间点对点消息。

1. 安装依赖 #

pip install paho-mqtt

2. 设备端监听消息 #

import paho.mqtt.client as mqtt

# 自定义设备ID，用于标识当前设备
DEVICE_ID = "YOUR_UNIQUE_DEVICE_ID"

def on_connect(client, userdata, flags, rc):
    print(f"连接MQTT Broker成功，状态码: {rc}")
    # 订阅当前设备的专属主题
    client.subscribe(f"device/{DEVICE_ID}/messages")

def on_message(client, userdata, msg):
    print(f"收到消息: {msg.payload.decode('utf-8')}")

client = mqtt.Client(client_id=DEVICE_ID)
client.on_connect = on_connect
client.on_message = on_message

# 连接公共MQTT Broker（也可自行部署私有Broker）
client.connect("mqtt.eclipseprojects.io", 1883, 60)

# 启动持续监听
client.loop_forever()

3. 设备间发送消息 #

import paho.mqtt.client as mqtt

# 目标设备的ID
TARGET_DEVICE_ID = "TARGET_UNIQUE_DEVICE_ID"

client = mqtt.Client()
client.connect("mqtt.eclipseprojects.io", 1883, 60)

# 向目标设备的专属主题发送消息
client.publish(f"device/{TARGET_DEVICE_ID}/messages", "这是点对点的定向消息")

client.disconnect()

三、关于Google Cloud Pub/Sub的补充优化 #

Pub/Sub无需为每个设备创建独立主题，可通过订阅过滤实现定向消息：

• 所有设备订阅同一个全局主题，订阅时设置过滤条件（如device_id="xxx"）
• 发送消息时添加自定义属性（如device_id: "TARGET_DEVICE_ID"），只有匹配过滤条件的设备会收到消息
  这种方式既规避了主题数量上限，又能实现类似单发的效果。

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