阿里云Rocket MQ Java Http SDK发送消费消息示例Demo

简介: 消息队列 RocketMQ 版支持 RESTful 风格的 HTTP 协议通信,并提供了以下 7 种语言的 SDK,下面以最新的Java Http SDK为范例介绍RocketMQ消息的发送和接收。

Step By Step

1、创建实例,登陆阿里云控制台
图片.png
图片.png

2、实例下面分别创建Topic和Http
Group
图片.png
图片.png

3、pom.xml

        <dependency>
            <groupId>com.aliyun.mq</groupId>
            <artifactId>mq-http-sdk</artifactId>
            <version>1.0.2</version>
        </dependency>

4、Producer Code Sample

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 ProducerDemo {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 设置HTTP接入域名(此处以公共云生产环境为例)
                "http://18482178********.mqrest.cn-shanghai.aliyuncs.com",
                // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
                "LTAIOZZg********",
                // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
                "v7CjUJCMk7j9aK****************"
        );

        // 所属的 Topic
        final String topic = "****";
        // Topic所属实例ID,默认实例为空
        final String instanceId = "MQ_INST_********";

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

        try {
            // 循环发送40条消息
            for (int i = 0; i < 40; i++) {
                TopicMessage pubMsg;
                if (i % 2 == 0) {
                    // 普通消息
                    pubMsg = new TopicMessage(
                            // 消息内容
                            "hello common mq!".getBytes(),
                            // 消息标签
                            "A"
                    );
                    // 设置属性
                    pubMsg.getProperties().put("a", String.valueOf(i));
                    // 设置KEY
                    pubMsg.setMessageKey("MessageKey");
                } else {
                    pubMsg = new TopicMessage(
                            // 消息内容
                            "hello delay mq!".getBytes(),
                            // 消息标签
                            "B"
                    );
                    // 设置属性
                    pubMsg.getProperties().put("b", String.valueOf(i));
                    // 定时消息, 定时时间为10s后
                    pubMsg.setStartDeliverTime(System.currentTimeMillis() + 10 * 1000);
                }
                // 同步发送消息,只要不抛异常就是成功
                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();
    }
}

5、Consumer Code Sample

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.model.Message;
import java.util.*;

public class ConsumerDemo {

    public static void main(String[] args) {

        MQClient mqClient = new MQClient(
                // 设置HTTP接入域名(此处以公共云生产环境为例)
                "http://18482178********.mqrest.cn-shanghai.aliyuncs.com",
                // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
                "LTAIOZZg********",
                // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
                "v7CjUJCMk7j9aK****************"
        );

        String topicName = "****";
        String consumer = "GID_****"; //Http Consumer Group Name
        String messageTag =""; // Tag,为空表示订阅全部Tag
        String instanceId = "MQ_INST_******";

        MQConsumer mqConsumer = mqClient.getConsumer(instanceId,topicName, consumer,messageTag);

        while(true) {
            try {

                // 消费消息,轮训时间设置为3秒,一次至多拉去三条消息
                List<Message> listMessage = mqConsumer.consumeMessage(3, 3);

                if (listMessage == null || listMessage.size() == 0) {
                    System.out.println("Message is not exist!");
                } else {
                    List<String> receiptHandles = new ArrayList<String>();
                    for (Message message : listMessage
                    ) {
                        System.out.println("MessageBody" + message.getMessageBodyString());
                        receiptHandles.add(message.getReceiptHandle());
                    }
                    // 回调删除
                    mqConsumer.ackMessage(receiptHandles);
                }
            }catch (Exception ex)
            {
                System.out.println("error:" + ex.getMessage());
                mqClient.close();
            }
        }
    }
}

参考链接

RocketMQ HTTP 协议 SDK

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
5月前
|
数据采集 Java
selenium+java入门demo
selenium+java入门demo
45 4
|
6月前
|
消息中间件 API 开发工具
如何实现一个简单易用的 RocketMQ SDK
2018 年,做为架构负责人,接到一个架构需求:实现一个简单易用的 RocketMQ SDK 。 因为各个团队 RocketMQ 原生客户端配置起来千奇百怪,有的配置存在风险,各团队负责人都需要一个简洁易用的 RocketMQ SDK 。 我立马调研相关开源的方案,当时 RocketMQ-Spring 项目并没有开源,而阿里云的 ONS SDK 是开源的,我只能讲目标转向 阿里云 ONS 。 通过学习 ONS 的设计方式,我对于 RocketMQ 的客户端原理有了进一步了解,同时参考 ONS 的设计,也实现了公司内部使用的 RocketMQ SDK 。
如何实现一个简单易用的 RocketMQ SDK
|
3月前
|
Java
MQTT(EMQX) - Java 调用 MQTT Demo 代码
MQTT(EMQX) - Java 调用 MQTT Demo 代码
155 0
MQTT(EMQX) - Java 调用 MQTT Demo 代码
|
4月前
|
JSON 运维 Serverless
Serverless 应用引擎使用问题之ThinkPHP框架是否有基于SDK的demo
阿里云Serverless 应用引擎(SAE)提供了完整的微服务应用生命周期管理能力,包括应用部署、服务治理、开发运维、资源管理等功能,并通过扩展功能支持多环境管理、API Gateway、事件驱动等高级应用场景,帮助企业快速构建、部署、运维和扩展微服务架构,实现Serverless化的应用部署与运维模式。以下是对SAE产品使用合集的概述,包括应用管理、服务治理、开发运维、资源管理等方面。
|
3月前
|
缓存 JavaScript 前端开发
微信 JS-SDK Demo “分享信息设置” API 及数字签名生成方法(NodeJS版本)
微信 JS-SDK Demo “分享信息设置” API 及数字签名生成方法(NodeJS版本)更新时间(2020-10-29)
|
5月前
|
消息中间件 负载均衡 网络性能优化
消息队列 MQ产品使用合集之 终端sdk和云端sdk的区别是什么
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
4月前
|
消息中间件 负载均衡 Java
JAVA面试之MQ
JAVA面试之MQ
69 0
|
6月前
|
消息中间件 网络协议 开发工具
MQ产品使用合集之rocketmq5.x只有tcp接入点吗,python sdk需要http接入点,请问怎么使用
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
205 2
|
6月前
|
消息中间件 Java RocketMQ
MQ产品使用合集之在同一个 Java 进程内建立三个消费对象并设置三个消费者组订阅同一主题和标签的情况下,是否会发生其中一个消费者组无法接收到消息的现象
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
121 1
|
5月前
|
Java
Java Swing 层次面板(JLayeredPanel) Demo
Java Swing 层次面板(JLayeredPanel) Demo
37 0