如何使用消息队列、Spring Boot和Kubernetes扩展微服务

当您进行大规模设计和构建应用程序时，您将面临两个重大挑战：可伸缩性和健壮性。

您应该精心设计您的服务，即使它受到间歇性重载，它仍然可靠地运行。以苹果商店为例，每年都有数百万的Apple客户预先注册购买新的iPhone，这是数百万人同时购买物品，苹果商店流量如果描述为每秒的请求数量，那么在下午6点到8点之间是有峰值最高峰。

现在想象一下，你的任务是构建这样的应用程序。

• 您正在建立一个商店，用户可以在那里购买自己喜欢的商品。
• 您构建一个微服务来呈现网页并提供静态资产。您还构建了一个后端REST API来处理传入的请求。
• 您希望将两个组件分开，因为使用相同的REST API，您可以为网站和移动应用程序提供服务。

你的商店上线了！

您决定将应用程序扩展为前端的四个实例和后端的四个实例，因为您预测网站比平常更繁忙。网站开始接收越来越多的流量。

前端服务正在处理繁忙流量。但是您注意到连接到数据库的后端正在努力跟上事务的数量，不用担心，您可以将后端的服务器数量扩展到8个实例。

收到的流量更多，后端无法应对。

一些服务开始丢弃连接。愤怒的客户与您的客户服务取得联系。而现在你被淹没在大流量中，你的后端无法应付这种大流量，失去了很多连接。电子商务网站丢失连接就是丢失交易，就是收入损失。

您的应用程序并非设计为健壮且高度可用：

• 前端和后端紧密耦合 -  实际上它不能在没有后端的情况下处理应用程序
• 前端和后端必须一致扩展 -  如果没有足够的后端，你可能会淹没在流量中
• 如果后端不可用，则无法处理传入的事务。

您可以重新设计体系结构，以便将前端和后端与队列分离：

前端将消息发布到队列，而后端则一次处理一个待处理消息。

新架构有一些明显的好处：

• 如果后端不可用，则队列充当缓冲区
• 如果前端产生的消息多于后端可以处理的消息，则这些消息将缓冲在队列中
• 您可以独立于前端扩展后端 - 即您可以拥有数百个前端服务和后端的单个实例

太好了，但是你如何构建这样的应用程序？您如何设计可处理数十万个请求的服务？您如何部署动态扩展的应用程序？在深入了解部署和扩展的细节之前，让我们关注应用程序。

编写Spring应用程序

现在服务有三个组件：前端，后端和消息代理。

前端是一个简单的Spring Boot Web应用程序，带有Thymeleaf模板引擎；后端是一个消耗队列消息的工作者。由于 Spring Boot与JSM具有出色的集成 ，因此您可以使用它来发送和接收异步消息。您可以在 learnk8s / spring-boot-k8s-hpa中 找到一个连接到JSM的前端和后端应用程序的示例项目。

请注意，该应用程序是用 Java 10编写的，才能利用 改进的 Docker 容器集成 。只有一个代码库，您可以将项目配置为作为前端或后端运行。

您应该知道该应用程序具有：

• 你可以购买物品的主页
• 管理面板，您可以在其中检查队列中的消息数
• /health当应用程序准备好接收流量时发出信号的端点
• /submit从表单接收提交并在队列中创建消息的端点
• 一个/metrics端点，以露出待处理的消息的数量在队列中（以后会更多）

该应用程序可以在两种模式下运行：

1. 作为前端，应用程序呈现人们可以购买物品的网页。
2. 作为工作者，应用程序等待队列中的消息并处理它们。

请注意，在示例项目中，通过等待五秒来模拟处理Thread.sleep(5000)。您可以通过更改application.yaml为您的值来配置任一模式的应用程序。

运行应用程序

默认情况下，应用程序作为前端和工作程序启动。您可以运行该应用程序，只要您在本地运行ActiveMQ实例，您就应该能够购买物品并让系统处理这些物品。

如果检查日志，则应该看到工作程序处理项目。

有效！编写Spring Boot应用程序很容易。

