快速部署

快速部署一个单节点单副本 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 -> {
  					// 获取消息体的ByteBuffer
                    ByteBuffer 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 broker

sh bin/mqshutdown namesrv

注意：如果是部署在云服务器上，需要把9876、 8080、10911端口都对外授权

参考资料
《https://rocketmq.apache.org/zh/docs/quickStart/01quickstart》