前提條件

您已完成以下操作：

• 安裝Java SDK。更多信息，請參見準備環境。

• 創建資源。代碼中涉及的資源信息，例如實例、Topic和Group ID等，需要在控制臺上提前創建。更多信息，請參見創建資源。

• 獲取阿里云訪問密鑰AccessKey ID和AccessKey Secret。更多信息，請參見創建AccessKey。

發送普通消息

發送普通消息的示例代碼如下。

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQProducer;
import com.aliyun.mq.http.model.TopicMessage;

import java.util.Date;

public class Producer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 設置HTTP協議客戶端接入點，進入消息隊列RocketMQ版控制臺實例詳情頁面的接入點區域查看。
                "${HTTP_ENDPOINT}",
                // 請確保環境變量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設置。
                // AccessKey ID，阿里云身份驗證標識。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	              // AccessKey Secret，阿里云身份驗證密鑰。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        // 消息所屬的Topic，在消息隊列RocketMQ版控制臺創建。
        // 不同消息類型的Topic不能混用，例如普通消息的Topic只能用于收發普通消息，不能用于收發其他類型的消息。
        final String topic = "${TOPIC}";
        // Topic所屬的實例ID，在消息隊列RocketMQ版控制臺創建。
        // 若實例有命名空間，則實例ID必須傳入；若實例無命名空間，則實例ID傳入null空值或字符串空值。實例的命名空間可以在消息隊列RocketMQ版控制臺的實例詳情頁面查看。
        final String instanceId = "${INSTANCE_ID}";

        // 獲取Topic的生產者。
        MQProducer producer;
        if (instanceId != null && instanceId != "") {
            producer = mqClient.getProducer(instanceId, topic);
        } else {
            producer = mqClient.getProducer(topic);
        }

        try {
            // 循環發送4條消息。
            for (int i = 0; i < 4; i++) {
                TopicMessage pubMsg;        // 普通消息。
                pubMsg = new TopicMessage(
                // 消息內容。
                "hello mq!".getBytes(),
                // 消息標簽。
                "A"
                );
                // 設置消息的自定義屬性。
                pubMsg.getProperties().put("a", String.valueOf(i));
                // 設置消息的Key。
                pubMsg.setMessageKey("MessageKey");
               
            // 同步發送消息，只要不拋異常就是成功。
            TopicMessage pubResultMsg = producer.publishMessage(pubMsg);

            // 同步發送消息，只要不拋異常就是成功。
            System.out.println(new Date() + " Send mq message success. Topic is:" + topic + ", msgId is: " + pubResultMsg.getMessageId()
                        + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5());
            }
        } catch (Throwable e) {
            // 消息發送失敗，需要進行重試處理，可重新發送這條消息或持久化這條數據進行補償處理。
            System.out.println(new Date() + " Send mq message failed. Topic is:" + topic);
            e.printStackTrace();
        }

        mqClient.close();
    }

}

訂閱普通消息

訂閱普通消息的示例代碼如下。

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;

import java.util.ArrayList;
import java.util.List;

public class Consumer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 設置HTTP協議客戶端接入點，進入消息隊列RocketMQ版控制臺實例詳情頁面的接入點區域查看。
                "${HTTP_ENDPOINT}",
                // 請確保環境變量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設置。
                // AccessKey ID，阿里云身份驗證標識。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	              // AccessKey Secret，阿里云身份驗證密鑰。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        // 消息所屬的Topic，在消息隊列RocketMQ版控制臺創建。
        //不同消息類型的Topic不能混用，例如普通消息的Topic只能用于收發普通消息，不能用于收發其他類型的消息。
        final String topic = "${TOPIC}";
        // 您在消息隊列RocketMQ版控制臺創建的Group ID。
        final String groupId = "${GROUP_ID}";
        // Topic所屬的實例ID，在消息隊列RocketMQ版控制臺創建。
        // 若實例有命名空間，則實例ID必須傳入；若實例無命名空間，則實例ID傳入null空值或字符串空值。實例的命名空間可以在消息隊列RocketMQ版控制臺的實例詳情頁面查看。
        final String instanceId = "${INSTANCE_ID}";

        final MQConsumer consumer;
        if (instanceId != null && instanceId != "") {
            consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
        } else {
            consumer = mqClient.getConsumer(topic, groupId);
        }

        // 在當前線程循環消費消息，建議多開個幾個線程并發消費消息。
        do {
            List<Message> messages = null;

            try {
                // 長輪詢消費消息。
                // 長輪詢表示如果Topic沒有消息,則請求會在服務端掛起3s，3s內如果有消息可以消費則立即返回客戶端。
                messages = consumer.consumeMessage(
                        3,// 一次最多消費3條消息（最多可設置為16條）。
                        3// 長輪詢時間3秒（最多可設置為30秒）。
                );
            } catch (Throwable e) {
                e.printStackTrace();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
            // Topic中沒有消息可消費。
            if (messages == null || messages.isEmpty()) {
                System.out.println(Thread.currentThread().getName() + ": no new message, continue!");
                continue;
            }

            // 處理業務邏輯。
            for (Message message : messages) {
                System.out.println("Receive message: " + message);
            }

            // 消息重試時間到達前若不確認消息消費成功，則消息會被重復消費。
            // 消息句柄有時間戳，同一條消息每次消費的時間戳都不一樣。
            {
                List<String> handles = new ArrayList<String>();
                for (Message message : messages) {
                    handles.add(message.getReceiptHandle());
                }

                try {
                    consumer.ackMessage(handles);
                } catch (Throwable e) {
                    // 某些消息的句柄可能超時，會導致消息消費狀態確認不成功。
                    if (e instanceof AckMessageException) {
                        AckMessageException errors = (AckMessageException) e;
                        System.out.println("Ack message fail, requestId is:" + errors.getRequestId() + ", fail handles:");
                        if (errors.getErrorMessages() != null) {
                            for (String errorHandle :errors.getErrorMessages().keySet()) {
                                System.out.println("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                        + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                            }
                        }
                        continue;
                    }
                    e.printStackTrace();
                }
            }
        } while (true);
    }
}