日本免码va在线看免费最,亚洲欧美人成网站在线观看看,婷婷六月中文字幕

本文為您提供當前主流編程語言的消息回執示例，您可通過下載相應的SDK安裝包拉取隊列消息。

所屬語言	SDK下載鏈接
Java	Java SDK
Node.js	Node.js SDK
.NET/C#	阿里云SDK核心庫 dybaseapi 說明依賴.NET語言的阿里云SDK核心庫及dybaseapi，其中dybaseapi包用于拉取隊列消息。
Python	阿里云SDK核心庫（Python） dybaseapi（Python）說明依賴Python語言的阿里云SDK核心庫及dybaseapi，其中dybaseapi包用于拉取隊列消息。
Go	Go SDK

消息回執示例

說明

調用接口前需配置環境變量，通過環境變量讀取訪問憑證。AccessKey ID和AccessKey Secret的環境變量名：ALIBABA_CLOUD_ACCESS_KEY_ID 、ALIBABA_CLOUD_ACCESS_KEY_SECRET。配置詳情請參見在Linux、macOS和Windows系統配置環境變量。

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)
        {
                
            // 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。
            // 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。
            // 本示例以把AccessKey ID和AccessKey Secret保存在環境變量為例說明，來實現API訪問的身份驗證。
            IClientProfile profile = DefaultProfile.GetProfile("cn-hangzhou", Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_ID"), Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_SECRET")); // todo: 補充AK信息
            profile.AddEndpoint("cn-hangzhou", "cn-hangzhou", "Dybaseapi", "dybaseapi.aliyuncs.com");

            DefaultAcsClient client = new DefaultAcsClient(profile);

            String queueName = "<QueueName>"; // todo: 補充隊列名稱
            String messageType = "<MessageType>"; // todo: 補充消息類型

            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));
                //啟動線程
                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; // 過期時間小于2分鐘則重新獲取，防止服務器時間誤差
        const String mnsAccountEndpoint = "https://1943695596114318.mns.cn-hangzhou.aliyuncs.com/"; // 阿里通信消息的endpoint,固定

        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;
        }

        /// 處理消息
        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 具體消費邏輯,待客戶自己實現.
                            // 消費成功的前提下刪除消息
                            // queue.DeleteMessage(messages[i].ReceiptHandle);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine(e.ToString());
                        }
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine(e.ToString());
                }
                Thread.Sleep(sleepTime);
            }
        }
    }
}

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 需要替換成您需要接收的消息類型
message_type = "<MessageType>"
# TODO 需要替換成您的隊列名稱。在云通信頁面開通相應業務消息后，就能在頁面上獲得對應的queueName
queue_name = "<QueueName>"

# 云通信固定的endpoint地址
endpoint = "https://1943695596114318.mns.cn-hangzhou.aliyuncs.com/"
# 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。
# 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。
# 本示例以把AccessKey ID和AccessKey Secret保存在環境變量為例說明，來實現API訪問的身份驗證。
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")


# 云通信業務token存在失效時間，需動態更新。
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
        # 失效時間與當前系統時間比較，提前2分鐘刷新token
        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", "獲取token時無響應")

        response_body = json.loads(response)

        if response_body.get("Code") != "OK":
            raise ServerException("GET_TOKEN_FAIL", "獲取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...")


# 初始化 token, my_account, my_queue
token, my_account, my_queue = Token(), None, None

# 循環讀取刪除消息直到隊列空
# receive message請求使用long polling方式，通過wait_seconds指定長輪詢時間為3秒

# long polling 解析：
## 當隊列中有消息時，請求立即返回；
## 當隊列中沒有消息時，請求在MNS服務器端掛3秒鐘，在這期間，有消息寫入隊列，請求會立即返回消息，3秒后，請求返回隊列沒有消息；

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 = []
    # 讀取消息
    try:
        # token過期是否需要刷新
        if token.is_refresh() == 1:
            # 刷新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)

        # 接收消息
        recv_msgs = my_queue.batch_receive_message(10, wait_seconds)

        for recv_msg in recv_msgs:
            # TODO 業務處理

            # 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

    # 刪除消息
    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)

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 (
    mnsDomain = "1943695596114318.mns.cn-hangzhou.aliyuncs.com"
)

func main() {
    endpoints.AddEndpointMapping("cn-hangzhou", "Dybaseapi", "dybaseapi.aliyuncs.com")
     // 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。
     // 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。
     // 本示例以把AccessKey ID和AccessKey Secret保存在環境變量為例說明，來實現API訪問的身份驗證。
     // 創建client實例
    client, err := dybaseapi.NewClientWithAccessKey(
        "cn-hangzhou",           // 您的可用區ID
        os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),      // 您環境變量中配置的AccessKey ID
        os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))     // 您環境變量中配置的AccessKey Secret
    if err != nil {
        // 異常處理
        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 {
            // 創建 API 請求并設置參數
            request := dybaseapi.CreateQueryTokenForMnsQueueRequest()
            request.MessageType = messageType
            request.QueueName = queueName
            // 發起請求并處理異常
            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) //取消注釋將刪除隊列中的消息
            if err != nil {
                panic(err)
            }
        }
    }
}

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

輕量消息隊列（原MNS）消費Demo

消息回執示例

C# Demo

Python Demo

Go Demo