rocketMq整合springBoot快速入门

一、什么是消息队列 MQ？

MQ 是什么？队列是什么，MQ 我们可以理解为消息队列，队列我们可以理解为管道。以管道的方式做消息传递。

场景：

1.其实我们在双11的时候，当我们凌晨大量的秒杀和抢购商品，然后去结算的时候，就会发现，界面会提醒我们，让我们稍等，以及一些友好的图片文字提醒。而不是像前几年的时代，动不动就页面卡死，报错等来呈现给用户。

在这业务场景中，我们就可以采用队列的机制来处理，因为同时结算就只能达到这么多。

2.在我们平时的超市中购物也是一样，当我们在结算的时候，并不会一窝蜂一样涌入收银台，而是排队结算。这也是队列机制。

对，就是排队。一个接着一个的处理，不能插队。

二、rocketMq的配置

首先登录阿里云，搜索mq，点击立即开通，然后就会有管理控制平台，点击进去

1、创建消息主题（Topic）

消息主题（Topic）是 MQ 里对消息进行的一级归类，比如可以创建“Topic_Trade”这一主题用来识别交易类消息。 使用MQ的第一步需要先为您的应用创建 Topic。

您可以按照以下步骤创建 Topic：

1. 登录MQ 控制台，默认进入Topic 管理页面。

2. 在页面上方选择相应的地域（比如公网域），然后单击创建 Topic按钮。

   注意：

   • 如果只是测试，或者需要在本地（非阿里云 ECS 服务器）使用 MQ 服务，请将 Topic 创建在公网环境。 生产端和消费端可以部署在本地或者部署在任意地域的 ECS 上，前提是本地服务器或者相应的 ECS 需要能够访问公网。
   • 如果在生产环境使用 MQ 服务，需要将应用程序部署在阿里云 ECS 上，同时 Topic 也需要在应用程序所在的区域（即所部署的 ECS 区域）进行创建。
   • Topic 不能跨域使用。 比如 Topic 创建在“华北 2”这个域，那么消息生产端和消费端也必须运行在“华北 2”的 ECS 上。
   • 有关域的详细介绍请参见 ECS 文档中的地域和可用区。

3. 在创建 Topic对话框输入 Topic 名称及备注，单击确定。 您创建的 Topic 将出现在 Topic 列表中。

   注意：Topic 名称必须全局唯一。 如果名称已经被其他用户使用，您将无法创建相同名称的 Topic。

消息类型

消息类型分为以下几种：

• 无序消息：不保证先入先出（FIFO）的严格顺序，包含普通消息、定时/延时消息、事务消息。 建议创建不同的 Topic 来发送不同类型的消息，例如用 Topic A 发送普通消息，Topic B 发送事务消息， Topic C 发送延时/定时消息。

• 全局顺序消息：所有消息严格按照 FIFO 的严格顺序进行生产和消费。

• 分区顺序消息：消息根据 sharding key 进行分区，提高整体并发度与使用性能。 同一个分区的消息严格按照 FIFO 的严格顺序进行生产和消费。

• Kafka 消息：兼容 MQ 及 Kafka 协议（kafka-client-0.10 及以上版本）的消息类型。

  
  image.png

  创建topic成功的样子

2、创建生产者（Producer ID）

点击创建生产者

image.png

3、创建消费者（Consumer ID）

点击创建消费者

image.png

创建阿里云 AccessKey 和 SecretKey

在调用 SDK/API 进行消息发送和订阅的时候，除了需要指定创建的 Topic， Producer ID 以及 Consumer ID 以外，还需输入您在 RAM 控制台创建的身份验证信息，即 Access Key ID 和 Acess Key Secret。

关于如何创建 AccessKey 和 SecretKey, 请参阅创建AccessKey。

image.png

4、获取接入域名

在控制台创建 Topic 资源后，您还需要通过控制台获取 Producer ID 的 TCP 接入域名和 Consumer ID 的 TCP 接入域名。
请按照以下步骤获取 Producer ID 的接入域名或 Consumer ID 的接入域名：

1、在 MQ 控制台左侧菜单栏，选择生产者管理或消费者管理。
2、找到您创建的 Producer ID 或 Consumer ID，单击右侧操作列中的获取接入点。
3、在获取接入点提示框，单击复制。
完成以上准备工作后，您就可以运行示例代码，用 MQ 进行消息发送和订阅了。

image.png

5、发送消息

您可以通过控制台发送消息或者调用 SDK/API 发送消息。

• 控制台发送消息：用于快速验证 Topic 资源的可用性。

