快速部署
快速部署一个单节点单副本 RocketMQ 服务,并完成简单的消息收发。
安装Apache RocketMQ
下载地址:RocketMQ官网下载
这里我们下载二进制包:rocketmq-all-5.3.0-bin-release.zip
直接解压即可:tar -zxvf rocketmq-all-5.3.0-bin-release.zip
重命名:mv rocketmq-all-5.3.0-bin-release rocketmq
启动NameServer
安装完RocketMQ包后,我们启动NameServer
### 启动namesrv
$ nohup sh bin/mqnamesrv &### 验证namesrv是否启动成功
$ tail -f nohup.out
The Name Server boot success. serializeType=JSON, address 0.0.0.0:9876
启动Broker+Proxy
NameServer成功启动后,我们启动Broker和Proxy。这里我们使用 Local 模式部署,即 Broker 和 Proxy 同进程部署。
### 先启动broker
$ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &### 验证broker是否启动成功
$ tail -f nohup.out
Wed Sep 11 18:35:56 CST 2024 rocketmq-proxy startup successfully
工具测试消息收发
export NAMESRV_ADDR=localhost:9876
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
命令行测试
创建 Topic
bin/mqadmin updateTopic -c DefaultCluster -n localhost:9876 -t studyMq
查看 Topic 列表
bin/mqadmin topicList -n localhost:9876
发送消息
bin/mqadmin sendMessage -n localhost:9876 -t studyMq -p 111111
消费消息
bin/mqadmin consumeMessage -n localhost:9876 -t studyMq
SDK测试
生产消息
package rocketmq;import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;import java.time.Duration;public class ProducerExample {public static void main(String[] args) throws Exception {// 接入点地址,需要设置成Proxy的地址和端口列表,一般是xxx:8080;xxx:8081。String endpoint = "127.0.0.1:8080";// 消息发送的目标Topic名称,需要提前创建。String topic = "studyMq";ClientServiceProvider provider = ClientServiceProvider.loadService();ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);builder.setRequestTimeout(Duration.ofSeconds(10));ClientConfiguration configuration = builder.build();// 初始化Producer时需要设置通信配置以及预绑定的Topic。Producer producer = provider.newProducerBuilder().setTopics(topic).setClientConfiguration(configuration).build();// 普通消息发送。Message message = provider.newMessageBuilder().setTopic(topic)// 设置消息Tag,用于消费端根据指定Tag过滤消息。.setTag("yyh")// 消息体。.setBody("good night".getBytes()).build();try {// 发送消息,需要关注发送结果,并捕获失败等异常。SendReceipt sendReceipt = producer.send(message);System.out.println("Send message successfully, messageId:" + sendReceipt.getMessageId());} catch (ClientException e) {System.out.println("Failed to send message:" + e);}// producer.close();}
}
消费消息
package rocketmq;import lombok.Data;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;import java.io.IOException;
import java.time.Duration;
import java.util.Collections;@Data
public class ConsumerExample {private ConsumerExample() {}public static void main(String[] args) throws ClientException, IOException, InterruptedException {final ClientServiceProvider provider = ClientServiceProvider.loadService();// 接入点地址,需要设置成Proxy的地址和端口列表,一般是xxx:8081;xxx:8081。String endpoints = "127.0.0.1:8080";ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints(endpoints).setRequestTimeout(Duration.ofSeconds(20)).build();// 订阅消息的过滤规则,表示订阅所有Tag的消息。String tag = "*";FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);// 为消费者指定所属的消费者分组,Group需要提前创建。String consumerGroup = "YourConsumerGroup";// 指定需要订阅哪个目标Topic,Topic需要提前创建。String topic = "studyMq";// 初始化PushConsumer,需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。PushConsumer pushConsumer = provider.newPushConsumerBuilder().setClientConfiguration(clientConfiguration)// 设置消费者分组。.setConsumerGroup(consumerGroup)// 设置预绑定的订阅关系。.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))// 设置消费监听器。.setMessageListener(messageView -> {// 获取消息体的ByteBufferByteBuffer byteBuffer = messageView.getBody();// 将ByteBuffer转换为字节数组byte[] bytes = new byte[byteBuffer.remaining()];byteBuffer.get(bytes); //将ByteBuffer中的数据读取到字节数组// 将字节数组转换为字符串String msgBody = new String(bytes, StandardCharsets.UTF_8);// 处理消息并返回消费结果。System.out.println("Consume message successfully, messageId=" + messageView.getMessageId() +",msg=" + msgBody);return ConsumeResult.SUCCESS;}).build();Thread.sleep(Long.MAX_VALUE);// 如果不需要再使用 PushConsumer,可关闭该实例。// pushConsumer.close();}
}
关闭服务器
通过以下方式关闭服务
sh bin/mqshutdown brokersh bin/mqshutdown namesrv
注意:如果是部署在云服务器上,需要把9876、 8080、10911端口都对外授权
参考资料
《https://rocketmq.apache.org/zh/docs/quickStart/01quickstart》