IoT Platform:Go SDK 接続の例

開発環境の準備

この例では Go 1.12.7 を使用します。

SDK のダウンロード

次のコマンドを使用して Go AMQP SDK をインポートします。

import "pack.ag/amqp"

SDK の使用方法の詳細については、「package amqp」をご参照ください。

コード例

package main
import (
	"os"
    "context"
    "crypto/hmac"
    "crypto/sha1"
    "encoding/base64"
    "fmt"
    "pack.ag/amqp"
    "time"
)
// パラメーターの説明については、AMQP クライアント接続ガイドをご参照ください。
const consumerGroupId = "${YourConsumerGroupId}"
const clientId = "${YourClientId}"
// iotInstanceId: インスタンス ID。
const iotInstanceId = "${YourIotInstanceId}"
// エンドポイント。詳細については、AMQP クライアント接続ガイドをご参照ください。
const host = "${YourHost}"
func main() {
	// プロジェクトコードが漏洩すると、AccessKey が公開される可能性があります。これにより、アカウント内のすべてのリソースのセキュリティが危険にさらされます。
	// 次のコードは、環境変数を使用して AccessKey を取得する方法の例を示しています。この方法は参考用です。
	accessKey := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
	accessSecret := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    address := "amqps://" + host + ":5671"
    timestamp := time.Now().Nanosecond() / 1000000
    // ユーザー名の作成方法については、AMQP クライアント接続ガイドをご参照ください。
    userName := fmt.Sprintf("%s|authMode=aksign,signMethod=Hmacsha1,consumerGroupId=%s,authId=%s,iotInstanceId=%s,timestamp=%d|", 
        clientId, consumerGroupId, accessKey, iotInstanceId, timestamp)
    stringToSign := fmt.Sprintf("authId=%s&timestamp=%d", accessKey, timestamp)
    hmacKey := hmac.New(sha1.New, []byte(accessSecret))
    hmacKey.Write([]byte(stringToSign))
    // 署名を計算します。パスワードの作成方法については、AMQP クライアント接続ガイドをご参照ください。
    password := base64.StdEncoding.EncodeToString(hmacKey.Sum(nil))
    amqpManager := &AmqpManager{
        address:address,
        userName:userName,
        password:password,
    }
    // メッセージを受け入れるか、操作をキャンセルするには、context.Background() からコンテキストを派生させます。
    ctx := context.Background()
    amqpManager.startReceiveMessage(ctx)
}
// ビジネス関数。これはカスタム実装です。この関数は非同期で実行されます。システムのリソース消費を考慮してください。
func (am *AmqpManager) processMessage(message *amqp.Message) {
    fmt.Println("data received:", string(message.GetData()), " properties:", message.ApplicationProperties)
}
type AmqpManager struct {
    address     string
    userName     string
    password     string
    client       *amqp.Client
    session     *amqp.Session
    receiver     *amqp.Receiver
}
func (am *AmqpManager) startReceiveMessage(ctx context.Context)  {
    childCtx, _ := context.WithCancel(ctx)
    err := am.generateReceiverWithRetry(childCtx)
    if nil != err {
        return
    }
    defer func() {
        am.receiver.Close(childCtx)
        am.session.Close(childCtx)
        am.client.Close()
    }()
    for {
        // メッセージを受信するためにブロックします。コンテキストがバックグラウンドの場合、プロセスは中断されません。
        message, err := am.receiver.Receive(ctx)
        if nil == err {
            go am.processMessage(message)
            message.Accept()
        } else {
            fmt.Println("amqp receive data error:", err)
            // 操作がアクティブにキャンセルされた場合は、プログラムを終了します。
            select {
            case <- childCtx.Done(): return
            default:
            }
            // 操作がアクティブにキャンセルされていない場合は、接続を再確立します。
            err := am.generateReceiverWithRetry(childCtx)
            if nil != err {
                return
            }
        }
    }
}
func (am *AmqpManager) generateReceiverWithRetry(ctx context.Context) error {
    // バックオフして再接続します。10 ミリ秒から開始し、最大 20 秒まで間隔を 2 倍にします。
    duration := 10 * time.Millisecond
    maxDuration := 20000 * time.Millisecond
    times := 1
    // 例外が発生した場合は、バックオフして再接続します。
    for {
        select {
        case <- ctx.Done(): return amqp.ErrConnClosed
        default:
        }
        err := am.generateReceiver()
        if nil != err {
            time.Sleep(duration)
            if duration < maxDuration {
                duration *= 2
            }
            fmt.Println("amqp connect retry,times:", times, ",duration:", duration)
            times ++
        } else {
            fmt.Println("amqp connect init success")
            return nil
        }
    }
}
// パッケージが可視でないため、接続とセッションの状態を判断できません。接続を再起動して状態を取得します。
func (am *AmqpManager) generateReceiver() error {
    if am.session != nil {
        receiver, err := am.session.NewReceiver(
            amqp.LinkSourceAddress("/queue-name"),
            amqp.LinkCredit(20),
        )
        // ネットワークが切断されると、接続が閉じられ、セッションの作成に失敗します。
        // 接続が開いたままであれば、セッションは正常に作成されます。
        if err == nil {
            am.receiver = receiver
            return nil
        }
    }
    // 以前の接続をクリーンアップします。
    if am.client != nil {
        am.client.Close()
    }
    client, err := amqp.Dial(am.address, amqp.ConnSASLPlain(am.userName, am.password), )
    if err != nil {
        return err
    }
    am.client = client
    session, err := client.NewSession()
    if err != nil {
        return err
    }
    am.session = session
    receiver, err := am.session.NewReceiver(
        amqp.LinkSourceAddress("/queue-name"),
        amqp.LinkCredit(20),
    )
    if err != nil {
        return err
    }
    am.receiver = receiver
    return nil
}

