Skip to content

Commit

Permalink
Add files via upload
Browse files Browse the repository at this point in the history
  • Loading branch information
@Baiyelovecode
Baiyelovecode authored Aug 13, 2024
1 parent e6a0873 commit bf48f4b
Showing 1 changed file with 142 additions and 105 deletions.
247 changes: 142 additions & 105 deletions docs/01-quickstart/04quickstartWithHelmInKubernetes.md
View file
Original file line number Diff line number Diff line change
@@ -1,143 +1,180 @@
### QuickStart
本文介绍在kubernetesk环境下如何使用Helm快速部署一个单节点单副本 RocketMQ 服务，并完成简单的消息收发。

#### 前提条件

- 一个运行中的 `Kubernetes` 集群
- 已安装并配置好的 `kubectl`
- 已安装的 `Helm`
- 64位 `JDK 1.8+`

#### 第一步：安装 Helm

确保你的系统上已经安装了 Helm：

```bash
$ helm version
```

如果未安装 Helm，可以使用以下命令进行安装：

```bash
$ curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash
```

#### 第二步：添加 RocketMQ Helm 仓库
#### 第二步：下载 RocketMQ Helm 仓库

添加 RocketMQ 的 Helm chart 仓库：
使用 Helm chart 部署 RocketMQ：

```bash
$ helm repo add rocketmq-repo https://helm-charts.itboon.top/rocketmq
$ helm repo update rocketmq-repo
$ cd /opt/helm/
$ helm pull oci://registry-1.docker.io/apache/rocketmq --version 0.0.1
$ tar -zxvf rocketmq-0.0.1.tgz
```

#### 第三步：部署 RocketMQ

使用 Helm chart 部署 RocketMQ：

##### 关闭持久化存储：

```bash
#这里关闭持久化存储，仅演示部署效果
$ helm upgrade --install rocketmq \
--namespace rocketmq-demo \
--create-namespace \
--set broker.persistence.enabled="false" \
rocketmq-repo/rocketmq
#修改values.yaml中的配置（根据实际需求修改相关配置信息，如资源大小、副本数等）
$ helm install rocketmq-demo ./rocketmq

#查看pod状态
$ kubectl get pods -o wide -n rocketmq-demo
$ kubectl get pods -o wide -n default
NAME READY STATUS RESTARTS AGE IP NODE NOMINATED NODE READINESS GATES
rocketmq-demo-broker-0 1/1 Running 0 6h3m 192.168.58.225 k8s-node02 <none> <none>
rocketmq-demo-nameserver-757877747b-k669k 1/1 Running 0 6h3m 192.168.58.226 k8s-node02 <none> <none>
rocketmq-demo-proxy-6c569bd457-wcg6g 1/1 Running 3 (6h2m ago) 6h3m 192.168.85.227 k8s-node01 <none> <none>
```

#### 第四步：验证消息发送和接收
使用JAVA SDK测试消息收发（由于本地网络和k8s网络不属于同一内网，需要将项目本地打包后放在远程运行），具体如下：

1）IDEA中创建一个Java工程。

##### 开启持久化存储：

```bash
## 这里开启持久化存储（部署2个master节点，0副本），使用宿主机磁盘存储（线上环境不推荐），仅演示部署效果 ##

#创建storageClass
$ touch storageClass.yaml
kind: StorageClass
apiVersion: storage.k8s.io/v1
metadata:
name: local-storage
provisioner: kubernetes.io/no-provisioner
volumeBindingMode: WaitForFirstConsumer

$ kubectl apply -f storageClass.yaml

#创建pv，用于master一个节点
$ touch pv-local.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
name: broker-storage
spec:
capacity:
storage: 20Gi # 替换为你的存储大小
volumeMode: Filesystem
accessModes:
- ReadWriteOnce
persistentVolumeReclaimPolicy: Retain
storageClassName: local-storage
local:
path: /mnt/data # 替换为你的宿主机目录
nodeAffinity:
required:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/hostname
operator: In
values:
- k8s-node1 # 替换为你的节点名称
- k8s-node2 # 替换为你的节点名称
- k8s-node3 # 替换为你的节点名称

$ kubectl apply -f pv-local.yaml

#创建pv，用于master另一个节点
$ touch pv2-local.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
name: broker-storage-2
spec:
capacity:
storage: 20Gi # 替换为你的存储大小
volumeMode: Filesystem
accessModes:
- ReadWriteOnce
persistentVolumeReclaimPolicy: Retain
storageClassName: local-storage
local:
path: /mnt/data # 替换为你的宿主机目录
nodeAffinity:
required:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/hostname
operator: In
values:
- k8s-node1 # 替换为你的节点名称
- k8s-node2 # 替换为你的节点名称
- k8s-node3 # 替换为你的节点名称

$ kubectl apply -f pv2-local.yaml

#查看pvc是否与pv绑定
$ kubectl get pvc -n rocketmq-demo

#查看pod状态
$ kubectl get pods -o wide -n rocketmq-demo
2）在 *pom.xml* 文件中添加以下依赖引入Java依赖库：
```xml
......
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client-java</artifactId>
<version>5.0.7</version>
</dependency>
.....
```
3）使用mqadmin创建 Topic和消费者。
```bash
#登录pod内（需要管理工具），也可以在宿主机执行
$ kubectl exec -ti rocketmq-demo-broker-0 -- /bin/bash

