---
title: 使用工作队列进行粗粒度并行处理
content_type: task
weight: 20
---

<!--
title: Coarse Parallel Processing Using a Work Queue
content_type: task
weight: 20
-->

<!-- overview -->

<!--
In this example, you will run a Kubernetes Job with multiple parallel
worker processes.

In this example, as each pod is created, it picks up one unit of work
from a task queue, completes it, deletes it from the queue, and exits.

Here is an overview of the steps in this example:

1. **Start a message queue service.**  In this example, you use RabbitMQ, but you could use another
   one.  In practice you would set up a message queue service once and reuse it for many jobs.
1. **Create a queue, and fill it with messages.**  Each message represents one task to be done.  In
   this example, a message is an integer that we will do a lengthy computation on.
1. **Start a Job that works on tasks from the queue**.  The Job starts several pods.  Each pod takes
   one task from the message queue, processes it, and exits.
-->
本例中，你将会运行包含多个并行工作进程的 Kubernetes Job。

本例中，每个 Pod 一旦被创建，会立即从任务队列中取走一个工作单元并完成它，然后将工作单元从队列中删除后再退出。

下面是本次示例的主要步骤：

1. **启动一个消息队列服务**。
   本例中，我们使用 RabbitMQ，你也可以用其他的消息队列服务。
   在实际工作环境中，你可以创建一次消息队列服务然后在多个任务中重复使用。

1. **创建一个队列，放上消息数据**。
   每个消息表示一个要执行的任务。本例中，每个消息是一个整数值。
   我们将基于这个整数值执行很长的计算操作。

1. **启动一个在队列中执行这些任务的 Job**。
   该 Job 启动多个 Pod。每个 Pod 从消息队列中取走一个任务，处理任务，然后退出。

## {{% heading "prerequisites" %}}


<!--
You should already be familiar with the basic,
non-parallel, use of [Job](/docs/concepts/workloads/controllers/job/).
-->

你应当熟悉 Job 的基本用法（非并行的），请参考
[Job](/zh-cn/docs/concepts/workloads/controllers/job/)。

{{< include "task-tutorial-prereqs.md" >}} 

<!--
You will need a container image registry where you can upload images to run in your cluster.

This task example also assumes that you have Docker installed locally.
-->
你需要一个容器镜像仓库，用来向其中上传镜像以在集群中运行。

此任务示例还假设你已在本地安装了 Docker。

<!-- steps -->

<!--
## Starting a message queue service

This example uses RabbitMQ, however, you can adapt the example to use another AMQP-type message service.

In practice you could set up a message queue service once in a
cluster and reuse it for many jobs, as well as for long-running services.

Start RabbitMQ as follows:
-->
## 启动消息队列服务   {#starting-a-message-queue-service}

本例使用了 RabbitMQ，但你可以更改该示例，使用其他 AMQP 类型的消息服务。

在实际工作中，在集群中一次性部署某个消息队列服务，之后在很多 Job 中复用，包括需要长期运行的服务。

按下面的方法启动 RabbitMQ：

<!--
# make a Service for the StatefulSet to use
-->
```shell
# 为 StatefulSet 创建一个 Service 来使用
kubectl create -f https://kubernetes.io/examples/application/job/rabbitmq/rabbitmq-service.yaml
```
```
service "rabbitmq-service" created
```

```shell
kubectl create -f https://kubernetes.io/examples/application/job/rabbitmq/rabbitmq-statefulset.yaml
```

```
statefulset "rabbitmq" created
```

<!--
## Testing the message queue service

Now, we can experiment with accessing the message queue.  We will
create a temporary interactive pod, install some tools on it,
and experiment with queues.

First create a temporary interactive Pod.
-->
## 测试消息队列服务   {#testing-the-message-queue-service}

现在，我们可以试着访问消息队列。我们将会创建一个临时的可交互的 Pod，
在它上面安装一些工具，然后用队列做实验。

首先创建一个临时的可交互的 Pod：

```shell
# 创建一个临时的可交互的 Pod
kubectl run -i --tty temp --image ubuntu:22.04
```
```
Waiting for pod default/temp-loe07 to be running, status is Pending, pod ready: false
... [ previous line repeats several times .. hit return when it stops ] ...
```

<!--
Note that your pod name and command prompt will be different.

Next install the `amqp-tools` so you can work with message queues.
The next commands show what you need to run inside the interactive shell in that Pod:
-->
请注意你的 Pod 名称和命令提示符将会不同。

接下来安装 `amqp-tools`，这样你就能用消息队列了。
下面是在该 Pod 的交互式 shell 中需要运行的命令：

```shell
apt-get update && apt-get install -y curl ca-certificates amqp-tools python3 dnsutils
```

