Shengshenlan opened a new issue, #1176: URL: https://github.com/apache/rocketmq-clients/issues/1176
### Programming Language of the Client Java ### Is Your Feature Request Related to a Problem? ```python # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from rocketmq import ClientConfiguration, Credentials, Message, Producer def handle_send_result(result_future): try: # don't write time-consuming code in the callback. if needed, use other thread res = result_future.result() print(f"send message success, {res}") except Exception as exception: print(f"send message failed, raise exception: {exception}") if __name__ == '__main__': endpoints = "foobar.com:8080" credentials = Credentials() # if auth enable # credentials = Credentials("ak", "sk") config = ClientConfiguration(endpoints, credentials) # with namespace # config = ClientConfiguration(endpoints, credentials, "namespace") topic = "topic" producer = Producer(config, (topic,)) try: producer.startup() try: for i in range(10): msg = Message() # topic for the current message msg.topic = topic msg.body = "hello, rocketmq.".encode('utf-8') # secondary classifier of message besides topic msg.tag = "rocketmq-send-message" # key(s) of the message, another way to mark message besides message id msg.keys = "send_async" # user property for the message 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"{producer} raise exception: {e}") except Exception as e: print(f"{producer} startup raise exception: {e}") input("Please Enter to Stop the Application.") producer.shutdown() print(f"{producer} shutdown.") ``` 异步发送消息的步骤太繁琐了，向做成异步还要自己写个一个回调。 ### Describe the Solution You'd Like ```python import asyncio import aio_pika async def send_message(): # 连接到 RabbitMQ 服务器 connection = await aio_pika.connect_robust( "amqp://guest:guest@localhost/" # 替换为你的 RabbitMQ 地址和凭据 ) async with connection: # 创建通道 channel = await connection.channel() # 声明一个交换机（可选，如果使用默认交换机可跳过） exchange = await channel.declare_exchange( "logs", aio_pika.ExchangeType.FANOUT, durable=True ) # 发送消息 message_body = "Hello, async RabbitMQ!" message = aio_pika.Message(body=message_body.encode()) await exchange.publish(message, routing_key="") # FANOUT 交换机忽略 routing_key print("Message sent") # 运行异步函数 asyncio.run(send_message()) ``` 参考 aio_pika 一行解决 ### Describe Alternatives You've Considered 方便异步调用。 ### Additional Context _No response_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