• 调用 SDK/API 发送消息：用于生产环境下使用 MQ。

控制台发送消息步骤如下：

1. 在 MQ 控制台的左侧菜单栏单击生产者管理。
2. 在列表中找到您刚刚创建的 Topic，单击右侧操作栏里的发送。
3. 在发送消息对话框输入消息的具体内容，单击确定。控制台会返回消息发送成功通知以及相应的 Message ID。

调用 SDK/API 发送消息

1、导入jar包

  <dependency>
       <groupId>com.aliyun.openservices</groupId>
       <artifactId>ons-client</artifactId>
       <version>"XXX"</version>
  </dependency>

2. • 关于 Java SDK 的最新版本号，请查看版本说明。
     示例代码-----说明：关于 TCP 接入点域名，请进入 MQ 控制台的生产者管理页面，在 PID 右侧操作列单击获取接入点按钮获取
     因为阿里云没有jar可能要手动安装，要先配置maven环境变量然后在cmd首先定位你下载jar的地方然后输入maven命令：

mvn install:install-file -Dfile="ons-client-1.7.8.Final.jar" -DgroupId=com.aliyun.openservices -DartifactId=ons-client -Dversion=1.7.8 -Dpackaging=jar

package com.rocket.rocket.test;

import com.aliyun.openservices.ons.api.*;

import java.util.Date;
import java.util.Properties;
/**
 * 要去RAM完成角色授权要不然会报com.aliyun.openservices.ons.api.exception.ONSClientException: Receive a broker exception
 */
public class ProducerTest {
        public static void main(String[] args) {
            Properties properties = new Properties();
            // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey,"XXX");
            // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            //设置发送超时时间，单位毫秒
            properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
            //  设置 TCP 接入域名，进入 MQ 控制台的生产者管理页面，在右侧操作栏单击获取接入点获取（此处以公共云生产环境为例）
            properties.put(PropertyKeyConst.ONSAddr,
                    "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet");

            Producer producer = ONSFactory.createProducer(properties);
            // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
            producer.start();

            //循环发送消息
            for (int i = 0; i < 100; i++){
                Message msg = new Message( //
                        // Message 所属的 Topic
                        "smile_1",
                        // Message Tag 可理解为 Gmail 中的标签，对消息进行再归类，方便 Consumer 指定过滤条件在 MQ 服务器过滤
                        "TagA",
                        // Message Body 可以是任何二进制形式的数据， MQ 不做任何干预，
                        // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                        "Hello MQ".getBytes());
                // 设置代表消息的业务关键属性，请尽可能全局唯一。
                // 以方便您在无法正常收到消息情况下，可通过阿里云服务器管理控制台查询消息并补发
                // 注意：不设置也不会影响消息正常收发
                msg.setKey("ORDERID_" + i);

                try {
                    SendResult sendResult = producer.send(msg);
                    // 同步发送消息，只要不抛异常就是成功
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                    }
                }
                catch (Exception e) {
                    // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                    e.printStackTrace();
                }
            }

            // 在应用退出前，销毁 Producer 对象
            // 注意：如果不销毁也没有问题
            producer.shutdown();
        }
    }

查看消息是否发送成功(成功的状态)

image.png

6：订阅消息

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Properties;
public class ConsumerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // 您在控制台创建的 Consumer ID
        properties.put(PropertyKeyConst.ConsumerId, "CID_smile1");
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.AccessKey, "XXX");
        // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // 设置 TCP 接入域名（此处以公共云生产环境为例）
        properties.put(PropertyKeyConst.ONSAddr,
          "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet");
          // 集群订阅方式 (默认)
          // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
          // 广播订阅方式
          // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() { //订阅多个 Tag
            public Action consume(Message message, ConsumeContext context) {
                System.out.println("Receive: " + message);
                return Action.CommitMessage;
            }
        });
        //订阅另外一个 Topic
        consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() { //订阅全部 Tag
            public Action consume(Message message, ConsumeContext context) {
                System.out.println("Receive: " + message);
                return Action.CommitMessage;
            }
        });
        consumer.start();
        System.out.println("Consumer Started");
    }
}

查看消息订阅是否成功

找到要查看的 Topic，单击右侧操作选项里的消费者状态。
如果是否在线显示为是，则说明订阅端已成功启动。如果消费者状态是否在线显示为否，说明消费端没有启动或者启动失败。

image.png image.png

完成以上所有步骤后，您就成功接入了 MQ 服务，可以用 MQ 进行消息发送和订阅了。