一个更有趣的主题是学习如何将Spring Boot连接到消息代理。

使用JMS发送和接收消息

Spring JMS（Java消息服务）是一种使用标准协议发送和接收消息的强大机制。如果您以前使用过JDBC API，那么您应该熟悉JMS API，因为它的工作方式类似。您可以使用JMS使用的最流行的消息代理是 ActiveMQ  - 一个开源消息服务器。使用这两个组件，您可以使用熟悉的接口（JMS）将消息发布到队列（ActiveMQ），并使用相同的接口来接收消息。更妙的是，Spring Boot与JMS的集成非常好，因此您可以立即加快速度。

实际上，以下短类封装了用于与队列交互的逻辑：

@Component
public class QueueService implements MessageListener {
private static final Logger LOGGER = LoggerFactory.getLogger(QueueService.class);
@Autowired
  private JmsTemplate jmsTemplate;
  public void send(String destination, String message) {
    LOGGER.info("sending message='{}' to destination='{}'", message, destination);
    jmsTemplate.convertAndSend(destination, message);
  }
@Override
  public void onMessage(Message message) {
    if (message instanceof ActiveMQTextMessage) {
      ActiveMQTextMessage textMessage = (ActiveMQTextMessage) message;
      try {
        LOGGER.info("Processing task " + textMessage.getText());
        Thread.sleep(5000);
        LOGGER.info("Completed task " + textMessage.getText());
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (JMSException e) {
        e.printStackTrace();
      }
    } else {
      LOGGER.error("Message is not a text message " + message.toString());
    }
  }
}

您可以使用该send方法将消息发布到命名队列。此外，Spring Boot将为onMessage每个传入消息执行该方法。

最后一个难题是如何让Spring Boot使用该类。您可以通过 在Spring Boot应用程序中注册侦听器来在 后台处理消息，如下所示：

@SpringBootApplication
@EnableJms
public class SpringBootApplication implements JmsListenerConfigurer {
  @Autowired
  private QueueService queueService;
public static void main(String[] args) {
    SpringApplication.run(SpringBootApplication.class, args);
  }
@Override
  public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
    endpoint.setId("myId");
    endpoint.setDestination("queueName");
    endpoint.setMessageListener(queueService);
    registrar.registerEndpoint(endpoint);
  }
}

其中id是使用者的唯一标识符，destination是队列的名称。

您可以从GitHub上的项目 中完整地读取Spring队列服务的源代码 。

部署

验证了应用程序的工作原理，现在是时候部署它了。您可以启动VPS，安装Tomcat，并花些时间制作自定义脚本来测试，构建，打包和部署应用程序。或者您可以编写您希望拥有的描述：一个消息代理和两个使用负载均衡器部署的应用程序。

Kubernetes之类的协调器可以阅读您的愿望清单并提供正确的基础设施，花在基础架构上的时间减少意味着更多的时间编码，这次你将把应用程序部署到Kubernetes。但在开始之前，您需要一个Kubernetes集群。

您可以注册Google云平台或Azure，并使用云提供商Kubernetes提供的服务。或者，您可以在将应用程序移动到云之前在本地尝试Kubernetes。

minikube是一个打包为虚拟机的本地Kubernetes集群。如果您使用的是Windows，Linux和Mac，那就太好了，因为创建群集需要五分钟。

您还应该安装kubectl客户端以连接到您的群集。

你可以找到关于如何安装的说明minikube，并kubectl从 官方文档 。如果您在Windows上运行，则应查看 有关如何安装Kubernetes和Docker的详细指南 。

启动一个具有8GB RAM和一些额外配置的集群：

minikube start \
  --memory 8096 \
  --extra-config=controller-manager.horizontal-pod-autoscaler-upscale-delay=1m \
  --extra-config=controller-manager.horizontal-pod-autoscaler-downscale-delay=2m \
  --extra-config=controller-manager.horizontal-pod-autoscaler-sync-period=10s