<!--
Later, you will make a container image that includes these packages.

Next, you will check that you can discover the Service for RabbitMQ:
-->
后续，你将制作一个包含这些包的容器镜像。

接着，你将要验证可以发现 RabbitMQ 服务：

<!--
# Run these commands inside the Pod
# Note the rabbitmq-service has a DNS name, provided by Kubernetes:
-->
```
# 在 Pod 内运行这些命令
# 请注意 rabbitmq-service 拥有一个由 Kubernetes 提供的 DNS 名称：

nslookup rabbitmq-service
```
```
Server:        10.0.0.10
Address:    10.0.0.10#53

Name:    rabbitmq-service.default.svc.cluster.local
Address: 10.0.147.152
```
<!--
(the IP addresses will vary)
-->
（IP 地址会有所不同）

<!--
If the kube-dns addon is not set up correctly, the previous step may not work for you.
You can also find the IP address for that Service in an environment variable:
-->
如果 kube-dns 插件没有正确安装，上一步可能会出错。
你也可以在环境变量中找到该服务的 IP 地址。

<!--
# run this check inside the Pod
-->
```shell
# 在 Pod 内运行此检查
env | grep RABBITMQ_SERVICE | grep HOST
```
```
RABBITMQ_SERVICE_SERVICE_HOST=10.0.147.152
```
<!--
(the IP address will vary)
-->
（IP 地址会有所不同）

<!--
Next you will verify that you can create a queue, and publish and consume messages.
-->
接下来，你将验证是否可以创建队列以及发布和使用消息。

<!--
# Run these commands inside the Pod
# In the next line, rabbitmq-service is the hostname where the rabbitmq-service
# can be reached.  5672 is the standard port for rabbitmq.

# If you could not resolve "rabbitmq-service" in the previous step,
# then use this command instead:
# Now create a queue:
-->

```shell
# 在 Pod 内运行这些命令
# 下一行，rabbitmq-service 是访问 rabbitmq-service 的主机名。5672是 rabbitmq 的标准端口。

export BROKER_URL=amqp://guest:guest@rabbitmq-service:5672

# 如果上一步中你不能解析 "rabbitmq-service"，可以用下面的命令替换：
BROKER_URL=amqp://guest:guest@$RABBITMQ_SERVICE_SERVICE_HOST:5672

# 现在创建队列：

/usr/bin/amqp-declare-queue --url=$BROKER_URL -q foo -d foo
```
```
foo
```

<!--
Publish one message to the queue:
# And get it back.
-->
向队列推送一条消息：

```shell
/usr/bin/amqp-publish --url=$BROKER_URL -r foo -p -b Hello

# 然后取回它：

/usr/bin/amqp-consume --url=$BROKER_URL -q foo -c 1 cat && echo
```
```
Hello
```

<!--
In the last command, the `amqp-consume` tool took one message (`-c 1`)
from the queue, and passes that message to the standard input of an arbitrary command.
In this case, the program `cat` prints out the characters read from standard input, and
the echo adds a carriage return so the example is readable.
-->

最后一个命令中，`amqp-consume` 工具从队列中取走了一个消息，并把该消息传递给了随机命令的标准输出。
在这种情况下，`cat` 会打印它从标准输入中读取的字符，echo 会添加回车符以便示例可读。

<!--
## Fill the Queue with tasks

Now, fill the queue with some simulated tasks.  In this example, the tasks are strings to be
printed.

In a practice, the content of the messages might be:

- names of files to that need to be processed
- extra flags to the program
- ranges of keys in a database table
- configuration parameters to a simulation
- frame numbers of a scene to be rendered
-->
## 为队列增加任务  {#filling-the-queue-with-tasks}

现在用一些模拟任务填充队列。在此示例中，任务是多个待打印的字符串。

实践中，消息的内容可以是：

- 待处理的文件名
- 程序额外的参数
- 数据库表的关键字范围
- 模拟任务的配置参数
- 待渲染的场景的帧序列号

<!--
If there is large data that is needed in a read-only mode by all pods
of the Job, you typically put that in a shared file system like NFS and mount
that readonly on all the pods, or write the program in the pod so that it can natively read
data from a cluster file system (for example: HDFS).

For this example, you will create the queue and fill it using the AMQP command line tools.
In practice, you might write a program to fill the queue using an AMQP client library.
-->

如果有大量的数据需要被 Job 的所有 Pod 读取，典型的做法是把它们放在一个共享文件系统中，
如 NFS（Network File System 网络文件系统），并以只读的方式挂载到所有 Pod，或者 Pod 中的程序从类似 HDFS
（Hadoop Distributed File System 分布式文件系统）的集群文件系统中读取。

