messense / rocketmq-python Goto Github PK

View Code? Open in Web Editor NEW

94.0 9.0 16.0 79 KB

RocketMQ Python client

License: Apache License 2.0

Python 96.81% Shell 3.19%

rocketmq rocketmq-client

rocketmq-python's Introduction

rocketmq-python

RocketMQ Python client, based on rocketmq-client-cpp, supports Linux and macOS

This project has been upstreamed to apache/rocketmq-client-python, it's recommended to use that package instead.

Installation

pip install rocketmq

Usage

Producer

from rocketmq.client import Producer, Message

producer = Producer('PID-XXX')
producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
# producer.set_namesrv_addr('127.0.0.1:9887')
producer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
producer.start()

msg = Message('YOUR-TOPIC')
msg.set_keys('XXX')
msg.set_tags('XXX')
msg.set_body('XXXX')
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
producer.shutdown()

PushConsumer

import time

from rocketmq.client import PushConsumer


def callback(msg):
    print(msg.id, msg.body)


consumer = PushConsumer('CID_XXX')
consumer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
# consumer.set_namesrv_addr('127.0.0.1:9887')
consumer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
consumer.subscribe('YOUR-TOPIC', callback)
consumer.start()

while True:
    time.sleep(3600)

consumer.shutdown()

PullConsumer

from rocketmq.client import PullConsumer


consumer = PullConsumer('CID_XXX')
consumer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
# consumer.set_namesrv_addr('127.0.0.1:9887')
consumer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
consumer.start()

for msg in consumer.pull('YOUR-TOPIC'):
    print(msg.id, msg.body)

consumer.shutdown()

License

This work is released under the MIT license. A copy of the license is provided in the LICENSE file.

rocketmq-python's People

Contributors

Stargazers

Watchers

Forkers

francisoliverlee wangchaowei ghostsun fanren twz915 duruyi lin9long hivencent czzlegend deepln zhengwei2020 laashub-soa laashub-sda trustmeta visionyimian githql

rocketmq-python's Issues

rocketmq.exceptions.ProducerSendSyncFailed ffi_check错误

实例生产者代码中注释掉producer.set_session_credentials，其他都一样。报如下错误。
python 2.7， rocketmq 4.5

$python rocketmq_client_producer.py
Traceback (most recent call last):
  File "rocketmq_client_producer.py", line 15, in <module>
    ret = producer.send_sync(msg)
  File "/home/songchuan.zhou/env/py-env/lib/python2.7/site-packages/rocketmq/client.py", line 189, in send_sync
    ffi_check(dll.SendMessageSync(self._handle, msg, ctypes.pointer(cres)))
  File "/home/songchuan.zhou/env/py-env/lib/python2.7/site-packages/rocketmq/exceptions.py", line 19, in ffi_check
    raise exc_cls()
rocketmq.exceptions.ProducerSendSyncFailed

不知道为什么从 java 端发送的消息 body 一直为空

如题，在阿里云控制台下载消息是有内容的，但是使用PushConsumer 时，从 callback 读出来的 body 都为空，tags 和 keys没问题

No route info of this topic

I used this client with AIiyun MQ, it gives an exception No route info of this topic.
Every param I used is working with Java SDK from Aliyun Docs.

producer = Producer(group_id)

producer.set_namesrv_addr(server_addr)
# producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')

producer.set_session_credentials(app_key, secret, 'ALIYUN')

producer.start()

msg = Message(topic)

msg.set_keys("hd_test")
msg.set_tags("test")
msg.set_body("This is a test message from python.")

ret = producer.send_sync(msg)

print(ret)

producer.shutdown()

Traceback (most recent call last):
File "/Users/ahrli/Desktop/github/ce/1.py", line 15, in
ret = producer.send_sync(msg)
File "/Users/ahrli/anaconda3/lib/python3.7/site-packages/rocketmq/client.py", line 189, in send_sync
ffi_check(dll.SendMessageSync(self._handle, msg, ctypes.pointer(cres)))
File "/Users/ahrli/anaconda3/lib/python3.7/site-packages/rocketmq/exceptions.py", line 19, in ffi_check
raise exc_cls()
rocketmq.exceptions.ProducerSendSyncFailed