请注意，如果您使用的是预先存在的minikube实例，则可以通过销毁VM来重新调整VM的大小。只是添加--memory 8096将不会有任何影响。

验证安装是否成功。您应该看到列为表的一些资源。集群已经准备就绪，也许您应该立即开始部​​署？

还没。

你必须先装好你的东西。

部署到Kubernetes的应用程序必须打包为容器。毕竟，Kubernetes是一个容器协调器，所以它本身无法运行你的jar。容器类似于fat jar：它们包含运行应用程序所需的所有依赖项。甚至JVM也是容器的一部分。所以他们在技术上是一个更胖的fat-jar。

Docker

将应用程序打包为容器的流行技术是Docker。如果您没有安装Docker，可以按照 Docker官方网站上的说明进行操作 。

直接在minikube创建容器图像。

先，按照此命令打印的说明连接Docker客户端到minikube：

minikube docker-env

请注意，如果切换终端，则需要在minikube重新连接到内部的Docker守护程序。每次使用不同的终端时都应遵循相同的说明。

从项目的根目录构建容器图像：

docker build -t spring-k8s-hpa .

可以验证镜像是否已构建并准备好运行：

docker images | grep spring

群集已准备好，您打包应用程序，可以要求Kubernetes部署应用程序。

部署到Kubernetes

您的应用程序有三个组件：

• 呈现前端的Spring Boot应用程序
• ActiveMQ作为消息代理
• 处理事务的Spring Boot后端

您应该分别部署这三个组件。

对于他们每个人你应该创建：

• 一个部署描述什么容器部署和配置对象
• 一个Service对象，充当Deployment部署创建的应用程序的所有实例的负载均衡器

部署中的每个应用程序实例都称为Pod。

部署ActiveMQ

让我们从ActiveMQ开始吧。

您应该创建一个activemq-deployment.yaml包含以下内容的文件：

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: queue
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: queue
    spec:
      containers:
      - name: web
        image: webcenter/activemq:5.14.3
        imagePullPolicy: IfNotPresent
        ports:
          - containerPort: 61616
        resources:
          limits:
            memory: 512Mi

该模板冗长但直接易读：

• 您从名为 webcenter / activemq 的官方注册表中请求了一个activemq容器
• 容器在端口61616上公开消息代理
• 为容器分配了512MB的内存
• 您要求提供单个副本 - 您的应用程序的单个实例 

创建一个activemq-service.yaml ：

apiVersion: v1
kind: Service
metadata:
  name: queue
spec:
  ports:
  - port: 61616 
    targetPort: 61616
  selector:
    app: queue

幸运的是，这个模板更短！

yaml解读：

• 您创建了一个公开端口61616的负载均衡器
• 传入流量将分发到具有类型标签的所有Pod（请参阅上面的部署） app: queue
• 这targetPort是Pods暴露的端口

您可以使用以下命令创建资源：

kubectl create -f activemq-deployment.yaml
kubectl create -f activemq-service.yaml

您可以使用以下命令验证数据库的一个实例是否正在运行：

kubectl get pods -l=app=queue

部署前端

创建fe-deployment.yaml：

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: frontend
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: frontend
    spec:
      containers:
      - name: frontend
        image: spring-boot-hpa
        imagePullPolicy: IfNotPresent
        env:
        - name: ACTIVEMQ_BROKER_URL
          value: "tcp://queue:61616"
        - name: STORE_ENABLED
          value: "true"
        - name: WORKER_ENABLED
          value: "false"
        ports:
        - containerPort: 8080
        livenessProbe:
          initialDelaySeconds: 5
          periodSeconds: 5
          httpGet:
            path: /health
            port: 8080
        resources:
          limits:
            memory: 512Mi

署看起来很像以前的一个。

但是有一些新的字段：

• 有一个section 可以注入环境变量
• 还有liveness probe探针，可以告诉您应用程序何时可以接受流量 

创建fe-service.yaml：

apiVersion: v1
kind: Service
metadata:
  name: frontend
spec:
  ports:
  - nodePort: 32000
    port: 80
    targetPort: 8080
  selector:
    app: frontend
  type: NodePort