例如，你将创建队列并使用 AMQP 命令行工具向队列中填充消息。实践中，你可以写个程序来利用 AMQP 客户端库来填充这些队列。

<!--
# Run this on your computer, not in the Pod
-->
```shell
# 在你的计算机上运行此命令，而不是在 Pod 中
/usr/bin/amqp-declare-queue --url=$BROKER_URL -q job1  -d
```
```
job1
```
<!--
Add items to the queue:
-->
将这几项添加到队列中：

```shell
for f in apple banana cherry date fig grape lemon melon
do
  /usr/bin/amqp-publish --url=$BROKER_URL -r job1 -p -b $f
done
```

<!--
You added 8 messages to the queue.

## Create a container Image

Now you are ready to create an image that you will run as a Job.

The job will use the `amqp-consume` utility to read the message
from the queue and run the actual work.  Here is a very simple
example program:
-->
你给队列中填充了 8 个消息。

## 创建容器镜像   {#create-a-container-image}

现在你可以创建一个做为 Job 来运行的镜像。

这个 Job 将用 `amqp-consume` 实用程序从队列中读取消息并进行实际工作。
这里给出一个非常简单的示例程序：

{{% code_sample language="python" file="application/job/rabbitmq/worker.py" %}}

<!--
Give the script execution permission:
-->
赋予脚本执行权限：

```shell
chmod +x worker.py
```


<!--
Now, build an image. Make a temporary directory, change to it,
download the [Dockerfile](/examples/application/job/rabbitmq/Dockerfile),
and [worker.py](/examples/application/job/rabbitmq/worker.py).  In either case,
build the image with this command:
-->

现在，编译镜像。创建一个临时目录，切换到这个目录。下载
[Dockerfile](/examples/application/job/rabbitmq/Dockerfile) 和
[worker.py](/examples/application/job/rabbitmq/worker.py)。
无论哪种情况，都可以用下面的命令编译镜像：

```shell
docker build -t job-wq-1 .
```