前述のコードのパラメーターを次の表の説明に従って設定します。詳細については、「AMQP クライアントの IoT Platform への接続」をご参照ください。

実行結果の例

• 成功：次のようなログメッセージが返されます。これは、AMQP クライアントが IoT Platform に正常に接続し、デバイスからメッセージを受信したことを示します。

  amqp connect init success
  data received: {"deviceType":"CustomCategory","iotId":"xxx","requestId":"1613726251726","checkFailedData":0,"productKey":"xxx","gmtCreate":"1613726121717,"deviceName":"xxx","items":{"Temperature":{"value":24,"time":1613726121715},"Humidity":{"value":19,"time":1613726121715}}}  properties: map[generateTime:1613726121721 messageId:xxx xxx s:1 topic: /xxx/thing/event/property/post]
  data received: {"deviceType":"CustomCategory","iotId":"xxx","requestId":"1613725651726","checkFailedData":0,"productKey":"xxx","gmtCreate":"1613725521715,"deviceName":"xxx","items":{"Temperature":{"value":28,"time":1613725521712},"Humidity":{"value":19,"time":1613725521712}}}  properties: map[generateTime:1613725521719 messageId:1362689721104473600 qos:1 topic: /xxx/thing/event/property/post]

• 失敗：次のようなログメッセージが返され、AMQP クライアントが IoT Platform への接続に失敗したことを示します。

  エラーログを使用して、コードとネットワーク設定を確認してください。問題を解決してから、コードを再度実行します。

  amqp connect retry,times: 1 ,duration: 20ms
  amqp connect retry,times: 2 ,duration: 40ms
  amqp connect retry,times: 3 ,duration: 80ms
  amqp connect retry,times: 4 ,duration: 160ms
  amqp connect retry,times: 5 ,duration: 320ms
  amqp connect retry,times: 6 ,duration: 640ms
  amqp connect retry,times: 7 ,duration: 1.28s
  amqp connect retry,times: 8 ,duration: 2.56s
  amqp connect retry,times: 9 ,duration: 5.12s
  amqp connect retry,times: 10 ,duration: 10.24s
  amqp connect retry,times: 11 ,duration: 20.48s

リファレンス

サーバー側サブスクリプションメッセージのエラーコードの詳細については、「メッセージ関連のエラーコード」をご参照ください。