消费失败重试机制不起作用

def test_push_consumer_reconsume_later():
    stop_event = threading.Event()
    raised_exc = threading.Event()

    def on_message(msg):
        print(msg.body.decode('utf-8'))
        if not raised_exc.is_set():
            raised_exc.set()
            raise Exception('Should reconsume later')
        stop_event.set()

    consumer = PushConsumer('xxxx')
    consumer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
    # For ip and port name server address, use `set_namesrv_addr` method, for example:
    # consumer.set_namesrv_addr('127.0.0.1:9887')
    consumer.set_session_credentials('LTAI0Dl0VYPDiExz', 'KOFlHDMPlVboJPbBcy07wkGO6rRyeB',
                                     'ALIYUN')  # No need to call this function if you don't use Aliyun.
    consumer.subscribe('xxxxx', on_message, 'hello')
    consumer.start()
    while not stop_event.is_set():
        time.sleep(10)


test_push_consumer_reconsume_later()

我按照你的代码这样测了，但是抛了异常的那条消息并不会再来第二次，按理说应该会再来第二次然后停止阻塞终止程序的对吧？

Add Windows support

there is no documnets for python user

There is an bug when i use pull method, some message can not be pulled . i can saw the message infomations such as id, datetime and content in web site . But I can not pull it . I try many times.

Build and distribute wheels

I don't use aliyun

I don't use aliyun，Don't need certification
How do I get the message status？

RocketMQ integration tests

Can you invite me to be a maintainer of the RocketMQ PyPI package?

@messense Can you invite me to be a maintainer of the RocketMQ PyPI package? Since the community are developing the new python client in rocketmq-clients.

Here is my username: aaron-ai, Thank you!

producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet') 使用方法

我是一名初学者，初次接触python和erocketmq。
producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
这里可以使用非aliyun的host吗？可否使用ip+port的形式？

PushConsumer的callback方法抛异常，不能重新消费消息