使用下面命令创建k8s资源：

kubectl create -f fe-deployment.yaml
kubectl create -f fe-service.yaml

可以使用以下命令验证前端应用程序的一个实例是否正在运行：

kubectl get pods -l=app=frontend

部署后端

创建backend-deployment.yaml：

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: backend
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: backend
      annotations:
        prometheus.io/scrape: 'true'
    spec:
      containers:
      - name: backend
        image: spring-boot-hpa
        imagePullPolicy: IfNotPresent
        env:
        - name: ACTIVEMQ_BROKER_URL
          value: "tcp://queue:61616"
        - name: STORE_ENABLED
          value: "false"
        - name: WORKER_ENABLED
          value: "true"
        ports:
        - containerPort: 8080
        livenessProbe:
          initialDelaySeconds: 5
          periodSeconds: 5
          httpGet:
            path: /health
            port: 8080
        resources:
          limits:
            memory: 512Mi

创建backend-service.yaml：

apiVersion: v1
kind: Service
metadata:
  name: backend
  spec:
    ports:
    - nodePort: 31000
      port: 80
      targetPort: 8080
    selector:
      app: backend
    type: NodePort

创建者两个资源：

kubectl create -f backend-deployment.yaml
kubectl create -f backend-service.yaml

可以验证后端的一个实例是否正在运行：

kubectl get pods -l=app=backend

部署完成。

它真的有效吗？

您可以使用以下命令启动，然后在浏览器中访问该应用程序：

minikube service backend

minikube service frontend

手动扩展以满足不断增长的需求

单个工作程序可能无法处理大量消息。实际上，它当时只能处理一条消息。

如果您决定购买数千件物品，则需要数小时才能清除队列。

此时您有两个选择：

• 你可以手动放大和缩小
• 您可以创建自动缩放规则以自动向上或向下扩展

让我们先从基础知识开始。

您可以使用以下方法将后端扩展为三个实例：

kubectl scale --replicas=5 deployment/backend

可以验证Kubernetes是否创建了另外五个实例：

kubectl get pods

应用程序可以处理五倍以上的消息。

一旦消息队列排空，您可以缩小：

kubectl scale --replicas=1 deployment/backend

如果您知道最多的流量何时达到您的服务，手动扩大和缩小都很棒。

如果不这样做，设置自动缩放器允许应用程序自动缩放而无需手动干预。

您只需要定义一些规则。

公开应用程序指标

Kubernetes如何知道何时扩展您的应用？

很简单，你必须告诉它。

自动调节器通过监控指标来工作。只有这样，它才能增加或减少应用程序的实例。

因此，您可以将队列长度公开为度量标准，并要求autoscaler观察该值。队列中的待处理消息越多，Kubernetes将创建的应用程序实例就越多。

那么你如何公开这些指标呢？

应用程序有一个/metrics端点，用于公开队列中的消息数。如果您尝试访问该页面，您会注意到以下内容：

# HELP messages Number of messages in the queue
# TYPE messages gauge
messages 0

应用程序不会将指标公开为JSON格式。格式为纯文本，是公开 Prometheus指标的标准 。不要担心记忆格式。大多数情况下，您将使用其中一个 Prometheus客户端库 。

在Kubernetes中使用应用程序指标

已准备好进行自动缩放 - 但您应首先安装度量服务器。实际上，默认情况下，Kubernetes不会从您的应用程序中提取指标。如果您愿意，可以启用 Custom Metrics API 。

要安装Custom Metrics API，您还需要 Prometheus  - 时间序列数据库。安装Custom Metrics API所需的所有文件都可以方便地打包在 learnk8s / spring-boot-k8s-hpa中 。

应下载该存储库的内容，并将当前目录更改monitoring为该项目的文件夹。

cd spring-boot-k8s-hpa/monitoring

可以创建自定义指标API：

kubectl create -f ./metrics-server
kubectl create -f ./namespaces.yaml
kubectl create -f ./prometheus
kubectl create -f ./custom-metrics-api

