一、什么是消息队列 MQ？
MQ 是什么？队列是什么，MQ 我们可以理解为消息队列，队列我们可以理解为管道。以管道的方式做消息传递。
场景：
1.其实我们在双11的时候，当我们凌晨大量的秒杀和抢购商品，然后去结算的时候，就会发现，界面会提醒我们，让我们稍等，以及一些友好的图片文字提醒。而不是像前几年的时代，动不动就页面卡死，报错等来呈现给用户。
在这业务场景中，我们就可以采用队列的机制来处理，因为同时结算就只能达到这么多。
2.在我们平时的超市中购物也是一样，当我们在结算的时候，并不会一窝蜂一样涌入收银台，而是排队结算。这也是队列机制。
对，就是排队。一个接着一个的处理，不能插队。
二、rocketMq的配置
首先登录阿里云，搜索mq，点击立即开通，然后就会有管理控制平台，点击进去
1、创建消息主题（Topic）
消息主题（Topic）是 MQ 里对消息进行的一级归类，比如可以创建“Topic_Trade”这一主题用来识别交易类消息。 使用MQ的第一步需要先为您的应用创建 Topic。
您可以按照以下步骤创建 Topic：
登录MQ 控制台，默认进入Topic 管理页面。
-
在页面上方选择相应的地域（比如公网域），然后单击创建 Topic按钮。
注意：
- 如果只是测试，或者需要在本地（非阿里云 ECS 服务器）使用 MQ 服务，请将 Topic 创建在公网环境。 生产端和消费端可以部署在本地或者部署在任意地域的 ECS 上，前提是本地服务器或者相应的 ECS 需要能够访问公网。
- 如果在生产环境使用 MQ 服务，需要将应用程序部署在阿里云 ECS 上，同时 Topic 也需要在应用程序所在的区域（即所部署的 ECS 区域）进行创建。
- Topic 不能跨域使用。 比如 Topic 创建在“华北 2”这个域，那么消息生产端和消费端也必须运行在“华北 2”的 ECS 上。
- 有关域的详细介绍请参见 ECS 文档中的地域和可用区。
-
在创建 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）
点击创建生产者
3、创建消费者（Consumer ID）
点击创建消费者
创建阿里云 AccessKey 和 SecretKey
在调用 SDK/API 进行消息发送和订阅的时候，除了需要指定创建的 Topic， Producer ID 以及 Consumer ID 以外，还需输入您在 RAM 控制台创建的身份验证信息，即 Access Key ID 和 Acess Key Secret。
关于如何创建 AccessKey 和 SecretKey, 请参阅创建AccessKey。
4、获取接入域名
在控制台创建 Topic 资源后，您还需要通过控制台获取 Producer ID 的 TCP 接入域名和 Consumer ID 的 TCP 接入域名。
请按照以下步骤获取 Producer ID 的接入域名或 Consumer ID 的接入域名：
1、在 MQ 控制台左侧菜单栏，选择生产者管理或消费者管理。
2、找到您创建的 Producer ID 或 Consumer ID，单击右侧操作列中的获取接入点。
3、在获取接入点提示框，单击复制。
完成以上准备工作后，您就可以运行示例代码，用 MQ 进行消息发送和订阅了。
5、发送消息
您可以通过控制台发送消息或者调用 SDK/API 发送消息。
控制台发送消息：用于快速验证 Topic 资源的可用性。
调用 SDK/API 发送消息：用于生产环境下使用 MQ。
控制台发送消息步骤如下：
- 在 MQ 控制台的左侧菜单栏单击生产者管理。
- 在列表中找到您刚刚创建的 Topic，单击右侧操作栏里的发送。
- 在发送消息对话框输入消息的具体内容，单击确定。控制台会返回消息发送成功通知以及相应的 Message ID。
调用 SDK/API 发送消息
1、导入jar包
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>"XXX"</version>
</dependency>
- 关于 Java SDK 的最新版本号，请查看版本说明。
示例代码-----说明：关于 TCP 接入点域名，请进入 MQ 控制台的生产者管理页面，在 PID 右侧操作列单击获取接入点按钮获取
因为阿里云没有jar可能要手动安装，要先配置maven环境变量然后在cmd首先定位你下载jar的地方然后输入maven命令：
- 关于 Java SDK 的最新版本号，请查看版本说明。
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();
}
}
查看消息是否发送成功(成功的状态)
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，单击右侧操作选项里的消费者状态。
如果是否在线显示为是，则说明订阅端已成功启动。如果消费者状态是否在线显示为否，说明消费端没有启动或者启动失败。
完成以上所有步骤后，您就成功接入了 MQ 服务，可以用 MQ 进行消息发送和订阅了。
