全部产品
云市场

使用开源 Python SDK 访问阿里云 MQ

更新时间:2019-09-13 22:00:01

使用开源 Python SDK 连接阿里云 MQ,只需修改部分配置信息即可,其余配置信息保持不变。

前提条件

  • 请确保平台版本满足以下条件:

    • Linux:CentOS 6.x、CentOS 7.x、ReHL 6.x、ReHL 7.x
    • Darwin:macOS Mojave 10.14.x
  • 已下载开源 Python SDK,以及 rocketmq-client-cpp 1.2.2 或以上版本的 CPP SDK。

    由于开源 SDK 从 rocketmq-client-cpp 1.2.2 版本开始支持连接阿里云 MQ,因此需要将 Python SDK 的 CPP 动态库依赖升级至 1.2.2 及以上版本来实现无缝上云。

  • 已在阿里云 MQ 控制台创建资源,包括 Topic、Group ID(GID)、接入点(Endpoint)以及 AccessKeyId 和 AccessKeySecret。针对新建实例(有独立命名空间),还包括 Instance ID。

    命名空间详情请参见产品更新日志

设置参数

访问阿里云 MQ 需要使用阿里云的鉴权方式和接入点。下文针对默认实例(无独立命名空间)和新建实例(有独立命名空间)分别列出需要修改的配置信息,其余的配置和开源 SDK 的使用保持一致。

请设置以下关键配置项:

  1. 设置 SessionCredentials 为阿里云的 AccessKey。

    Python SDK 提供了访问阿里云默认的鉴权方式,示例代码如下:

    1. SetProducerSessionCredentials(producerid,accessKey,secretKey,channel);
    2. SetConsumerSessionCredentials(consumerid,accessKey,secretKey,channel);
    3. //此处前两个参数分别为从控制台上获取 AccessKeyId 和 AccessKeySecret,如果访问阿里云,设置 channel 为 "ALIYUN"
  2. 设置NamesrvAddr为阿里云 MQ 控制台上获取的 TCP 接入点。

    阿里云 MQ 将访问的 Name Server 地址以接入点的方式提供,示例代码如下:

    1. SetProducerNameServerAddressr(producerid,endpoint);
    2. SetConsumerNameServerAddressr(consumerid,endpoint);
  3. 设置 GID(Producer ID/Consumer ID)。

    由于阿里云 MQ 将原来的 Producer ID(PID)和 Consumer ID(CID)合并,统一称作 GID,开源 SDK 不再额外提供设置 GID 的接口,将原来的 PID 和 CID 设置为新创建的 GID 即可。

    针对新建实例,这里需要将 GID 重新拼接,格式为 InstanceID%GID。请参见下文的示例代码

  4. 设置 Topic。

    对于新用户,需要在阿里云 MQ 的控制台创建 Topic,对于需要迁移的用户,在控制台上创建相同的 Topic 即可,这里不用修改,如果是新建实例(有独立命名空间),这里需要把实例 ID 拼接上,格式为 InstanceID%Topic

示例代码

下文中只涉及部分核心代码的示例,完整的代码请参见 Sample

默认实例(无独立命名空间)

  • 生产者示例代码

    1. //控制台上申请的 GID 或者 PID
    2. producer = CreateProducer("GID_XXXX");
    3. //控制台上获取的接入点信息
    4. SetProducerNameServerAddress(producer,"http://XXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
    5. //用户的 AccessKeyId 和 AccessKeySecret,控制台上可以获取,第三个参数固定默认使用 “ALIYUN”
    6. SetProducerSessionCredentials(producer,"mq acesskey", "mq secretkey", "ALIYUN");
  • 消费者示例代码

    1. //控制台上申请的 GID 或者 CID
    2. consumer = CreatePushConsumer("GID_XXXX");
    3. //控制台上获取的接入点信息
    4. SetPushConsumerNameServerAddress(consumer,"http://XXXX.mq-internet-access.mq-internet.aliyuncs.com:80")
    5. //用户的 AccessKeyId 和 AccessKeySecret,控制台上可以获取,第三个参数固定默认使用 “ALIYUN”
    6. SetProducerSessionCredentials(consumer,"mq acesskey", "mq secretkey", "ALIYUN");
    7. //设置订阅的 Topic 和 TAGs
    8. Subscribe(consumer,"TOPIC","TAGS");
    9. ........
    10. //注册处理消息的回调函数
    11. RegisterMessageCallback(consumer, consumerMessage, None);

新建实例(有独立命名空间)

由于新建实例给用户提供了自己的命名空间,用于逻辑资源隔离,所以使用开源 SDK 访问时,需要显式地设置实例 ID,即 InstanceId,从控制台获取,格式为 MQ_INST_XXXXX_XXXX

  • 生产者示例代码

    1. //控制台上申请的 GID 或者 PID,需要和 InstanceId 拼装
    2. producer = CreateProducer("MQ_INST_XXXX_XXXX%GID_XXXX");
    3. //控制台上获取的接入点信息,接入点包含了新建实例的命名空间信息
    4. SetProducerNameServerAddress(producer,"http://MQ_INST_XXXX_XXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
    5. //用户的 AccessKeyId 和 AccessKeySecret,控制台上可以获取,第三个参数固定默认使用 “ALIYUN”
    6. SetProducerSessionCredentials(producer,"mq acesskey", "mq secretkey", "ALIYUN");
    7. ...........
    8. //发送消息的时候,Topic 也需要指定 InstanceId
    9. msg = CreateMessage("MQ_INST_XXXX_XXXX%YOURTOPIC");
  • 消费者示例代码

    1. //控制台上申请的 GID 或者 CID,需要和 InstanceId 拼装
    2. consumer = CreatePushConsumer("MQ_INST_XXXX_XXXX%GID_XXXX");
    3. //控制台上获取的接入点信息
    4. SetPushConsumerNameServerAddress(consumer,"http://MQ_INST_XXXX_XXXX.mq-internet-access.mq-internet.aliyuncs.com:80")
    5. //用户的AK和SK,控制台上可以获取,第三个参数固定默认使用 “ALIYUN”
    6. SetProducerSessionCredentials(consumer,"mq acesskey", "mq secretkey", "ALIYUN");
    7. //设置订阅的 Topic 和 TAGs,由于新建实例的 Topic 需要拼装
    8. Subscribe(consumer,"MQ_INST_XXXX_XXXX%TOPIC","TAGS");
    9. ........
    10. //注册处理消息的回调函数
    11. RegisterMessageCallback(consumer, consumerMessage, None);

更多信息

Python SDK 的使用方法详情请参见 Introduction