RocketMQ的发展历史
Apache RocketMQ 是一个开源的分布式消息中间件系统，最初由阿里巴巴集团开发并于2016年贡献给 Apache 软件基金会。以下是 RocketMQ 的发展历史的主要里程碑：
-
2012年：阿里巴巴内部项目
-
RocketMQ 最初是阿里巴巴内部的一个消息中间件项目，用于满足阿里巴巴集团内部的消息传递需求。
-
-
2016年5月：贡献给 Apache 软件基金会
-
阿里巴巴宣布将 RocketMQ 开源并贡献给 Apache 软件基金会，成为 Apache 孵化器项目（Incubator）的一部分。
-
-
2017年11月：成为 Apache 顶级项目
-
RocketMQ 成为 Apache 顶级项目（Top-Level Project），标志着它已成为一个成熟的开源项目。
-
RocketMQ和Kafka
RocketMQ 和 Apache Kafka 都是开源的分布式消息中间件系统，它们用于实现高可靠性、高吞吐量的消息传递，RocketMQ的核心设计借鉴了Kafka，两者之间存在相似之处。
1. 发展历史：
-
RocketMQ 最初由阿里巴巴开发，后来贡献给 Apache 软件基金会。
-
Kafka 由LinkedIn开发，后来成为了Apache软件基金会的顶级项目。
2. 数据模型：
-
RocketMQ 使用主题（Topic）和标签（Tag）来组织消息。消息可以按主题进行分组，标签可以用来进一步过滤消息。
-
Kafka 使用主题（Topic）来组织消息。消息被发布到主题，消费者可以订阅主题来接收消息。
3. 存储机制：
-
RocketMQ 使用可定制的存储引擎来存储消息，默认使用的是 Apache RocketMQ Store。
-
Kafka 使用分布式文件系统（如 Apache Hadoop HDFS）来存储消息。
4. 消费模型：
-
RocketMQ 支持消息广播和集群模式。消息广播模式允许多个消费者同时接收相同的消息，而集群模式确保每个消息只有一个消费者接收。
-
Kafka 使用发布-订阅模型，其中每个主题的消息可以被多个消费者订阅。
5. 消费者偏移：
-
RocketMQ 使用消费者组概念，允许多个消费者协同消费消息。RocketMQ会跟踪每个消费者组的消费进度。
-
Kafka 使用偏移量（offset）来跟踪每个消费者的进度。消费者需要自己维护偏移量。
6. 消息保证：
-
RocketMQ 提供了严格有序消息传递的支持，可以用于需要严格的消息顺序性的场景。
-
Kafka 提供至少一次交付保证，但在某些情况下可能出现消息重复或乱序。
快速上手RocketMQ
简要介绍如何安装单机版的RocketMQ，以及简单地收发消息。
1.下载RocketMQ
地址下载 | RocketMQ，选择Binary下载的原因是Binary版本是一些编译好的jar和辅助的shell脚本。
2.解压缩
使用FinalShell远程连接工具上传下载好的压缩包，存放路径在usr/local/rocketmq，把压缩包解压到目录下。
# 创建文件夹目录 cd /usr/local/ mkdir rocketmq # 解压，我下上传是rocketmq-all-5.1.3-bin-release.zip unzip rocketmq-all-5.1.3-bin-release.zip
3.启动消息队列服务
单机的消息队列服务比较简单，不需要配置，一次启动NameServer和Broker就行。建议使用后台启动的方式。
#进入RocketMQ的bin目录下 cd rocketmq-all-5.1.3-bin-release/bin/ #后台启动NameServer zookeeper-server-start.sh #查看NameServer启动日志 tail -f ~/logs/rocketmqlogs/namesrv.log #后台启动Broker zookeeper-server-stop.sh #查看Broker启动日志 tail -f ~/logs/rocketmqlogs/broker.log
4.内存不足启动报错
查看Broker启动日志，提示找不到文件，那就查看当前目录下的nohup.out。
如果报错“There is insufficient memory for the Java Runtime Environment to continue.”，那就是RocketMQ启动内存不足导致的。
解决方案
找到 runserver.sh 和 runbroker.sh 以及 tools.sh 修改下 JAVA_OPT 的参数，都在JVM Configuration下面
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn125m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
对应参数说明
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512g"
Xms 是指设定程序启动时占用内存大小。一般来讲，大点，程序会启动的快一点，但是也可能会导致机器暂时
间变慢。
Xmx 是指设定程序运行期间最大可占用的内存大小。如果程序运行需要占用更多的内存，超出了这个设置值，
就会抛出OutOfMemory异常。
xmn 年轻代的heap大小，一般设置为Xmx的3、4分之一
保存修改后，重新启动namesrv和broker就可以了。
5.停止服务，建议先停止broker，然后停止namesrv
-
停止Broker
sh mqshutdown broker
-
停止NameSrv
sh mqshutdown namesrv
简单示例
以下是一个简单的 Java 示例，演示如何使用 Apache RocketMQ 发送和接收消息。在运行这个示例之前，确保已经设置好 RocketMQ 环境并正确配置了生产者和消费者。
发送消息的生产者示例：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
public class RocketMQProducer {public static void main(String[] args) throws Exception {// 创建生产者实例DefaultMQProducer producer = new DefaultMQProducer("producer_group");producer.setNamesrvAddr("localhost:9876");
// 启动生产者producer.start();
// 创建消息Message message = new Message("topic_name", "tag_name", "Hello, RocketMQ".getBytes());
// 发送消息producer.send(message);
// 关闭生产者producer.shutdown();}
}
接收消息的消费者示例：
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class RocketMQConsumer {public static void main(String[] args) throws Exception {// 创建消费者实例DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");consumer.setNamesrvAddr("localhost:9876");
// 订阅消息主题和标签consumer.subscribe("topic_name", "*");
// 注册消息监听器consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {for (MessageExt message : messages) {System.out.println("Received message: " + new String(message.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});
// 启动消费者consumer.start();
System.out.println("Consumer started");}
}
在上述示例中，生产者向名为 "topic_name" 的主题发送消息，而消费者订阅相同的主题，并通过消息监听器接收并处理消息。
![一、PHP环境搭建[phpstorm]](https://img-blog.csdnimg.cn/2e76ed6278314a99890a21337e7fa99a.png)
