This topic provides examples of how to receive messages in popular programming languages. You can download the corresponding software development kit (SDK) package to pull messages from a queue.
Downloads
Language | SDK download link |
Java | |
Node.js | Download the original Node.js SDK package to pull messages. |
.NET/C# | Note This depends on the Alibaba Cloud SDK core library for .NET and the dybaseapi package. The dybaseapi package is used to pull messages from the queue. |
Python | This depends on the Alibaba Cloud SDK core library for Python and the dybaseapi package. The dybaseapi package is used to pull messages. |
Go | Download the Go SDK package to pull messages. |
PHP |
Notes
When you use the examples, note the following:
Configure your AccessKey ID and AccessKey secret.
To prevent security risks, do not hard-code your AccessKey pair in your code. Instead, retrieve your AccessKey pair from environment variables. For more information, see Configure environment variables on Linux, macOS, and Windows systems.
This topic uses the environment variable names
ALIBABA_CLOUD_ACCESS_KEY_IDandALIBABA_CLOUD_ACCESS_KEY_SECRETas examples. The following code shows how to retrieve an AccessKey pair from environment variables:String accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"); String accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");Replace
messageTypewith the desired message type, such as DypnsSmsVerifyReport for text message verification code status reports.String messageType="messageType";queueNameis the name of the message queue. For example, for text message verification codes, you can find the queue name in the .String queueName="queueName";In the example code, the
dealMessagemethod processes the message content. You can add your business logic for processing the message content to this method. Theargparameter represents the parameters of the message body, such as start_time, end_time, duration, and status_code. Use these parameters as needed.// Parse the message body based on the message format described in the document. String arg = (String) contentMap.get("arg"); // Add your business logic.
Java demo
package com.alicom.mns.sample;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import com.alicom.mns.tools.DefaultAlicomMessagePuller;
import com.alicom.mns.tools.MessageListener;
import com.aliyun.mns.model.Message;
import com.google.gson.Gson;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
* This can be used only to receive messages from Alibaba Cloud Communications, not from other services.
* MNS message receiving demo
*/
public class ReceiveDemo {
private static Logger logger = Logger.getLogger(ReceiveDemo.class.getName());
static class MyMessageListener implements MessageListener{
private Gson gson=new Gson();
@Override
public boolean dealMessage(Message message) {
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// Key values of the message
System.out.println("message receiver time from mns:" + format.format(new Date()));
System.out.println("message handle: " + message.getReceiptHandle());
System.out.println("message body: " + message.getMessageBodyAsString());
System.out.println("message id: " + message.getMessageId());
System.out.println("message dequeue count:" + message.getDequeueCount());
System.out.println("Thread:" + Thread.currentThread().getName());
try{
Map<String,Object> contentMap=gson.fromJson(message.getMessageBodyAsString(), HashMap.class);
//TODO: Parse the message body based on the message format described in the document.
String arg = (String) contentMap.get("arg");
//TODO: Add your business logic here.
}catch(com.google.gson.JsonSyntaxException e){
logger.log(Level.SEVERE, "error_json_format:"+message.getMessageBodyAsString(),e);
// In theory, format errors do not occur. If a message has a format error, delete it. Otherwise, errors will persist upon redelivery.
return true;
} catch (Throwable e) {
// If an exception is caused by your code, return false. This prevents the message from being deleted and allows it to be redelivered based on the policy.
return false;
}
// If the message is processed, return true. The SDK then calls the MNS delete method to remove the message from the queue.
return true;
}
}
public static void main(String[] args) throws Exception, ParseException {
DefaultAlicomMessagePuller puller=new DefaultAlicomMessagePuller();
// Set the asynchronous thread pool size, task queue size, and hibernation time for threads with no data.
puller.setConsumeMinThreadSize(6);
puller.setConsumeMaxThreadSize(16);
puller.setThreadQueueSize(200);
puller.setPullMsgThreadSize(1);
// Enable this only when you debug issues with the server. Do not enable it for normal use because it affects performance.
puller.openDebugLog(false);
//TODO: Replace the following with your AccessKey information.
String accessKeyId="ALIBABA_CLOUD_ACCESS_KEY_ID";
String accessKeySecret="ALIBABA_CLOUD_ACCESS_KEY_SECRET";
/*
* TODO: Replace messageType and queueName with the required message type name and the corresponding queue name.
* All receipt message types for Artificial Intelligence Cloud Call Service:
* Communication AI Engine
* 1. Call record message: LlmSmartCallReport
* 2. Recording record message: LlmSmartCallRecord
* Communication Agent
* 1. Call record message: SecretStartReport
* 2. Recording record message: SecretRingReport
* 3. Summary receipt message: SecretPickUpReport
* Intelligent Contact Bot
* 1. Call record message: VoiceReportAiccs
* 2. Intermediate call status message: VoiceCallReportAiccs
* 3. Recording record message: VoiceRecordReportAiccs
* 4. Bot call record message: RobotCallReportAiccs
*/
String messageType="messageType"; // Replace this with the message type of your product.
String queueName="queueName"; // After you enable the message service for the corresponding business on the Alibaba Cloud Communications page, you can obtain the queueName on the page. The format is similar to Alicom-Queue-xxxxxx-SmsReport.
puller.startReceiveMsg(accessKeyId,accessKeySecret, messageType, queueName, new MyMessageListener());
}
}
Python demo
#!/usr/bin/env python
# coding=utf8
import os
import time
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.client import AcsClient
from aliyunsdkdybaseapi.request.v20170525.QueryTokenForMnsQueueRequest import QueryTokenForMnsQueueRequest
from aliyunsdkcore.profile import region_provider
from datetime import datetime
from aliyunsdkdybaseapi.mns.account import Account
from aliyunsdkdybaseapi.mns.queue import *
from aliyunsdkdybaseapi.mns.mns_exception import *
try:
import json
except ImportError:
import simplejson as json
# TODO: Replace this with the message type you want to receive.
message_type = "<MessageType>"
# TODO: Replace this with your queue name. After you enable the message service for the corresponding business on the Alibaba Cloud Communications page, you can obtain the queueName on the page.
queue_name = "<QueueName>"
# This is the fixed endpoint of Alibaba Cloud Communications. Do not modify it.
endpoint = "https://1943695596114318.mns.cn-hangzhou.aliyuncs.com/"
# An Alibaba Cloud account AccessKey pair has full permissions on all APIs. We recommend that you use a Resource Access Management (RAM) user to make API calls or perform routine O&M.
# We strongly recommend that you do not hard-code your AccessKey ID and AccessKey secret in your project code. Otherwise, the AccessKey pair may be leaked and the security of all resources in your account may be compromised.
# This example shows how to use environment variables to store your AccessKey ID and AccessKey secret for identity verification when you make API calls.
acs_client = AcsClient(os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "cn-hangzhou")
region_provider.add_endpoint("Dybaseapi", "dybaseapi.aliyuncs.com", "cn-hangzhou")
# The token for Alibaba Cloud Communications services expires. You must dynamically update it.
class Token():
def __init__(self):
self.token = None
self.tmp_access_id = None
self.tmp_access_key = None
self.expire_time = None
def is_refresh(self):
if self.expire_time is None:
return 1
# Compare the expiration time with the current system time. Refresh the token 2 minutes in advance.
now = datetime.now()
expire = datetime.strptime(self.expire_time, "%Y-%m-%d %H:%M:%S")
if expire <= now or (expire - now).seconds < 120:
return 1
return 0
def refresh(self):
print("start refresh token...")
request = QueryTokenForMnsQueueRequest()
request.set_MessageType(message_type)
request.set_QueueName(queue_name)
response = acs_client.do_action_with_exception(request)
# print response
if response is None:
raise ServerException("GET_TOKEN_FAIL", "No response when getting the token")
response_body = json.loads(response)
if response_body.get("Code") != "OK":
raise ServerException("GET_TOKEN_FAIL", "Failed to get the token")
sts_token = response_body.get("MessageTokenDTO")
self.tmp_access_key = sts_token.get("AccessKeySecret")
self.tmp_access_id = sts_token.get("AccessKeyId")
self.expire_time = sts_token.get("ExpireTime")
self.token = sts_token.get("SecurityToken")
print("finish refresh token...")
# Initialize token, my_account, and my_queue
token, my_account, my_queue = Token(), None, None
# Loop to read and delete messages until the queue is empty.
# The receive message request uses long polling. The wait_seconds parameter specifies a long polling period of 3 seconds.
# Long polling description:
## If the queue contains messages, the request returns immediately.
## If the queue is empty, the request is held on the MNS server for 3 seconds. If a message is written to the queue during this period, the request returns the message immediately. After 3 seconds, the request returns a response indicating that the queue is empty.
wait_seconds = 3
print("%sReceive And Delete Message From Queue%s\nQueueName:%s\nWaitSeconds:%s\n" % (
10 * "=", 10 * "=", queue_name, wait_seconds))
while True:
receipt_handles = []
# Read messages
try:
# Check if the token has expired and needs to be refreshed.
if token.is_refresh() == 1:
# Refresh the token.
token.refresh()
if my_account:
my_account.mns_client.close_connection()
my_account = None
if not my_account:
my_account = Account(endpoint, token.tmp_access_id, token.tmp_access_key, token.token)
my_queue = my_account.get_queue(queue_name)
# Receive messages.
recv_msgs = my_queue.batch_receive_message(10, wait_seconds)
for recv_msg in recv_msgs:
# TODO: Add business logic.
# receipt_handles.append(recv_msg.receipt_handle)
print("Receive Message Succeed! ReceiptHandle:%s MessageBody:%s MessageID:%s" % (
recv_msg.receipt_handle, recv_msg.message_body, recv_msg.message_id))
except MNSExceptionBase as e:
if e.type == "QueueNotExist":
print("Queue not exist, please create queue before receive message.")
break
elif e.type == "MessageNotExist":
print("Queue is empty! sleep 10s")
time.sleep(10)
continue
print("Receive Message Fail! Exception:%s\n" % e)
break
# Delete messages.
try:
if len(receipt_handles) > 0:
# my_queue.batch_delete_message(receipt_handles)
print("Delete Message Succeed! ReceiptHandles:%s" % receipt_handles)
except MNSExceptionBase as e:
print("Delete Message Fail! Exception:%s\n" % e)
C# demo
using Aliyun.Acs.Core;
using Aliyun.Acs.Core.Profile;
using Aliyun.Acs.Core.Exceptions;
using Aliyun.Acs.Dybaseapi.Model.V20170525;
using Aliyun.Acs.Dybaseapi.MNS;
using Aliyun.Acs.Dybaseapi.MNS.Model;
using System.Threading;
using System.Collections.Generic;
using System.Text;
using System;
using QueryTokenForMnsQueue_MessageTokenDTO = Aliyun.Acs.Dybaseapi.Model.V20170525.QueryTokenForMnsQueueResponse.QueryTokenForMnsQueue_MessageTokenDTO;
namespace CommonRpc
{
class Program
{
static void Main(string[] args)
{
// An Alibaba Cloud account AccessKey pair has full permissions on all APIs. We recommend that you use a RAM user to make API calls or perform routine O&M.
// We strongly recommend that you do not hard-code your AccessKey ID and AccessKey secret in your project code. Otherwise, the AccessKey pair may be leaked and the security of all resources in your account may be compromised.
// This example shows how to use environment variables to store your AccessKey ID and AccessKey secret for identity verification when you make API calls.
IClientProfile profile = DefaultProfile.GetProfile("cn-hangzhou", Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_ID"), Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_SECRET")); // todo: Add your AccessKey information.
profile.AddEndpoint("cn-hangzhou", "cn-hangzhou", "Dybaseapi", "dybaseapi.aliyuncs.com");
DefaultAcsClient client = new DefaultAcsClient(profile);
String queueName = "<QueueName>"; // todo: Add your queue name.
String messageType = "<MessageType>"; // todo: Add your message type.
int maxThread = 2;
for (int i = 0; i < maxThread; i++)
{
TestTask testTask = new TestTask("PullMessageTask-thread-" + i, messageType, queueName, client);
Thread t = new Thread(new ThreadStart(testTask.Handle));
// Start the thread.
t.Start();
}
Console.ReadKey();
try
{
QueryTokenForMnsQueueRequest request = new QueryTokenForMnsQueueRequest
{
MessageType = messageType,
QueueName = queueName
};
QueryTokenForMnsQueueResponse response = client.GetAcsResponse(request);
Console.WriteLine(response.MessageTokenDTO.SecurityToken);
}
catch (ServerException ex)
{
Console.WriteLine(ex.ToString());
}
catch (ClientException ex)
{
Console.WriteLine(ex.ToString());
}
}
}
class TestTask
{
object o = new object();
const int sleepTime = 50;
const long bufferTime = 60 * 2; // If the remaining validity period is less than 2 minutes, get the token again to prevent server time errors.
// This is the fixed endpoint of Alibaba Cloud Communications. Do not modify it.
const String mnsAccountEndpoint = "https://1943695596114318.mns.cn-hangzhou.aliyuncs.com/";
public String name { get; private set; }
public String messageType { get; private set; }
public String QueueName { get; private set; }
public int TaskID { get; private set; }
public IAcsClient AcsClient { get; private set; }
public TestTask(String name, String messageType, String queueName, IAcsClient acsClient)
{
this.name = name;
this.messageType = messageType;
this.QueueName = queueName;
this.AcsClient = acsClient;
}
readonly Dictionary<string, QueryTokenForMnsQueue_MessageTokenDTO> tokenMap = new Dictionary<string, QueryTokenForMnsQueue_MessageTokenDTO>();
readonly Dictionary<string, Queue> queueMap = new Dictionary<string, Queue>();
public QueryTokenForMnsQueue_MessageTokenDTO GetTokenByMessageType(IAcsClient acsClient, String messageType)
{
QueryTokenForMnsQueueRequest request = new QueryTokenForMnsQueueRequest
{
MessageType = messageType
};
QueryTokenForMnsQueueResponse queryTokenForMnsQueueResponse = acsClient.GetAcsResponse(request);
QueryTokenForMnsQueue_MessageTokenDTO token = queryTokenForMnsQueueResponse.MessageTokenDTO;
return token;
}
/// Process messages.
public void Handle()
{
while (true)
{
try
{
QueryTokenForMnsQueue_MessageTokenDTO token = null;
Queue queue = null;
lock (o)
{
if (tokenMap.ContainsKey(messageType))
{
token = tokenMap[messageType];
}
if (queueMap.ContainsKey(QueueName))
{
queue = queueMap[QueueName];
}
TimeSpan ts = new TimeSpan(0);
if (token != null)
{
DateTime b = Convert.ToDateTime(token.ExpireTime);
DateTime c = Convert.ToDateTime(DateTime.Now);
ts = b - c;
}
if (token == null || ts.TotalSeconds < bufferTime || queue == null)
{
token = GetTokenByMessageType(AcsClient, messageType);
IMNS client = new MNSClient(token.AccessKeyId, token.AccessKeySecret, mnsAccountEndpoint, token.SecurityToken);
queue = client.GetNativeQueue(QueueName);
if (tokenMap.ContainsKey(messageType))
{
tokenMap.Remove(messageType);
}
if (queueMap.ContainsKey(QueueName))
{
queueMap.Remove(QueueName);
}
tokenMap.Add(messageType, token);
queueMap.Add(QueueName, queue);
}
}
BatchReceiveMessageResponse batchReceiveMessageResponse = queue.BatchReceiveMessage(16);
List<Message> messages = batchReceiveMessageResponse.Messages;
for (int i = 0; i <= messages.Count - 1; i++)
{
try
{
byte[] outputb = Convert.FromBase64String(messages[i].Body);
string orgStr = Encoding.UTF8.GetString(outputb);
Console.WriteLine(orgStr);
// TODO: Add your specific consumer logic.
// Delete the message after it is successfully consumed.
// queue.DeleteMessage(messages[i].ReceiptHandle);
}
catch (Exception e)
{
Console.WriteLine(e.ToString());
}
}
}
catch (Exception e)
{
Console.WriteLine(e.ToString());
}
Thread.Sleep(sleepTime);
}
}
}
}PHP demo
<?php
use AlibabaCloud\Client\AlibabaCloud;
use AlibabaCloud\Client\Exception\ClientException;
use AlibabaCloud\Client\Exception\ServerException;
use AlibabaCloud\Dybaseapi\MNS\Requests\BatchReceiveMessage;
use AlibabaCloud\Dybaseapi\MNS\Requests\BatchDeleteMessage;
// An Alibaba Cloud account AccessKey pair has full permissions on all APIs. We recommend that you use a RAM user to make API calls or perform routine O&M.
// We strongly recommend that you do not hard-code your AccessKey ID and AccessKey secret in your project code. Otherwise, the AccessKey pair may be leaked and the security of all resources in your account may be compromised.
// This example shows how to use environment variables to store your AccessKey ID and AccessKey secret for identity verification when you make API calls.
AlibabaCloud::accessKeyClient(
getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
)
->regionId('cn-hangzhou')
->asGlobalClient();
$queueName = '<QueueName>'; // The queue name. Replace it with your queue name.
$messageType = '<MessageType>'; // The type of message to receive. Replace it with the message type you need, such as SmsReport.
$response = null;
$token = null;
$i = 0;
do {
try {
if (null == $token || strtotime($token['ExpireTime']) - time() > 2 * 60) {
$response = AlibabaCloud::rpcRequest()
->product('Dybaseapi')
->version('2017-05-25')
->action('QueryTokenForMnsQueue')
->method('POST')
->host("dybaseapi.aliyuncs.com")
->options([
'query' => [
'MessageType' => $messageType,
'QueueName' => $queueName,
],
])
->request()
->toArray();
}
$token = $response['MessageTokenDTO'];
$mnsClient = new \AlibabaCloud\Dybaseapi\MNS\MnsClient(
"http://1943695596114318.mns.cn-hangzhou.aliyuncs.com", // This is the fixed endpoint of Alibaba Cloud Communications. Do not modify it.
$token['AccessKeyId'],
$token['AccessKeySecret'],
$token['SecurityToken']
);
$mnsRequest = new BatchReceiveMessage(10, 5);
$mnsRequest->setQueueName($queueName);
$mnsResponse = $mnsClient->sendRequest($mnsRequest);
$receiptHandles = Array();
foreach ($mnsResponse->Message as $message) {
// User logic:
// $receiptHandles[] = $message->ReceiptHandle; // Records added to the $receiptHandles array will be deleted.
$messageBody = base64_decode($message->MessageBody); // The JSON string after Base64 decoding.
print_r($messageBody . "\n");
}
if (count($receiptHandles) > 0) {
$deleteRequest = new BatchDeleteMessage($queueName, $receiptHandles);
$mnsClient->sendRequest($deleteRequest);
}
} catch (ClientException $e) {
echo $e->getErrorMessage() . PHP_EOL;
} catch (ServerException $e) {
if ($e->getCode() == 404) {
$i++;
}
echo $e->getErrorMessage() . PHP_EOL;
}
} while ($i < 3);Go demo
package main
import (
"os"
"encoding/base64"
"fmt"
"github.com/aliyun/alibaba-cloud-sdk-go/sdk/endpoints"
"github.com/aliyun/alibaba-cloud-sdk-go/services/dybaseapi"
"github.com/aliyun/alibaba-cloud-sdk-go/services/dybaseapi/mns"
"time"
)
const (
// This is the fixed endpoint of Alibaba Cloud Communications. Do not modify it.
mnsDomain = "1943695596114318.mns.cn-hangzhou.aliyuncs.com"
)
func main() {
endpoints.AddEndpointMapping("cn-hangzhou", "Dybaseapi", "dybaseapi.aliyuncs.com")
// An Alibaba Cloud account AccessKey pair has full permissions on all APIs. We recommend that you use a RAM user to make API calls or perform routine O&M.
// We strongly recommend that you do not hard-code your AccessKey ID and AccessKey secret in your project code. Otherwise, the AccessKey pair may be leaked and the security of all resources in your account may be compromised.
// This example shows how to use environment variables to store your AccessKey ID and AccessKey secret for identity verification when you make API calls.
// Create a client instance.
client, err := dybaseapi.NewClientWithAccessKey(
"cn-hangzhou", // Your zone ID.
os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), // The AccessKey ID configured in your environment variables.
os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")) // The AccessKey secret configured in your environment variables.
if err != nil {
// Handle exceptions.
panic(err)
}
queueName := "<QueueName>"
messageType := "<MessageType>"
cstLoc, _ := time.LoadLocation("Asia/Shanghai")
var token *dybaseapi.MessageTokenDTO
var expireTime time.Time
for {
if token == nil || expireTime.Unix()-time.Now().Unix() < 2*60 {
// Create an API request and set its parameters.
request := dybaseapi.CreateQueryTokenForMnsQueueRequest()
request.MessageType = messageType
request.QueueName = queueName
// Send the request and handle exceptions.
response, err := client.QueryTokenForMnsQueue(request)
if err != nil {
panic(err)
}
token = &response.MessageTokenDTO
}
expireTime, err = time.ParseInLocation("2006-01-02 15:04:05", token.ExpireTime, cstLoc)
if err != nil {
panic(err)
}
mnsClient, err := mns.NewClientWithStsToken(
"cn-hangzhou",
token.AccessKeyId,
token.AccessKeySecret,
token.SecurityToken,
)
if err != nil {
panic(err)
}
mnsRequest := mns.CreateBatchReceiveMessageRequest()
mnsRequest.Domain = mnsDomain
mnsRequest.QueueName = queueName
mnsRequest.NumOfMessages = "10"
mnsRequest.WaitSeconds = "5"
mnsResponse, err := mnsClient.BatchReceiveMessage(mnsRequest)
if err != nil {
panic(err)
}
// fmt.Println(mnsResponse)
receiptHandles := make([]string, len(mnsResponse.Message))
for i, message := range mnsResponse.Message {
messageBody, decodeErr := base64.StdEncoding.DecodeString(message.MessageBody)
if decodeErr != nil {
panic(decodeErr)
}
fmt.Println(string(messageBody))
receiptHandles[i] = message.ReceiptHandle
}
if len(receiptHandles) > 0 {
mnsDeleteRequest := mns.CreateBatchDeleteMessageRequest()
mnsDeleteRequest.Domain = mnsDomain
mnsDeleteRequest.QueueName = queueName
mnsDeleteRequest.SetReceiptHandles(receiptHandles)
//_, err = mnsClient.BatchDeleteMessage(mnsDeleteRequest) // Uncomment this line to delete messages from the queue.
if err != nil {
panic(err)
}
}
}
}该文章对您有帮助吗？