消息队列 RocketMQ版
消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
产品动态
用户指南
实例管理
后台任务管理
Terraform
RocketMQ Agent
API 参考
实例管理
Topic管理
Group管理
消息查询
死信队列
密钥管理
风险巡检
白名单管理
终端节点服务
SDK 参考
RocketMQ 5.x
RocketMQ 4.x
RocketMQ 管控面SDK
最佳实践
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 4.xPython SDK普通消息
复制全文
下载 pdf普通消息
文档反馈
问问助手火山引擎消息队列 RocketMQ版提供同步发送和单向（Oneway）发送两种方式来发送普通消息。本文介绍如何通过不同方式发送普通消息。
前提条件
- 已完成准备工作。
- 已阅读参数说明，了解常用参数的配置方式与填写格式。
发送方式
火山引擎消息队列 RocketMQ版提供的普通消息发送方式包括以下三种，您可以根据业务要求选择合适的发送方式。
/ | 同步发送 | 单向（Oneway）发送 | 单向发送 |
/ | 同步发送 | 单向（Oneway）发送 | 单向发送 |
发送方式 | 消息发送方发出一条消息到服务端，服务端接收并处理消息，然后返回响应给发送方，发送方收到响应后才会发送下一条消息。 | 发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。 | 发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。 |
应用场景 | 重要通知邮件、报名短信通知、营销短信系统等。 | 适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。 | 适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。 |
发送TPS | 快 | 最快 | 最快 |
是否反馈发送结果 | 反馈 | 不反馈 | 不反馈 |
可靠性 | 不丢失消息 | 可能丢失消息 | 可能丢失消息 |
同步发送
同步发送是指消息发送方发出一条消息后，会在收到服务端返回响应之后才发下一条消息的通讯方式。一般用于较为重要的消息发送场景。
同步发送方式发送普通消息的示例代码如下。
from rocketmq.client import Producer, Message
producer_group = "" # 生产者group
name_server_addr = "http://rocketmq-xxxxxxxx.rocketmq.ivolces.com:9876" # 火山引擎控制台展示的TCP接入点
topic = "" # 在火山引擎控制台Topic管理页面创建的topic名称
access_key = "" # RocketMQ实例密钥管理页面获取到的AccessKey ID
access_secret = "" # RocketMQ实例密钥管理页面获取到的AccessKey Secret
key = "" # 消息key
tag = "" # 消息tag
# 创建并启动生产者实例
producer = Producer(producer_group)
producer.set_name_server_address(name_server_addr)
producer.set_session_credentials(access_key, access_secret, "")
producer.start()
# 组装消息
msg = Message(topic)
msg.set_keys(key)
msg.set_tags(tag)
msg.set_body("hello volcano engine")
ret = producer.send_sync(msg) #同步发送消息
print(ret.status, ret.msg_id, ret.offset)
# 关闭生产者实例，释放资源
producer.shutdown()
单向发送
发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。
单向发送适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集等场景。
单向发送的示例代码如下。
from rocketmq.client import Producer, Message
producer_group = "" # 生产者group
name_server_addr = "http://rocketmq-xxxxxxxx.rocketmq.ivolces.com:9876" # 火山引擎控制台展示的TCP接入点
topic = "" # 在火山引擎控制台Topic管理页面创建的topic名称
access_key = "" # RocketMQ实例密钥管理页面获取到的AccessKey ID
access_secret = "" # RocketMQ实例密钥管理页面获取到的AccessKey Secret
key = "" # 消息key
tag = "" # 消息tag
# 创建并启动生产者实例
producer = Producer(producer_group)
producer.set_name_server_address(name_server_addr)
producer.set_session_credentials(access_key, access_secret, "")
producer.start()
# 组装消息
msg = Message(topic)
msg.set_keys(key)
msg.set_tags(tag)
msg.set_body("hello volcano engine")
ret = producer.send_oneway(msg) # 单向发送消息
print(ret.status, ret.msg_id, ret.offset)
# 关闭生产者实例，释放资源
producer.shutdown()
订阅普通消息
集群模式消费
集群模式消费消息的示例代码如下。
import time
from rocketmq.client import PushConsumer, ConsumeStatus
name_server_addr = "http://rocketmq-xxxxxxxx.rocketmq.ivolces.com:9876" # 火山引擎控制台展示的TCP接入点
topic = "" # 在火山引擎控制台Topic管理页面创建的topic名称
group = "" # 在火山引擎控制台Group管理页面创建的group名称
access_key = "" # RocketMQ实例密钥管理页面获取到的AccessKey ID
access_secret = "" # RocketMQ实例密钥管理页面获取到的AccessKey Secret
tags = "" # 消费消息匹配的tag，多个tag使用||进行分隔
def callback(msg):
print(msg.id, msg.body)
return ConsumeStatus.CONSUME_SUCCESS
consumer = PushConsumer(group)
consumer.set_name_server_address(name_server_addr)
consumer.set_session_credentials(access_key, access_secret, "")
consumer.subscribe(topic, callback, tags)
consumer.start()
while True:
time.sleep(3600)
consumer.shutdown()
广播模式消费
广播模式消费的示例代码如下。
import time
from rocketmq.client import PushConsumer, ConsumeStatus
from rocketmq.ffi import MessageModel
name_server_addr = "http://rocketmq-xxxxxxxx.rocketmq.ivolces.com:9876" # 火山引擎控制台展示的TCP接入点
topic = "" # 在火山引擎控制台Topic管理页面创建的topic名称
group = "" # 在火山引擎控制台Group管理页面创建的group名称
access_key = "" # RocketMQ实例密钥管理页面获取到的AccessKey ID
access_secret = "" # RocketMQ实例密钥管理页面获取到的AccessKey Secret
tags = "" # 消费消息匹配的tag，多个tag使用||进行分隔
def callback(msg):
print(msg.id, msg.body)
return ConsumeStatus.CONSUME_SUCCESS
consumer = PushConsumer(group, message_model=MessageModel.BROADCASTING) # 指定消费模式为广播模式
consumer.set_name_server_address(name_server_addr)
consumer.set_session_credentials(access_key, access_secret, "")
consumer.subscribe(topic, callback, tags)
consumer.start()
while True:
time.sleep(3600)
consumer.shutdown()
最近更新时间：2025.07.09 16:22:42
这个页面对您有帮助吗？
有用
有用
无用
无用