{
  "require": {
     "aliyunmq/mq-http-sdk": ">=1.0.4"
  }
}

   private $client;
   private $consumer;

   public function __construct()
    {
        // Obtain the client.
        $this->client = new MQClient(
            endpoint,
            accessKey,
            secretKey
        );

        $topic = topicName;
        $groupId = groupName;
        $instanceId = namespace;

        // Obtain the consumer.
        $this->consumer = $this->client->getConsumer($instanceId, $topic, $groupId);
    }

public function run()
    {
        while (True) {
            try {
                // Consume messages through long round-robin scheduling.
                // Long round-robin scheduling means that if there are no messages in the topic, the request waits on the server side. If messages are available for consumption, they are returned immediately.
                // If your business is sensitive to consumption delays, it is highly recommended to pull messages concurrently.
                $messages = $this->consumer->consumeMessage(
                    batchSize,
                    waitSeconds
                );
            } catch (\\Exception $e) {
                if ($e instanceof MQ\\Exception\\MessageNotExistException) {
                    printf("No message, contine long polling!RequestId:%s\\n", $e->getRequestId());
                    continue;
                }

                print_r($e->getMessage() . "\\n");

                sleep(1);
                continue;
            }

            print "consume finish, messages:\\n";

            $receiptHandles = array();
            foreach ($messages as $message) {
                $receiptHandles[] = $message->getReceiptHandle();
                printf("MessageID:%s TAG:%s BODY:%s \\nPublishTime:%d, FirstConsumeTime:%d, \\nConsumedTimes:%d, NextConsumeTime:%d,MessageKey:%s\\n",
                    $message->getMessageId(), $message->getMessageTag(), $message->getMessageBody(),
                    $message->getPublishTime(), $message->getFirstConsumeTime(), $message->getConsumedTimes(), $message->getNextConsumeTime(),
                    $message->getMessageKey());
                print_r($message->getProperties());
                // Process business logic.
            }

            print_r($receiptHandles);
            try {
                $this->consumer->ackMessage($receiptHandles);
            } catch (\\Exception $e) {
                if ($e instanceof MQ\\Exception\\AckMessageException) {
                    printf("Ack Error, RequestId:%s\\n", $e->getRequestId());
                    foreach ($e->getAckMessageErrorItems() as $errorItem) {
                        printf("\\tReceiptHandle:%s, ErrorCode:%s, ErrorMsg:%s\\n", $errorItem->getReceiptHandle(), $errorItem->getErrorCode(), $errorItem->getErrorCode());
                    }
                }
            }
            print "ack finish\\n";
        }
    }