ApsaraMQ for RocketMQ 5.x Python SDK ile mesaj gönderin ve alın. Her örnek belirli bir mesaj türünü kapsar ve yer tutucu değerleri kendi örnek bilgilerinizle değiştirdikten sonra çalıştırmaya hazırdır.
Ön koşullar
Başlamadan önce şunlara sahip olduğunuzdan emin olun:
Yüklenmiş RocketMQ Python SDK'sı. Sürüm ayrıntıları için bkz. Sürüm kılavuzu
Uç noktası olan bir ApsaraMQ for RocketMQ 5.x örneği. Kurulum talimatları için bkz. Ortam hazırlığı
Örnek üzerinde oluşturulmuş bir konu ve bir tüketici grubu
Parametreler
Her kod örneğindeki aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin.
| Parametre | Örnek değer | Açıklama |
|---|---|---|
<your-endpoint> |
rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080 |
ApsaraMQ for RocketMQ örneğinizin uç noktası. Uç noktayı almak için bkz. Örnek uç noktasını alma. İnternet erişimi için genel uç noktayı veya VPC erişimi için VPC uç noktasını kullanın. |
<your-instance-id> |
rmq-cn-xxx |
ApsaraMQ for RocketMQ örneğinizin ID'si. Yalnızca sunucusuz örneklerde genel ağ erişimi için gereklidir. |
<your-topic> |
normal_test |
Mesajların gönderildiği veya tüketildiği konu. Konuyu önceden oluşturun. |
<your-consumer-group> |
GID_test |
Mesajları tüketmek için kullanılan tüketici grubu. Tüketici grubunu önceden oluşturun. |
<your-ak> |
1XVg0hzgKm****** |
Örneğinizin kullanıcı adı. İnternet erişimi için gereklidir. VPC erişimi için yalnızca örnek sunucusuzsa ve VPC'lerde kimlik doğrulamasız erişim devre dışı bırakılmışsa gereklidir. Bkz. Örnek kullanıcı adı ve parolasını alma. |
<your-sk> |
ijSt8rEc45****** |
Örneğinizin parolası. Kullanıcı adı ile aynı erişim koşulları geçerlidir. |
Normal mesajlar gönderme
Normal mesajlar özel teslimat semantiğine sahip değildir ve en yaygın mesaj türüdür.
Eşzamanlı gönderme
send() yöntemi, aracı mesajı onaylayana kadar bloke olur.
from rocketmq import ClientConfiguration, Credentials, Message, Producer
if __name__ == '__main__':
# Instance endpoint
endpoints = "<your-endpoint>"
# Authenticate with your instance username and password
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
topic = "<your-topic>"
producer = Producer(config, (topic,))
try:
producer.startup()
try:
msg = Message()
msg.topic = topic
msg.body = "hello, rocketmq.".encode('utf-8')
# Tag: secondary classifier of messages besides topic
msg.tag = "tag"
# Keys: alternative way to identify messages besides message ID
msg.keys = "keys"
# Custom user property
msg.add_property("key", "value")
for i in range(10):
res = producer.send(msg)
print(f"Send message success. {res}")
producer.shutdown()
except Exception as e:
print(f"Send message failed: {e}")
producer.shutdown()
except Exception as e:
print(f"Producer startup failed: {e}")
producer.shutdown()Eşzamansız gönderme
send_async() yöntemi hemen döner ve aracı yanıt verdiğinde bir geri arama çağırır. Her onayı beklemeniz gerekmediğinde daha yüksek aktarım hızı için eşzamansız göndermeyi kullanın.
from rocketmq import ClientConfiguration, Credentials, Message, Producer
def handle_send_result(result_future):
try:
# Avoid time-consuming logic in the callback; offload to another thread if needed
res = result_future.result()
print(f"Send message success. {res}")
except Exception as exception:
print(f"Send message failed: {exception}")
if __name__ == '__main__':
endpoints = "<your-endpoint>"
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
topic = "<your-topic>"
producer = Producer(config, (topic,))
try:
producer.startup()
try:
for i in range(10):
msg = Message()
msg.topic = topic
msg.body = "hello, rocketmq.".encode('utf-8')
msg.tag = "tag"
msg.keys = "keys"
msg.add_property("send", "async")
send_result_future = producer.send_async(msg)
send_result_future.add_done_callback(handle_send_result)
except Exception as e:
print(f"Send message failed: {e}")
except Exception as e:
print(f"Producer startup failed: {e}")
input("Press Enter to stop the application.")
producer.shutdown()Sıralı mesajlar gönderme
Sıralı mesajlar aynı mesaj grubu içinde gönderildikleri sırayla teslim edilir. İlgili mesajları gruplandırmak için message_group özelliğini ayarlayın.
from rocketmq import ClientConfiguration, Credentials, Message, Producer
if __name__ == '__main__':
endpoints = "<your-endpoint>"
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
topic = "<your-topic>"
producer = Producer(config, (topic,))
try:
producer.startup()
try:
msg = Message()
msg.topic = topic
msg.body = "hello, rocketmq.".encode('utf-8')
msg.tag = "rocketmq-send-fifo-message"
msg.keys = "keys"
# Messages with the same message_group are delivered in order
msg.message_group = "fifo-group"
for i in range(10):
res = producer.send(msg)
print(f"Send message success. {res}")
producer.shutdown()
except Exception as e:
print(f"Send message failed: {e}")
producer.shutdown()
except Exception as e:
print(f"Producer startup failed: {e}")
producer.shutdown()Sıralı mesajlar için kaynak kodu şu adreste mevcuttur: fifo_producer_example.py.
Zamanlanmış ve gecikmeli mesajlar gönderme
Zamanlanmış ve gecikmeli mesajlar belirtilen bir zaman damgasından sonra teslim edilir. delivery_timestamp değerini saniye cinsinden Unix epoch zamanına ayarlayın.
import time
from rocketmq import ClientConfiguration, Credentials, Message, Producer
if __name__ == '__main__':
endpoints = "<your-endpoint>"
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
topic = "<your-topic>"
producer = Producer(config, (topic,))
try:
producer.startup()
try:
msg = Message()
msg.topic = topic
msg.body = "hello, rocketmq.".encode('utf-8')
msg.tag = "rocketmq-send-delay-message"
# Deliver the message 10 seconds from now
msg.delivery_timestamp = int(time.time()) + 10
res = producer.send(msg)
print(f"Send message success. {res}")
producer.shutdown()
except Exception as e:
print(f"Send message failed: {e}")
producer.shutdown()
except Exception as e:
print(f"Producer startup failed: {e}")
producer.shutdown()İşlemsel mesajlar gönderme
İşlemsel mesajlar iki aşamalı commit işlemini destekler. Aracı, işlenmemiş mesajları onaylamak veya geri almak için bir TransactionChecker aracılığıyla üreticiyle tekrar iletişim kurar.
from rocketmq import (ClientConfiguration, Credentials, Message, Producer,
TransactionChecker, TransactionResolution)
class OrderTransactionChecker(TransactionChecker):
"""Check whether the local transaction succeeded and return COMMIT or ROLLBACK."""
def check(self, message: Message) -> TransactionResolution:
print(f"Transaction check for message: {message}. Committing.")
return TransactionResolution.COMMIT
if __name__ == '__main__':
endpoints = "<your-endpoint>"
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
topic = "<your-topic>"
# Pass the transaction checker to the producer
producer = Producer(config, (topic,), checker=OrderTransactionChecker())
try:
producer.startup()
except Exception as e:
print(f"Producer startup failed: {e}")
try:
# Begin a half-message transaction
transaction = producer.begin_transaction()
msg = Message()
msg.topic = topic
msg.body = "hello, rocketmq.".encode('utf-8')
msg.tag = "rocketmq-send-transaction-message"
res = producer.send(msg, transaction)
print(f"Send half message success. {res}")
# Option 1: Commit from client side
transaction.commit()
print(f"Committed message: {transaction.message_id}")
# Option 2: Roll back instead
# transaction.rollback()
# print(f"Rolled back message: {transaction.message_id}")
producer.shutdown()
except Exception as e:
print(f"Transaction send failed: {e}")
producer.shutdown()Lite mesajlar gönderme
Lite mesajlar mesajları daha ince granülerlikte sınıflandırmak için alt konular (lite konular) içeren bir ana konu kullanır. Alt konuyu belirtmek için mesajda lite_topic ayarlayın.
from rocketmq import ClientConfiguration, Credentials, Message, Producer
if __name__ == '__main__':
endpoints = "<your-endpoint>"
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
# Use the parent topic
topic = "<your-topic>"
producer = Producer(config, (topic,))
try:
producer.startup()
try:
msg = Message()
msg.topic = topic
msg.body = "hello, rocketmq.".encode('utf-8')
for i in range(10):
# Route each message to a lite topic under the parent topic
msg.lite_topic = f"lite-test-{i}"
res = producer.send(msg)
print(f"Send message success. {res}")
producer.shutdown()
except Exception as e:
print(f"Send message failed: {e}")
producer.shutdown()
except Exception as e:
print(f"Producer startup failed: {e}")
producer.shutdown()Basit tüketici ile mesajları tüketme
Bir basit tüketici uzun oylama ile aracıdan mesajları çeker. Her mesajı işledikten sonra onaylamak için ack() yöntemini çağırın.
from rocketmq import (ClientConfiguration, Credentials, FilterExpression,
SimpleConsumer)
if __name__ == '__main__':
endpoints = "<your-endpoint>"
credentials = Credentials("<your-ak>", "<your-sk>")
config = ClientConfiguration(endpoints, credentials)
topic = "<your-topic>"
consumer_group = "<your-consumer-group>"
# Use a singleton consumer instance in production; avoid creating multiple consumers
simple_consumer = SimpleConsumer(config, consumer_group, {topic: FilterExpression()})
try:
simple_consumer.startup()
try:
# Optional: filter by tag
# simple_consumer.subscribe(topic, FilterExpression("your-tag"))
while True:
try:
# Long poll for up to 32 messages; each message is invisible
# for 15 seconds after receipt
messages = simple_consumer.receive(32, 15)
if messages is not None:
for msg in messages:
simple_consumer.ack(msg)
print(f"Acknowledged message: [{msg.message_id}]")
except Exception as e:
print(f"Receive or ack failed: {e}")
except Exception as e:
print(f"Consumer error: {e}")
simple_consumer.shutdown()
except Exception as e:
print(f"Consumer startup failed: {e}")
simple_consumer.shutdown()Push tüketici ile lite mesajları tüketme
Bir push tüketici kayıtlı bir dinleyici aracılığıyla mesajları otomatik olarak alır. Lite mesaj push tüketici örneği için bkz. lite_push_consumer_example.py.
Sunucusuz örnekler için genel ağ erişimi
Sunucusuz bir ApsaraMQ for RocketMQ örneğine genel ağ üzerinden erişmek için örnek ID'sini ClientConfiguration öğesinin üçüncü bağımsız değişkeni olarak iletin:
config = ClientConfiguration(endpoints, credentials, "<your-instance-id>")<your-instance-id> yerine örneğinizin gerçek ID'sini girin, örneğin rmq-cn-xxx.
GitHub kaynak dosyaları
Aşağıdaki tablo her mesaj türünü ilgili GitHub kaynak dosyasıyla eşleştirir.