应该等到以下命令返回自定义指标列表：

kubectl get --raw "/apis/custom.metrics.k8s.io/v1beta1" | jq .

任务完成！

您已准备好使用指标。

实际上，您应该已经找到了队列中消息数量的自定义指标：

kubectl get --raw "/apis/custom.metrics.k8s.io/v1beta1/namespaces/default/pods/*/messages" | jq .

恭喜，您有一个公开指标的应用程序和使用它们的指标服务器。

您最终可以启用自动缩放！

在Kubernetes中进行自动扩展部署

Kubernetes有一个名为 Horizo​​ntal Pod Autoscaler 的对象，用于监视部署并上下调整 Pod 的数量。

您将需要其中一个来自动扩展实例。

您应该创建一个hpa.yaml包含以下内容的文件：

apiVersion: autoscaling/v2beta1
kind: HorizontalPodAutoscaler
metadata:
  name: spring-boot-hpa
spec:
  scaleTargetRef:
    apiVersion: extensions/v1beta1
    kind: Deployment
    name: backend 
  minReplicas: 1
  maxReplicas: 10
  metrics:
  - type: Pods
    pods:
      metricName: messages
      targetAverageValue: 10

这个文件很神秘，所以让我为你解释一下：

• 在scaleTargetRef指定Kubernetes监视的部署。在这种情况下，它是前面的worker部署。
• 使用messages指标metrics来扩展您的Pod。当队列中有超过十条消息时，Kubernetes将触发自动扩展。
• 至少，部署应该有两个Pod。10 个Pods是上限。

创建资源：

kubectl create -f hpa.yaml

提交这个自动缩放器autoscaler后，您应该注意到后端的副本数量是两个。这是有道理的，因为您要求自动缩放器始终至少运行两个副本。

您可以检查触发自动缩放器的条件以及由此产生的事件：

kubectl describe hpa

自动缩放器autoscaler能够将Pod数量扩展到2，并且它已准备好监视部署。

令人兴奋的东西，但它有效吗？

负载测试

只有一种方法可以知道它是否有效：在队列中创建大量消息。

转到前端应用程序并开始添加大量消息。在添加消息时，使用以下方法监视Horizo​​ntal Pod Autoscaler的状态：

kubectl describe hpa

Pod的数量从2上升到4，然后是8，最后是10。

该应用程序随消息数量而变化！欢呼！

您刚刚部署了一个完全可伸缩的应用程序，可根据队列中的待处理消息数进行扩展。

另外，缩放算法如下：

MAX(CURRENT_REPLICAS_LENGTH * 2, 4)

在解释算法时，文档没有多大帮助。您可以 在代码中找到详细信息 。

此外，每分钟都会重新评估每个放大，而每两分钟缩小一次。

以上所有都是可以调整的设置。

但是你还没有完成。

什么比自动缩放实例更好？自动缩放集群。​​​​​​​

跨节点缩放Pod非常有效。但是，如果群集中没有足够的容量来扩展Pod，该怎么办？如果达到峰值容量，Kubernetes将使Pods处于暂挂状态并等待更多资源可用。如果您可以使用类似于横向Pod Autoscaler的自动缩放器，对于节点则会很棒。

您可以拥有一个集群自动缩放器，可以在您需要更多资源时为Kubernetes集群添加更多节点。集群自动缩放器具有不同的形式和大小。它也是由云提供商指定的。

请注意，使用minikube您将无法使用自动缩放器进行测试，因为它根据定义是单节点。

您可以在Github上找到 有关集群自动调节器 和 云提供程序实现的 [url=https://github.com/kubernetes/autoscaler/tree/master/cluster-autoscaler#cluster-autoscaler]更多信息[/url]。

概括

大规模设计应用程序需要仔细规划和测试。

基于队列的体系结构是一种出色的设计模式，可以解耦您的微服务并确保它们可以独立扩展和部署。

虽然您可以部署部署脚本，但可以更轻松地利用容器协调器（如Kubernetes）自动部署和扩展应用程序。​​​​​​​