国内网站免费服务器,网络营销是什么专业,如何制作电子印章,个人电脑做网站违法吗目录
1. 介绍
2. 实践
2.1. 启动消费者
2.2. 启动生产者 1. 介绍
RocketMQ是一个开源的分布式消息传递系统#xff0c;最初由阿里巴巴集团开发并于2012年开源。它旨在解决高可靠性、高吞吐量、低延迟和可伸缩性等大规模分布式系统下的消息通信需求。
RocketMQ的设计目标…目录
1. 介绍
2. 实践
2.1. 启动消费者
2.2. 启动生产者 1. 介绍
RocketMQ是一个开源的分布式消息传递系统最初由阿里巴巴集团开发并于2012年开源。它旨在解决高可靠性、高吞吐量、低延迟和可伸缩性等大规模分布式系统下的消息通信需求。
RocketMQ的设计目标是提供一种灵活、可靠、高性能的消息传递解决方案适用于各种场景包括在线消息通信、日志处理、流式处理、事件驱动架构等。
下面是对RocketMQ的主要特点和关键功能的简要介绍 分布式架构RocketMQ采用了分布式架构支持水平扩展和高可用性。它的架构包括多个消息生产者、多个消息消费者和多个消息服务器Broker。消息生产者将消息发送到Broker消息消费者从Broker订阅并消费消息。 高可靠性和容错性RocketMQ通过复制和故障转移机制提供高可靠性和容错性。它支持主从同步复制和异步复制方式确保消息不会丢失并且在Broker故障时能够自动切换到备用Broker。 高吞吐量和低延迟RocketMQ通过优化存储、网络和消息传递等方面的性能实现高吞吐量和低延迟。它支持批量发送和批量消费消息有效地减少网络开销和提高消息处理效率。 灵活的消息模型RocketMQ支持多种消息模型包括点对点P2P和发布-订阅Pub-Sub模型。在P2P模型中消息生产者直接发送消息给特定的消费者在Pub-Sub模型中消息生产者发布消息到特定的主题Topic消息消费者订阅感兴趣的主题并接收相应的消息。 丰富的消息过滤和顺序消息支持RocketMQ提供灵活的消息过滤功能可以根据消息的属性、标签或SQL表达式进行过滤。此外它还支持顺序消息确保相同主题的消息按照发送顺序被消费。 可伸缩性和扩展性RocketMQ具有良好的可伸缩性和扩展性可以根据需求增加或减少Broker、生产者和消费者的数量以适应不断增长的消息流量。 丰富的生态系统和社区支持RocketMQ拥有活跃的开源社区提供了丰富的文档、示例和工具。此外它还与其他开源项目如Apache Storm、Apache Flume、Apache Flink等集成为用户提供更多选择和灵活性。
总之RocketMQ是一个功能强大的分布式消息传递系统具有高可靠性、高吞吐量、低延迟和可伸缩性等特点。它适用于构建大规模分布式系统中的消息通信基础设施为开发者提供了一种可靠、高效的消息传递解决方案。 2. 实践 注意以下配置需从阿里云 RocketMQ 获取 HTTP_ENDPOINT xxx ACCESS_KEY xxx SECRET_KEY xxx TOPIC xxx GROUP_ID xxx INSTANCE_ID xxx 2.1. 启动消费者 The code was copied from this link
https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/send-and-subscribe-to-transactional-messages-2?spma2c4g.11186623.0.0.4b6f8707cKp9eNfrom mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_consumer import *
from mq_http_sdk.mq_client import *HTTP_ENDPOINT xxx
ACCESS_KEY xxx
SECRET_KEY xxxTOPIC xxx
GROUP_ID xxx
INSTANCE_ID xxx
# 初始化client。
mq_client MQClient(# 设置HTTP协议客户端接入点进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。HTTP_ENDPOINT,# 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。# AccessKey ID阿里云身份验证标识。ACCESS_KEY,# AccessKey Secret阿里云身份验证密钥。SECRET_KEY)
# 消息所属的Topic在消息队列RocketMQ版控制台创建。
topic_name TOPIC
# 您在消息队列RocketMQ版控制台创建的Group ID。
group_id GROUP_ID
# Topic所属的实例ID在消息队列RocketMQ版控制台创建。
# 若实例有命名空间则实例ID必须传入若实例无命名空间则实例ID传入空字符串。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
instance_id INSTANCE_IDconsumer mq_client.get_consumer(instance_id, topic_name, group_id)# 长轮询表示如果Topic没有消息则客户端请求会在服务端挂起3秒3秒内如果有消息可以消费则立即返回响应。
# 长轮询时间3秒最多可设置为30秒。
wait_seconds 3
# 一次最多消费3条最多可设置为16条。
batch 3
print((%sConsume And Ak Message From Topic%s\nTopicName:%s\nMQConsumer:%s\nWaitSeconds:%s\n \% (10 * , 10 * , topic_name, group_id, wait_seconds)))
while True:try:# 长轮询消费消息。recv_msgs consumer.consume_message(batch, wait_seconds)for msg in recv_msgs:print((Receive, MessageId: %s\nMessageBodyMD5: %s \\nMessageTag: %s\nConsumedTimes: %s \\nPublishTime: %s\nBody: %s \\nNextConsumeTime: %s \\nReceiptHandle: %s \\nProperties: %s\n % \(msg.message_id, msg.message_body_md5,msg.message_tag, msg.consumed_times,msg.publish_time, msg.message_body,msg.next_consume_time, msg.receipt_handle, msg.properties)))except MQExceptionBase as e:# Topic中没有消息可消费。if e.type MessageNotExist:print((No new message! RequestId: %s % e.req_id))continueprint((Consume Message Fail! Exception:%s\n % e))time.sleep(2)continue# msg.next_consume_time前若不确认消息消费成功则消息会被重复消费。# 消息句柄有时间戳同一条消息每次消费拿到的都不一样。try:receipt_handle_list [msg.receipt_handle for msg in recv_msgs]consumer.ack_message(receipt_handle_list)print((Ak %s Message Succeed.\n\n % len(receipt_handle_list)))except MQExceptionBase as e:print((\nAk Message Fail! Exception:%s % e))# 某些消息的句柄可能超时会导致消息消费状态确认不成功。if e.sub_errors:for sub_error in e.sub_errors:print((\tErrorHandle:%s,ErrorCode:%s,ErrorMsg:%s % \(sub_error[ReceiptHandle], sub_error[ErrorCode], sub_error[ErrorMessage]))) 2.2. 启动生产者 The code was copied from this link
https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/send-and-subscribe-to-transactional-messages-2?spma2c4g.11186623.0.0.4b6f8707cKp9eNimport sysfrom mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_producer import *
from mq_http_sdk.mq_client import *
import time
import threadingHTTP_ENDPOINT xxx
ACCESS_KEY xxx
SECRET_KEY xxxTOPIC xxx
GROUP_ID xxx
INSTANCE_ID xxx# 初始化client。
mq_client MQClient(# 设置HTTP协议客户端接入点进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。HTTP_ENDPOINT,# 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。# AccessKey ID阿里云身份验证标识。ACCESS_KEY,# AccessKey Secret阿里云身份验证密钥。SECRET_KEY)
# 消息所属的Topic在消息队列RocketMQ版控制台创建。
topic_name TOPIC
# Topic所属的实例ID在消息队列RocketMQ版控制台创建。
# 若实例有命名空间则实例ID必须传入若实例无命名空间则实例ID传入空字符串。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
instance_id INSTANCE_IDproducer mq_client.get_producer(instance_id, topic_name)# 循环发送4条消息。
msg_count 4
print(%sPublish Message To %s\nTopicName:%s\nMessageCount:%s\n % (10 * , 10 * , topic_name, msg_count))try:for i in range(msg_count):msg TopicMessage(# 消息内容。I am test message %s.hello % i,# 消息标签。tag %s % i)# 设置消息的自定义属性。msg.put_property(a, i)# 设置消息的Key。msg.set_message_key(MessageKey)re_msg producer.publish_message(msg)print(Publish Message Succeed. MessageID:%s, BodyMD5:%s % (re_msg.message_id, re_msg.message_body_md5))except MQExceptionBase as e:if e.type TopicNotExist:print(Topic not exist, please create it.)sys.exit(1)print(Publish Message Fail. Exception:%s % e)