`
def subscribe(self, topic, callback, expression='*'):
def _on_message(consumer, msg):
exc = None
try:
callback(RecvMessage(msg))
except Exception as e:
exc = e
return _CConsumeStatus.RECONSUME_LATER.value
finally:
if exc:
raise exc

        return _CConsumeStatus.CONSUME_SUCCESS.value

    ffi_check(dll.Subscribe(self._handle, _to_bytes(topic), _to_bytes(expression)))
    self._register_callback(_on_message)

**callback抛异常，消息不能重新消费，finally代码是否有问题，
重写了subscribe，去掉finally也不能重新消费消息，请问下是怎么回事
**

Relicense under Apache License

To upstream this Python client implementation to official repository, we need to relicense under Apache License.

cc @dushujun @twz915

消费模式设置问题

consumer.set_message_model() 这个方法如果不调的话，默认的消费模式是什么，还有调这个方法传入的参数应该是什么，可以给个例子吗

PushConsumer AuthenticationException: valid resource owner failed

[2019-Jan-02 16:29:24.058953](info):connect server with addr:118.190.213.23:80 success
[2019-Jan-02 16:29:24.138858](warning):get error response:1 of heartbeat to addr:118.190.213.23:80
[2019-Jan-02 16:29:24.139347](info):try to connect to fd:33, addr:118.190.213.26
[2019-Jan-02 16:29:24.139471](info):Wait till event base is looping
[2019-Jan-02 16:29:24.139553](info):Notify on event_base_dispatch
[2019-Jan-02 16:29:24.139670](info):eventcb: received event:80 on fd:33
[2019-Jan-02 16:29:24.139714](info):eventcb:connect to fd:33 successfully
[2019-Jan-02 16:29:24.139742](info):received libevent callback event
[2019-Jan-02 16:29:28.145475](info):connect timeout
[2019-Jan-02 16:29:28.149316](info):connect server with addr:118.190.213.26:80 success
[2019-Jan-02 16:29:28.232673](warning):get error response:1 of heartbeat to addr:118.190.213.26:80
[2019-Jan-02 16:29:28.233149](info):try to connect to fd:38, addr:118.190.213.24
[2019-Jan-02 16:29:28.233311](info):Wait till event base is looping
[2019-Jan-02 16:29:28.233332](info):Notify on event_base_dispatch
[2019-Jan-02 16:29:28.233484](info):eventcb: received event:80 on fd:38
[2019-Jan-02 16:29:28.233554](info):eventcb:connect to fd:38 successfully
[2019-Jan-02 16:29:28.233665](info):received libevent callback event
[2019-Jan-02 16:29:32.241188](info):connect timeout
[2019-Jan-02 16:29:32.241284](info):connect server with addr:118.190.213.24:80 success
[2019-Jan-02 16:29:32.270711](warning):get error response:1 of heartbeat to addr:118.190.213.24:80
[2019-Jan-02 16:29:32.270824](info):Client factory:9080-30.17.224.83@DEFAULT start dorebalance
[2019-Jan-02 16:29:32.270880](info):current topic is:%RETRY%CID-messense-test
[2019-Jan-02 16:29:32.270924](info):current topic is:messense-test
[2019-Jan-02 16:29:32.270974](info):getConsumerIdList from broker:118.190.213.25:80
[2019-Jan-02 16:29:32.300483](error):msg: com.aliyun.openservices.ons.api.impl.authority.exception.AuthenticationException: valid resource owner failed. maybe the resource CID-messense-test not created, com.alibaba.ons.open.auth.validator.PermissionValidator.check(PermissionValidator.java:57),error:1,in file </Users/messense/Projects/rocketmq-client-cpp/src/MQClientAPIImpl.cpp> line:766
[2019-Jan-02 16:29:32.300594](error):msg: doRebalance the cidAll is empty,error:-1,in file </Users/messense/Projects/rocketmq-client-cpp/src/consumer/Rebalance.cpp> line:106
[2019-Jan-02 16:29:32.300631](info):Client factory:9080-30.17.224.83@DEFAULT finish dorebalance
[2019-Jan-02 16:29:32.300716](info):clientFactory:9080-30.17.224.83@DEFAULT start consumer_timerOperation
[2019-Jan-02 16:29:41.733052](warning):get error response:1 of heartbeat to addr:118.190.213.25:80
[2019-Jan-02 16:29:41.767404](warning):get error response:1 of heartbeat to addr:118.190.213.23:80
[2019-Jan-02 16:29:41.794702](warning):get error response:1 of heartbeat to addr:118.190.213.26:80
[2019-Jan-02 16:29:41.823902](warning):get error response:1 of heartbeat to addr:118.190.213.24:80

可以直接使用rocketmq的静态库吗？

目前开源版本编译的rocketmq动态库，有日志文件太大和路径的问题，想直接使用同事修改过的静态库，ctypes.cdll.LoadLibrary这个方法可以加载c++的静态库吗？

Add orderly producer support

connect to broker too much time

2019-Apr-11 13:52:05.839388:unregisterClient to broker:10.4.44.76:10911
2019-Apr-11 13:52:05.839669:try to connect to fd:28, addr:10.4.44.76
2019-Apr-11 13:52:05.839762:Wait till event base is looping
2019-Apr-11 13:52:05.839773:Notify on event_base_dispatch
2019-Apr-11 13:52:05.853544:eventcb: received event:80 on fd:28
2019-Apr-11 13:52:05.853618:eventcb:connect to fd:28 successfully
2019-Apr-11 13:52:05.853643:received libevent callback event
2019-Apr-11 13:52:09.843417:connect timeout
2019-Apr-11 13:52:09.843516:connect server with addr:10.4.44.76:10911 success
2019-Apr-11 13:52:09.857889:unregisterClient to:10.4.44.76:10911 success

以上是日志信息，connect每个broker都耗时接近4s，看上去应该很早就connect成功了，但是依然wait了4s，日志里还出现'connect timeout'。具体表现为，在python中第一次调用send_sync时，耗时很长。

windows本地调用出现error 193

windows本地模拟的时候，在import rocketmq文件，调用dll，读取librocketmq.so的时候出现异常。windows和python都是64位的，是不是rocketmq的位数只有32位？换成linux 64位，python 64位的时候，是不是还是会有这个问题？

消费者进程异常

我连的是阿里云，可能是由于硬件或者什么原因，会出现一种现象，消费者进程还在，但会失去和nameserver的连接，消息收不到（查了控制台，也确实消息生产成功，但没有被消费），问下有什么解决办法吗

Apple Silicon support

librocketmq.dylib.zip

rocketmq-python does not support Windows

NotImplementedError: rocketmq-python does not support Windows

push模式可以设置多个NameServer吗

PushConsumer segfault when it gets deleted

Reproduce:

pip install rocketmq==0.1.0

from rocketmq.client import PushConsumer

c = PushConsumer('test')
exit()

apache/rocketmq-client-cpp#50 will fix this.

非阿里云，日志提示仍然需要认证

不是用的阿里云，而是公司内部的中间件。采用python3.7搭建消费者，启动服务任何错误和消息输出，但是就是接收不到客户端（java）的请求。查看logs/rocketmq-cpp日志，发现以下error：

(error):updateTopicRouteInfo: didn't get the session_credentials from any producers and consumers, please re-intialize it[getSessionCredentialsFromOneOfProducerOrConsumer:1128]

这里是因为无论是不是阿里云，都需要配置key和secret才能连接吗？
这是我的消费者端代码：

import time
from rocketmq.client import PushConsumer
from rocketmq.ffi import MessageModel
import logging
logging.basicConfig(level=logging.DEBUG)

def callback(msg):
    print(type(msg.body))
    print(msg.id, msg.body.decode('utf-8'))
consumer = PushConsumer('CID_CMS_PREDICTION_CALLBACK')
consumer.set_namesrv_addr('10.167.20.46:9876')
consumer.subscribe('TOPIC_CMS_KEYWORD_INNER', callback)
consumer.start()
logging.info("mq start...")
while True:
    time.sleep(15)
consumer.shutdown()

以下是部分日志：

[2019-Nov-13 20:39:16.364481](info):DefaultMQPushConsumer::boost asio async service runing[boost_asio_work:214]
[2019-Nov-13 20:39:16.367013](info):m_tcpConnectTimeout:3000, m_tcpTransportTryLockTimeout:3, m_pullThreadNum:8[TcpRemotingClient:61][2019-Nov-13 20:39:16.367125](info):MQClient start,groupname:CID_CMS_PREDICTION_CALLBACK,clientID:10228-10.167.21.179@DEFAULT,instanceName:DEFAULT,nameserveraddr:10.167.20.46:9876[start:141]
[2019-Nov-13 20:39:16.367152](info):DefaultMQPushConsumer:CID_CMS_PREDICTION_CALLBACK start[start:300]
[2019-Nov-13 20:39:16.367124](info):TcpRemotingClient::boost asio async service running[boost_asio_work:67]
[2019-Nov-13 20:39:16.367170](info):start concurrently consume service:CID_CMS_PREDICTION_CALLBACK[start:318]
[2019-Nov-13 20:39:16.368452](info):buildSubscriptionData,:TOPIC_CMS_KEYWORD_INNER,*[copySubscription:457]
[2019-Nov-13 20:39:16.368492](info):updateNameServerAddressList: [10.167.20.46:9876][updateNameServerAddressList:123]
[2019-Nov-13 20:39:16.368505](info):update Namesrv:10.167.20.46:9876[updateNameServerAddressList:148]
[2019-Nov-13 20:39:16.368519](info):user specfied name server address: 10.167.20.46:9876[registerConsumer:364]
[2019-Nov-13 20:39:16.368529](info):MQClientFactory:10228-10.167.21.179@DEFAULT start[start:68]
[2019-Nov-13 20:39:16.368872](info):updateTopicRouteInfoFromNameServer start:%RETRY%CID_CMS_PREDICTION_CALLBACK[updateTopicRouteInfoFromNameServer:159]
[2019-Nov-13 20:39:16.368893](info):start scheduled task:10228-10.167.21.179@DEFAULT[startScheduledTask:803]
[2019-Nov-13 20:39:16.368945](info):namesrvIndex is:1, index:0, namesrvaddrlist size:1[CreateNameServerTransport:379]
[2019-Nov-13 20:39:16.369085](info):try to connect to fd:16, addr:10.167.20.46[connect:172]
[2019-Nov-13 20:39:16.370117](info):eventcb: received event:80 on fd:16[eventCallback:191]
[2019-Nov-13 20:39:16.370141](info):eventcb: connect to fd:16 successfully[eventCallback:193]
[2019-Nov-13 20:39:16.370163](info):connect server with addr:10.167.20.46:9876 success[CreateTransport:348]
[2019-Nov-13 20:39:16.370581](warning):The topic:[%RETRY%CID_CMS_PREDICTION_CALLBACK] not exist[updateTopicSubscribeInfoWhenSubscriptionChanged:490]
[2019-Nov-13 20:39:16.370596](info):updateTopicRouteInfoFromNameServer start:TOPIC_CMS_KEYWORD_INNER[updateTopicRouteInfoFromNameServer:159]
[2019-Nov-13 20:39:16.370907](warning):The topic:[TOPIC_CMS_KEYWORD_INNER] not exist[updateTopicSubscribeInfoWhenSubscriptionChanged:490]
[2019-Nov-13 20:39:16.370920](warning):sendheartbeat brokeradd is empty[sendHeartbeatToAllBroker:703]
[2019-Nov-13 20:39:16.370929](info):Client factory:10228-10.167.21.179@DEFAULT start dorebalance[doRebalance:846]
[2019-Nov-13 20:39:16.370939](info):current topic is:%RETRY%CID_CMS_PREDICTION_CALLBACK[doRebalance:59]
[2019-Nov-13 20:39:16.370950](info):current topic is:TOPIC_CMS_KEYWORD_INNER[doRebalance:59]
[2019-Nov-13 20:39:16.370958](info):Client factory:10228-10.167.21.179@DEFAULT finish dorebalance[doRebalance:853]
[2019-Nov-13 20:39:16.370996](info):clientFactory:10228-10.167.21.179@DEFAULT start consumer_timerOperation[consumer_timerOperation:818]
[2019-Nov-13 20:39:16.379041](warning):sendheartbeat brokeradd is empty[sendHeartbeatToAllBroker:703]
[2019-Nov-13 20:39:19.369355](error):updateTopicRouteInfo: didn't get the session_credentials from any producers and consumers, please re-intialize it[getSessionCredentialsFromOneOfProducerOrConsumer:1128]

我是这方面的新手，感谢您的帮助！！

Producer的token属性有吗？

Producer的token属性有吗？我看了下代码。没有找到？

Python 2 support

Migrate CI to Azure Pipelines

How to configure nameServer with addr like :10.101.1.11:11033

Now I configure like :
producer.set_namesrv_domain('10.101.1.42:9887')
but in logfile ,see:
2019-May-21 15:36:36.679781:MQClient start,groupname:pythonClientProducer,clientID:55063-10.101.1.41@DEFAULT,instanceName:DEFAULT,nameserveraddr:
2019-May-21 15:36:36.679821:DefaultMQProducer:pythonClientProducer start
2019-May-21 15:36:36.679776:TcpRemotingClient::boost asio async service runing
2019-May-21 15:36:36.679924:fetchNSAddr protocol: 10.101.1.42:9887, port: , host:, path:,
2019-May-21 15:36:36.680032:Exception: resolve: Host not found (authoritative)
2019-May-21 15:36:36.680062:fetchNSAddr with domain failed, connect failure or wrong response
It seems didn't work out?

How to use it correctly in rocketmq-python-client to ensure at least one time or just once

BUG REPORT

Please describe the issue you observed:
What did you do (The steps to reproduce)?
Example of using the official website pullconsumer or pushconsumer
What did you expect to see?
Spend at least once or just once, and process the msg of the transaction failure, then msg roll back, wait for the next try again
What did you see instead?
read all msg repeatedly every time you start
Please tell us about your environment:
What is your OS?
ubuntu 18.04
What is your client version?
0.4.1
What is your RocketMQ version?
4.5.2
Other information (e.g. detailed explanation, logs, related issues, suggestions on how to fix, etc):
FEATURE REQUEST

Please describe the feature you are requesting.
I want to know the following two usages：
（1）How can I use it properly to ensure that I consume it at least once or just once,
（2）and how the message queue is rolled back when the transaction fails.
Looking forward to your reply，thanks.

部署消费端进程，会出现不定期重启问题。

重启的时间间隔为1天左右。有办法能定位原因吗？rocketmq-cpp的错误日志只有如下的：
(error):updateTopicRouteInfo: didn't get the session_credentials from any producers and consumers, please re-intialize it[getSessionCredentialsFromOneOfProducerOrConsumer:1100

updateTopicRouteInfo: didn't get the session_credentials from any producers and consumers, please re-intialize it

服务端显示MQClientException: CODE:207, DESC: consumer Timeout:
然后我收不到消息，服务端显示我消息已经接受了，我查看了consumer的日志中提示：
(error):updateTopicRouteInfo: didn't get the session_credentials from any producers and consumers, please re-intialize it[getSessionCredentialsFromOneOfProducerOrConsumer:1128]

除了这个ERROR之外，consumer日志中其他都正常，没有任何warn，连接也是success的。

关于rocketmq信息获取

您好，我现在想获取rockMQ的消费信息，能用您的这个插件么？有例子没？学习以下

PULL模式，设置消费的起始位置

使用PULL模式消费时，怎么设置消费的起始位置。
我没有找到设值的地方，但是rocketmq本身有提供FROM_FIRST、FROM_LAST、FROM_TIMESTAMP三种回溯模式。

消息增加systemProperties的设置

RocketMQ支持消息的延迟发送，属性是设置在systemProperties里的，希望可以增加相应的功能。

rocketmq.exceptions.ProducerSendSyncFailed

ERROR MESSAGE

Traceback (most recent call last):
  File "producer.py", line 14, in <module>
    ret = producer.send_sync(msg)
  File "/home/awok/Python/awok/lib/python3.7/site-packages/rocketmq/client.py", line 176, in send_sync
    ffi_check(dll.SendMessageSync(self._handle, msg, ctypes.pointer(cres)))
  File "/home/awok/Python/awok/lib/python3.7/site-packages/rocketmq/exceptions.py", line 19, in ffi_check
    raise exc_cls()
rocketmq.exceptions.ProducerSendSyncFailed

PYTHON CODE

from rocketmq.client import Producer, Message

producer = Producer('PID-1235')
producer.set_namesrv_domain(
    'http://ons-me-east-1-internal.aliyuncs.com:8080/rocketmq/nsaddr4client-internal')
producer.set_session_credentials(
    'XXX', 'XXXX', 'ALIYUN')
producer.start()

msg = Message('test')
msg.set_keys('testing')
msg.set_tags('casdfas')
msg.set_body('hello')
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
producer.shutdown()

@messense, please help me out in it

I am havving trouble with the use of rocketmq-python in linux,python3.7

I am having trouble with the use of rocketmq-python in linux,python3.7,Details are as follows：

Traceback (most recent call last):
  File "./work.py", line 31, in <module>
    ret = producer.send_sync(msg)
  File "/usr/local/python37/lib/python3.7/site-packages/rocketmq/client.py", line 176, in send_sync
    ffi_check(dll.SendMessageSync(self._handle, msg, ctypes.pointer(cres)))
  File "/usr/local/python37/lib/python3.7/site-packages/rocketmq/exceptions.py", line 19, in ffi_check
    raise exc_cls()
rocketmq.exceptions.ProducerSendSyncFailed

from rocketmq.client import Producer, Message
producer = Producer('group')
producer.set_namesrv_domain('192.168.1.1:9876')
# producer.set_session_credentials('XXX', 'XXXX', 'ALIYUN')
producer.start()
msg = Message('topic')
# msg.set_keys('XXX')
# msg.set_tags('XXX')
msg.set_body('XXXX')
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
producer.shutdown()

阿里云非ONSAddr接入方式

现在的阿里云RocketMQ新建示例已经不再使用ONSAddr接入方式，请问是否有办法支持新的接入方式

使用旧的实例中的ons接入方式是可以跑通的，使用新版的接入方式一直报No route info of this topic