rocketMQ消息队列简介及其实例

一、RocketMQ 核心的四大组件：

Producer：就是消息生产者，可以集群部署。它会先和 NameServer 集群中的随机一台建立长连接，得知当前要发送的 Topic 存在哪台 Broker Master上，然后再与其建立长连接，支持多种负载平衡模式发送消息。

Consumer：消息消费者，也可以集群部署。它也会先和 NameServer 集群中的随机一台建立长连接，得知当前要消息的 Topic 存在哪台 Broker Master、Slave上，然后它们建立长连接，支持集群消费和广播消费消息。

Broker：主要负责消息的存储、查询消费，支持主从部署，一个 Master 可以对应多个 Slave，Master 支持读写，Slave 只支持读。Broker 会向集群中的每一台 NameServer 注册自己的路由信息。

NameServer：类似Zookeeper，是一个很简单的 Topic 路由注册中心，支持 Broker 的动态注册和发现，保存 Topic 和 Borker 之间的关系。通常也是集群部署，但是各 NameServer 之间不会互相通信， 各 NameServer 都有完整的路由信息，即无状态。

二、rocketmq基本工作流程：

可以理解为如下：

broker:消息处理

procucer:生成消息

consumer:消费消息

每个组件都可以部署成集群模式进行水平扩展。
消息由topic区分消息类型(一级分类)：如订单消息，物流消息等
tag为二级分类
message queue为消息类型下的消息队列。
用于并行发送和接受消息。

四、基础
分布式事务：
对于分布式事务，通俗地说就是，一次操作由若干分支操作组成，这些分支操作分属不同应用，分布在不同服务器上。分布式事务需要保证这些分支操作要么全部成功，要么全部失败。分布式事务与普通事务一样，就是为了保证操作结果的一致性。

事务消息：
RocketMQ提供了类似X/Open XA的分布式事务功能，通过事务消息能达到分布式事务的最终一致。XA是一种分布式事务解决方案，一种分布式事务处理模式。

半事务消息：
暂不能投递的消息，发送方已经成功地将消息发送到了Broker，但Broker未收到最终确认指令，此时该消息被标记成“暂不能投递”状态，即不能被消费者看到。处于该种状态下的消息即半事务消息。

本地事务状态：
Producer回调操作执行的结果为本地事务状态，其会发送给TC，而TC会再发送给TM。TM会根据TC发送来的本地事务状态来决定全局事务确认令。

// 描述本地事务执行状态 public enum LocalTransactionState {
COMMIT_MESSAGE, // 本地事务执行成功
ROLLBACK_MESSAGE, // 本地事务执行失败
UNKNOW, // 不确定，表示需要进行回查以确定本地事务的执行结果
}

RocketMQ中的消息回查设置：
关于消息回查，有三个常见的属性设置。它们都在broker加载的配置文件中设置，例如：

transactionTimeout=20，指定TM在 20 秒内应将最终确认状态发送给TC，否则引发消息回查。默认为 60 秒
transactionCheckMax=5，指定最多回查 5 次，超过后将丢弃消息并记录错误日志。默认 15 次。
transactionCheckInterval=10，指定设置的多次消息回查的时间间隔为 10 秒。默认为 60 秒。
五、Topic与Broker的关系：

• Borker中有一个或多个Topic
• Topic中有一个或多个MessageQueue

Topic可以自动创建和手动创建；

1、手动创建也叫预先创建，就是在使用Topic之前就创建，可以通过命令行或者通过RocketMQ的管理界面（可视化控制台）创建Topic。

/**
* 创建topic，参数分别是：borker的名称，topic的名称，queue的数量
* broker要和虚拟机broker.conf配置文件中brokername的名字一致
* newTopic的名字随便起，queueNum8的意思是新建的消息队列数为8个
*/
producer.createTopic("broker_haoke_im","my-topic",8);

2、自动创建就是设置了autoCreateTopicEnable =true;

六、下面以windows服务器为例演示使用rocketmq如下：

1、下载rocketmq的安装包：https://rocketmq.apache.org/zh/download

2、下载rocketmq仪表盘（也就是可视化操作界面，是一个完整的java项目可以用idea运行）

3、修改conf/broker.conf配置在末尾添加如下配置（IP使用自己的），并保存。

brokerIP1=192.168.31.199

namesrvAddr=192.168.31.199:9876

4、配置ROCKET_HOME环境变量，路径使用下载路径；path中配置%ROCKET_HOME%\bin即可

5、启动Namesrv

在rocketmq文件的bin目录下，进入cmd使用如下命令：start mqnamesrv.cmd

6、启动Broker：start mqbroker.cmd -n 127.0.0.1:9876  autoCreateTopicEnable=true  （也就是说,producer使用RocketMQTemplate发送的消息,就算Booker上的topic之前不存在,rocket也会帮我们创建好）

7、将仪表盘项目导入idea，然后打开application.properties文件修改rocket.config.namesrvAddr=localhost:9876;

8、启动仪表盘项目：浏览器输入http://localhost:8080/#/即可看到可视化界面；

9、java代码创建生产者和消费者：

创建普通springboot项目，添加依赖

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.1</version>
</dependency>

10、修改配置文件

# 应用名称
spring:
application:
name: rocket-producer
# 应用服务 WEB 访问端口
server:
port: 8002
rocketmq:
name-server: localhost:9876
producer:
group: my-group

11、创建测试代码

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class SendMessage {
@Resource
private RocketMQTemplate rocketMQTemplate;


@Scheduled(fixedRate = 5000)
public void run(){
//发送消息
rocketMQTemplate.convertAndSend("test-topic-1", "Hello, World!");

}
}

12、创建消费者项目（同上）

消费端测试代码：
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
class MyConsumer1 implements RocketMQListener<String> {

/**
*需要注意的是，onMessage（）封装了ACK机制，消费者往外抛异常时，RocketMQ认为消费失败，重新发送该条消息，否则默认消费成功
*/

@Override
public void onMessage(String s) {
System.out.println(s);
}
}

热门相关：冉冉心动   朕   最强反套路系统   特工重生：快穿全能女神   最强反套路系统