<!--
For the [Docker Hub](https://hub.docker.com/), tag your app image with
your username and push to the Hub with the below commands. Replace
`<username>` with your Hub username.
-->
对于 [Docker Hub](https://hub.docker.com/), 给你的应用镜像打上标签，
标签为你的用户名，然后用下面的命令推送到 Hub。用你的 Hub 用户名替换 `<username>`。 

```shell
docker tag job-wq-1 <username>/job-wq-1
docker push <username>/job-wq-1
```

<!--
If you are using an alternative container image registry, tag the
image and push it there instead.
-->
如果你使用替代的镜像仓库，请标记该镜像并将其推送到那里。

<!--
## Defining a Job

Here is a job definition.  You'll need to make a copy of the Job and edit the
image to match the name you used, and call it `./job.yaml`.
Here is a manifest for a Job.  You'll need to make a copy of the Job manifest
(call it `./job.yaml`),
and edit the name of the container image to match the name you used.
-->
## 定义 Job   {#defining-a-job}

这里给出一个 Job 的清单。你需要复制一份 Job 清单的副本（将其命名为 `./job.yaml`），
并编辑容器镜像的名称以匹配使用的名称。

{{% code_sample file="application/job/rabbitmq/job.yaml" %}}

<!--
In this example, each pod works on one item from the queue and then exits.
So, the completion count of the Job corresponds to the number of work items
done. That is why the example manifest has `.spec.completions` set to `8`.

## Running the Job

Now, run the Job:
-->
本例中，每个 Pod 使用队列中的一个消息然后退出。
这样，Job 的完成计数就代表了完成的工作项的数量。
这就是示例清单将 `.spec.completions` 设置为 `8` 的原因。

## 运行 Job   {#running-the-job}

运行 Job：

<!--
# this assumes you downloaded and then edited the manifest already
-->
# 这假设你已经下载并编辑了清单
```shell
kubectl apply -f ./job.yaml
```

<!--
You can wait for the Job to succeed, with a timeout:
-->
你可以等待 Job 在某个超时时间后成功：

```shell
# 状况名称的检查不区分大小写
kubectl wait --for=condition=complete --timeout=300s job/job-wq-1
```

<!--
Next, check on the Job:
-->
接下来查看 Job：

```shell
kubectl describe jobs/job-wq-1
```
```
Name:             job-wq-1
Namespace:        default
Selector:         controller-uid=41d75705-92df-11e7-b85e-fa163ee3c11f
Labels:           controller-uid=41d75705-92df-11e7-b85e-fa163ee3c11f
                  job-name=job-wq-1
Annotations:      <none>
Parallelism:      2
Completions:      8
Start Time:       Wed, 06 Sep 2022 16:42:02 +0000
Pods Statuses:    0 Running / 8 Succeeded / 0 Failed
Pod Template:
  Labels:       controller-uid=41d75705-92df-11e7-b85e-fa163ee3c11f
                job-name=job-wq-1
  Containers:
   c:
    Image:      container-registry.example/causal-jigsaw-637/job-wq-1
    Port:
    Environment:
      BROKER_URL:       amqp://guest:guest@rabbitmq-service:5672
      QUEUE:            job1
    Mounts:             <none>
  Volumes:              <none>
Events:
  FirstSeen  LastSeen   Count    From    SubobjectPath    Type      Reason              Message
  ─────────  ────────   ─────    ────    ─────────────    ──────    ──────              ───────
  27s        27s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-hcobb
  27s        27s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-weytj
  27s        27s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-qaam5
  27s        27s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-b67sr
  26s        26s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-xe5hj
  15s        15s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-w2zqe
  14s        14s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-d6ppa
  14s        14s        1        {job }                   Normal    SuccessfulCreate    Created pod: job-wq-1-p17e0
```

<!--
All the pods for that Job succeeded! You're done.
-->
该 Job 的所有 Pod 都已成功！你完成了。

<!-- discussion -->

<!--
## Alternatives

This approach has the advantage that you do not need to modify your "worker" program to be
aware that there is a work queue.  You can include the worker program unmodified in your container
image.

Using this approach does require that you run a message queue service.
If running a queue service is inconvenient, you may
want to consider one of the other [job patterns](/docs/concepts/workloads/controllers/job/#job-patterns).
-->
## 替代方案   {#alternatives}

本文所讲述的处理方法的好处是你不需要修改你的 "worker" 程序使其知道工作队列的存在。
你可以将未修改的工作程序包含在容器镜像中。

使用此方法需要你运行消息队列服务。如果不方便运行消息队列服务，
你也许会考虑另外一种[任务模式](/zh-cn/docs/concepts/workloads/controllers/job/#job-patterns)。

<!--
This approach creates a pod for every work item.  If your work items only take a few seconds,
though, creating a Pod for every work item may add a lot of overhead.  Consider another
design, such as in the [fine parallel work queue example](/docs/tasks/job/fine-parallel-processing-work-queue/), that executes multiple work items per Pod.

In this example, you use the `amqp-consume` utility to read the message
from the queue and run the actual program.  This has the advantage that you
do not need to modify your program to be aware of the queue.
A [fine parallel work queue example](/docs/tasks/job/fine-parallel-processing-work-queue/)
shows how to communicate with the work queue using a client library.
-->
本文所述的方法为每个工作项创建了一个 Pod。
如果你的工作项仅需数秒钟，为每个工作项创建 Pod 会增加很多的常规消耗。
考虑另一种设计，例如[精细并行工作队列示例](/zh-cn/docs/tasks/job/fine-parallel-processing-work-queue/)，
这种方案可以实现每个 Pod 执行多个工作项。

示例中，你使用了 `amqp-consume` 从消息队列读取消息并执行真正的程序。
这样的好处是你不需要修改你的程序使其知道队列的存在。
要了解怎样使用客户端库和工作队列通信，
请参考[精细并行工作队列示例](/zh-cn/docs/tasks/job/fine-parallel-processing-work-queue/)。

<!--
## Caveats

If the number of completions is set to less than the number of items in the queue, then
not all items will be processed.

If the number of completions is set to more than the number of items in the queue,
then the Job will not appear to be completed, even though all items in the queue
have been processed.  It will start additional pods which will block waiting
for a message.
You would need to make your own mechanism to spot when there is work
to do and measure the size of the queue, setting the number of completions to match.

There is an unlikely race with this pattern.  If the container is killed in between the time
that the message is acknowledged by the `amqp-consume` command and the time that the container
exits with success, or if the node crashes before the kubelet is able to post the success of the pod
back to the API server, then the Job will not appear to be complete, even though all items
in the queue have been processed.
-->
## 友情提醒   {#caveats}

如果设置的完成数量小于队列中的消息数量，会导致一部分消息项不会被执行。

如果设置的完成数量大于队列中的消息数量，当队列中所有的消息都处理完成后，
Job 也会显示为未完成。Job 将创建 Pod 并阻塞等待消息输入。
你需要建立自己的机制来发现何时有工作要做，并测量队列的大小，设置匹配的完成数量。

当发生下面两种情况时，即使队列中所有的消息都处理完了，Job 也不会显示为完成状态：
* 在 `amqp-consume` 命令拿到消息和容器成功退出之间的时间段内，执行杀死容器操作；
* 在 kubelet 向 API 服务器传回 Pod 成功运行之前，发生节点崩溃。