花于陌上开个人站

第一章 RocketMQ安装与简介 v1.0 ​

第一部分 安装 ​

一、使用docker安装 ​

bash
docker pull rocketmqinc/rocketmq
#创建一个数据目录
mkdir -p /docker/rocketmq/nameserver/logs /docker/rocketmq/nameserver/store
#运行
docker run -d --restart=always --name rmqnamesrv --privileged=true -p 9876:9876 -v /docker/rocketmq/nameserver/logs:/root/logs -v /docker/rocketmq/nameserver/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv
#安装broker
mkdir /opt/docker/rocketmq
vim /opt/docker/rocketmq/broker.conf  #文件内容见下面
#启动broker
docker run -d --restart=always --name rmqbroker --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 --privileged=true -v /docker/rocketmq/data/broker/logs:/root/logs -v /docker/rocketmq/data/broker/store:/root/store -v /docker/rocketmq/conf/broker.conf:/opt/docker/rocketmq/broker.conf -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/broker.conf
#安装控制台
docker pull pangliang/rocketmq-console-ng
#启动控制台
docker run -d --restart=always --name rmqadmin -e "JAVA_OPTS=-Drocketmq.namesrv.addr=<ip>:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8080:8080 pangliang/rocketmq-console-ng

上面的部署包含 NameServer、Broker 和管理控制台三个部分：NameServer 使用 9876 端口，Broker 使用 10911 和 10909 端口，控制台使用 8080 端口。

bash
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
brokerIP1 = 主机的IP

第二部分 核心概念 ​

一、RocketMQ简介 ​

RocketMQ是阿里巴巴开源的分布式消息中间件，后来捐赠给Apache基金会成为顶级项目。它具有高可用、高可靠、高吞吐量的特点，广泛应用于订单、交易、充值、流计算、日志流式处理等领域。

二、核心组件 ​

1.NameServer ​

NameServer是RocketMQ的注册中心，负责管理Broker的路由信息。每个Broker启动时会向所有NameServer注册，Producer和Consumer也会从NameServer获取路由信息。

2.Broker ​

Broker是消息存储和转发的服务端，负责接收Producer发送的消息并存储，同时将消息投递给Consumer。

3.Producer ​

Producer是消息的生产者，负责发送消息到Broker。

4.Consumer ​

Consumer是消息的消费者，负责从Broker拉取或推送消息进行处理。

三、消息模型 ​

1.基本消息（Normal） ​

普通消息，Producer发送消息到Topic，Consumer订阅Topic进行消费。

2.顺序消息（Ordered） ​

保证消息按照发送顺序进行消费，分为全局顺序和分区顺序。

  • 全局顺序：一个Topic只有一个队列，所有消息按顺序消费
  • 分区顺序：相同业务Key的消息按顺序消费

3.延时消息（Delayed） ​

消息发送后不会立即投递，而是等待指定时间后才投递给Consumer。

4.事务消息（Transactional） ​

支持分布式事务，保证消息发送和本地事务的一致性。

第三部分 基本使用 ​

一、Spring Boot集成 ​

1.添加依赖 ​

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

2.配置文件 ​

yaml
rocketmq:
  name-server: localhost:9876
  producer:
    group: my-group

二、发送消息 ​

1.同步发送 ​

java
@Autowired
private RocketMQTemplate rocketMQTemplate;

public void sendSyncMessage() {
    rocketMQTemplate.syncSend("topic-name", "Hello RocketMQ");
}

2.异步发送 ​

java
public void sendAsyncMessage() {
    rocketMQTemplate.asyncSend("topic-name", "Hello RocketMQ", new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println("发送成功");
        }
        @Override
        public void onException(Throwable e) {
            System.out.println("发送失败");
        }
    });
}

三、消费消息 ​

java
@Component
@RocketMQMessageListener(topic = "topic-name", consumerGroup = "my-group")
public class MyConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息: " + message);
    }
}