#通过mqadmin创建Topic
$ sh mqadmin updatetopic -t TestTopic -c DefaultCluster

#通过mqadmin创建消费者Group
$ sh mqadmin updateSubGroup -c DefaultCluster -g TestGroup
```
4）在已创建的Java工程中，创建发送普通消息程序 （ProducerDemo.java），示例代码如下：

```java
package com.rocketmq.producer;

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;


public class ProducerDemo {
public static void main(String[] args) throws ClientException {
// 接入点地址，需要设置成Proxy的地址和端口列表，以下为k8s环境下rocketmq的proxy地址。
String endpoint = "192.168.85.227:8081";
// 消息发送的目标Topic名称，需要提前创建。
String topic = "TestTopic";
ClientServiceProvider provider = ClientServiceProvider.loadService();
ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
ClientConfiguration configuration = builder.build();
// 初始化Producer时需要设置通信配置以及预绑定的Topic。
Producer producer = provider.newProducerBuilder()
.setTopics(topic)
.setClientConfiguration(configuration)
.build();
// 普通消息发送。
Message message = provider.newMessageBuilder()
.setTopic(topic)
// 设置消息索引键，可根据关键字精确查找某条消息。
.setKeys("messageKey")
// 设置消息Tag，用于消费端根据指定Tag过滤消息。
.setTag("messageTag")
// 消息体。
.setBody("messageBody".getBytes())
.build();
try {
// 发送消息，需要关注发送结果，并捕获失败等异常。
SendReceipt sendReceipt = producer.send(message);
System.out.println("Send message successfully, messageId="+sendReceipt.getMessageId());
} catch (ClientException e) {

}
// producer.close();
}
}
```
5）在已创建的Java工程中，创建订阅普通消息程序(Consumer.java)。Apache RocketMQ 支持[SimpleConsumer](https://rocketmq.apache.org/zh/docs/featureBehavior/06consumertype)和[PushConsumer](https://rocketmq.apache.org/zh/docs/featureBehavior/06consumertype)两种消费者类型 ，这里使用PushConsumers类型。

```java

package com.rocketmq.consumer;

import java.io.IOException;
import java.util.Collections;
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.util.List;

public class Consumer {
public static void main(String[] args) throws ClientException, IOException, InterruptedException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();
// 接入点地址，需要设置成Proxy的地址和端口列表，以下为k8s环境下rocketmq的proxy地址。
String endpoints = "192.168.85.227:8081";
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
.build();
// 订阅消息的过滤规则，表示订阅所有Tag的消息。
String tag = "*";
FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
// 为消费者指定所属的消费者分组，Group需要提前创建。
String consumerGroup = "TestGroup";
// 指定需要订阅哪个目标Topic，Topic需要提前创建。
String topic = "TestTopic";
// 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
.setClientConfiguration(clientConfiguration)
// 设置消费者分组。
.setConsumerGroup(consumerGroup)
// 设置预绑定的订阅关系。
.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
// 设置消费监听器。
.setMessageListener(messageView -> {
// 处理消息并返回消费结果。
System.out.println("Consume message successfully, messageId="+ messageView.getMessageId());
return ConsumeResult.SUCCESS;
})
.build();
Thread.sleep(Long.MAX_VALUE);
// 如果不需要再使用 PushConsumer，可关闭该实例。
// pushConsumer.close();
}
}
```

#### 第四步：验证安装

现在你可以使用客户端工具连接到 RocketMQ 进行消息的发送和接收。





更多详细的配置和选项，请参考 [RocketMQ Helm Chart 文档](https://github.com/itboon/rocketmq-helm)。
#### 第四步：释放rocketmq资源
``` bash
#释放所有rocketmq资源
$ helm uninstall rocketmq-demo
```

0 comments on commit bf48f4b

Please sign in to comment.