## 简介

人工智能是当今开发者社区中最热门的话题之一，它为我们提供了许多新的可能性和挑战。然而，要想充分利用 `AI` 的潜力，我们需要有合适的工具和平台，让我们能够轻松地运行 `AI` 模型，而不必担心复杂的基础设施和隐私问题。

`Cloudflare` 是一家知名的云计算公司，它提供了一个全球性的网络，可以为网站和应用程序提供安全、快速和可靠的服务。`Cloudflare` 还提供了一个开发者平台，让开发者可以在其网络上构建和部署各种应用程序，无论是静态网站、动态内容、边缘计算，还是最新推出的 `Workers AI`。

`Workers AI` 是 `Cloudflare` 的无服务器 `GPU` 驱动的推理平台，它让开发者可以在 `Cloudflare` 的网络上，从自己的代码中运行 `AI` 模型，无论是从 `Workers`、`Pages`，还是通过 `REST API` 从任何地方。`Workers AI` 的设计理念是为开发者提供最佳的体验，让他们可以用几行代码就启动一个 AI 应用程序，而不必了解机器学习的细节。

`Workers AI` 提供了一系列经过精选的流行的开源模型，涵盖了多种推理任务：

> - **文本生成 （大语言模型）：** meta/llama-2-7b-chat-int8
> - **自动语音识别(ASR)：** openai/whisper
> - **翻译：** meta/m2m100-1.2
> - **文本分类：** huggingface/distilbert-sst-2-int8
> - **图像分类：** microsoft/resnet-50
> - **嵌入：** baai/bge-base-en-v1.5

这些模型都可以在 `Cloudflare` 的全球 `GPU` 网络上运行，为用户提供低延迟和高性能的服务。此外，`Workers AI` 还提供了一个向量数据库，叫做 `Vectorize`，可以让开发者进行语义搜索、推荐、异常检测等任务，或者为大型语言模型提供上下文和记忆。

在本文中，我们将介绍如何使用 `Workers AI` 构建一个简单的 `AI` 应用程序，它可以根据用户的输入生成一张图片。

## 开发

### 1. 创建一个 Workers 项目

运行以下命令创建一个名为 `text-to-image` 的新项目：

```bash
$ npm create cloudflare@latest
```

在设置 `text-to-image worker` 时，按照以下方式回答设置问题：

- 输入**text-to-image** 作为应用的名称
- 选择 **Hello World** 脚本作为应用的类型
- 对于使用 `TypeScript`，选择 **yes**
- 对于使用 `Git`，选择 **yes**
- 对于部署，选择 **no**

最后，导航到新应用的目录：

```bash
$ cd workers-ai
```

### 2. 将 Workers AI 连接到您的 worker

创建一个 `Workers AI` `binding`，使您的 `Worker` 可以访问 `Workers AI` 服务，而无需自己管理 `API` 密钥。

为了将 `Workers AI` 绑定到您的 `Worker`，请将以下内容添加到您的 **wrangler.toml** 文件末尾：

```toml
[ai]
binding = "AI" #available in your worker via env.AI
```

### 3. 安装 Workers AI 客户端库

```bash
$ npm install @cloudflare/ai
```

### 4. 在您的 worker 中运行指定模型

使用如下代码更新 **source/index.ts**：

```javascript
import { Ai } from './vendor/@cloudflare/ai.js';


export default {
  async fetch(request, env) {
    const ai = new Ai(env.AI);


    const inputs = {
      prompt: 'cyberpunk cat'
    };


    const response = await ai.run(
      '@cf/stabilityai/stable-diffusion-xl-base-1.0',
      inputs
    );


    return new Response(response, {
      headers: {
        'content-type': 'image/png'
      }
    });
  }
};


```

### 5. 使用 Wrangler 在本地开发

在项目目录中，运行以下命令测试 Workers AI:

```
$ npx wrangler dev --remote
```

**注：** 这些模型目前只运行在 `Cloudflare` 的 `GPU` 网络上（而不是本地），所以必须如上设置 `--remote`，此时您会被提示登录。

`Wrangler` 会给您一个 `URL`（通常是 `localhost:8787`）。访问该 `URL`，大约等待20~30秒你将会看到生成的图片。

### 6. 部署您的 worker

最后，部署 `worker`，使您的项目可以在互联网上访问。

```
$ npx wrangler deploy
# Outputs: https://workers-ai.<YOUR_SUBDOMAIN>.workers.dev
```

以下是两个在线示例：

* [文生图](https://trumanwl.com/image/text-to-image)
* [在线翻译](https://trumanwl.com/text/translate)

## 结语

`Workers AI` 是一个强大的工具，可以让开发者轻松地构建和部署各种 `AI` 应用程序，而不必担心复杂的基础设施和隐私问题。如果您想要了解更多关于 `Workers AI` 的信息，您可以访问 [Workers AI 的官方网站](https://dash.cloudflare.com/)，或者查看 [Workers AI 的文档](https://developers.cloudflare.com/workers-ai/)。

如何使用 Workers AI 构建强大的 AI 应用程序

当您创建表并意识到自己犯了一个错误，或者应用程序的要求发生了变化时，您可以删除该表并重新创建它。但是，如果表中已经有了数据，或者该表被其他数据库对象（例如外键约束）引用，那么这不是一个方便的选择。因此，`PostgreSQL`提供了一组命令来修改现有表。

## 添加列

```sql
ALTER TABLE products ADD COLUMN description text;
```

你也可以在添加列的时候定义约束：

```sql
ALTER TABLE products ADD COLUMN description text CHECK (description <> '');
```

## 删除列 

```
ALTER TABLE products DROP COLUMN description;
```

列中的任何数据都会消失。涉及该列的表约束也会被删除。但是，如果该列被另一个表的外键约束引用，`PostgreSQL`不会默默地删除该约束。您可以通过添加`CASCADE`来授权删除依赖于该列的所有内容：

```sql
ALTER TABLE products DROP COLUMN description CASCADE;
```

## 添加约束

```
ALTER TABLE products ADD CHECK (name <> '');
ALTER TABLE products ADD CONSTRAINT some_name UNIQUE (product_no);
ALTER TABLE products ADD FOREIGN KEY (product_group_id) REFERENCES product_groups;
```

添加非空约束

```
ALTER TABLE products ALTER COLUMN product_no SET NOT NULL;
```

该约束会被立即检查，因此表数据必须满足约束才可以添加。

## 删除约束 

要删除约束，您需要知道它的名称。如果你忘记了约束的名称，可以通过`psql`的`\d tablename`命令或其他方式查看。

```
ALTER TABLE products DROP CONSTRAINT some_name;
```

***与删除列一样，如果您要删除其他内容所依赖的约束，则需要添加`CASCADE`。例如，外键约束依赖于引用列上的唯一或主键约束。***

这对于除非空约束之外的所有约束类型都适用。要删除非空约束，请使用：:

```
ALTER TABLE products ALTER COLUMN product_no DROP NOT NULL;
```

## 更改列的默认值 

```
ALTER TABLE products ALTER COLUMN price SET DEFAULT 7.77;
```

请注意，这不会影响表中的任何现有行，它只会更改以后执行`INSERT`命令时的默认值。

删除默认值:

```
ALTER TABLE products ALTER COLUMN price DROP DEFAULT;
```

这实际上与将默认值设置为 `null` 相同。

## 更改列的数据类型 

```
ALTER TABLE products ALTER COLUMN price TYPE numeric(10,2);
```

只有被更改的列中的所有值都可以通过隐式转换转换为新类型时，此操作才会成功。

`PostgreSQL`将尝试将列的默认值（如果有）以及涉及该列的任何约束转换为新类型。但这些转换可能会失败，或可能产生意想不到的结果。通常最好在更改列类型之前删除列上的所有约束，然后在之后添加适当修改的约束。

## 重命名列

```
ALTER TABLE products RENAME COLUMN product_no TO product_number;
```

## 重命名表 

```
ALTER TABLE products RENAME TO items;
```

PostgreSQL修改表结构

首先查看配置文件位置：

```shell
[postgres@localhost ~]$ psql
psql (16.3)
Type "help" for help.

postgres=# show config_file;
              config_file               
----------------------------------------
 /var/lib/pgsql/16/data/postgresql.conf
(1 row)
```

然后修改`/var/lib/pgsql/16/data/postgresql.conf`，修改`listen_addresses`配置：

```
listen_addresses = '*'					# what IP address(es) to listen on;
                                        # comma-separated list of addresses;
                                        # defaults to 'localhost'; use '*' for all
                                        # (change requires restart)
```

接下来修改`/var/lib/pgsql/16/data/pg_hba.conf`，加入以下配置：

```
host    all             all             0.0.0.0/0               scram-sha-256
```

然后重启`postgresql`：

```bash
$ sudo systemctl restart postgresql-16.service
```





PostgreSQL开启远程访问

> *提示：本文基于Kafka 3.7.0写成*

`Kafka`发行版中自带的配置示例可以用来安装单机服务，主要用于概念验证，并不能满足大型集群的配置需求。`Kafka`有很多配置参数，涉及安装和调优的方方面面，其中的大多数参数可以使用默认值，除非你对调优有特别的需求。

## 常规配置参数

一些配置参数在单机安装时可以使用默认值，但部署到其他环境中时要格外小心。这些参数是针对单台服务器最基本的配置，其中大多数参数需要经过修改后才能用在集群中。

### `broker.id`

默认值：-1

每个`broker`都需要有一个整数标识符，该标识符是使用`broker.id`指定的。可以被设置成其他任意整数。这个值在整个`Kafka`集群中必须是唯一的，并且可以在服务器节点间移动。建议把`ID`设置成与主机名具有相关性的整数，这样就可以很容易地将`ID`与主机名映射起来。如果主机名包含唯一性的数字（比如`host1.example.com`、`host2.example.com`等），那么用`1`、`2`这些数字来设置`broker.id`就再好不过了。

### `listeners`

默认值：`PLAINTEXT://:9092`

`listeners`配置参数是一个用逗号分隔的`URI`列表，也就是要监听的地址和端口。如果没有为监听器指定安全协议，则还需要额外配置`listener.security.protocol.map`参数。监听器的格式为`<protocol>://<hostname>:<port>`，例如，`PLAINTEXT://localhost:9092,SSL://:9091`就是一个合法的配置。如果主机名是`0.0.0.0`，那么将绑定所有的网络接口地址。如果主机名为空，那么将绑定默认的网络接口地址。***需要注意的是，如果指定的端口号小于`1024`，则必须用`root`权限启动`Kafka`，但不建议这么配置***。

### `zookeeper.connect`

默认值：`null`

用于保存`broker`元数据的`ZooKeeper`地址是通过`zookeeper.connect`来指定的。示例配置使用了一个运行在`2181`端口上的`ZooKeeper`，所以指定了`localhost:2181`。这个参数的值是用逗号分隔的一组`hostname:port/path`，每一部分的含义如下。

* `hostname`

  `ZooKeeper`服务器的主机名或IP地址。

* `port`

  `ZooKeeper`的客户端连接端口。

* `/path`

  可选的`ZooKeeper`路径，以作为`Kafka`集群的`chroot`。如果不指定，则默认使用根路径。如果指定的`chroot`路径（一般作为应用程序的根目录）不存在，那么`broker`会在启动时创建它。

### `log.dirs`

默认值：`/tmp/kafka-logs`

`Kafka`把所有消息都保存在磁盘上，存放日志片段的目录是通过`log.dir`来指定的。如果有多个目录，则可以用`log.dirs`来指定。如果没有指定这个参数，则默认使用`log.dir`。***`log.dirs`是一组用逗号分隔的本地文件系统路径。如果指定了多条路径，那么`broker`会根据“最少使用”原则，把同一个分区的日志片段保存到同一条路径下。***需要注意的是，`broker`会向分区数量最少的目录新增分区，而不是向可用磁盘空间最小的目录新增分区，所以并不能保证数据会被均匀地分布在多个目录中。

### `num.recovery.threads.per.data.dir`

默认值：1

`Kafka`使用线程池来处理日志片段。目前，线程池被用于以下3种情形：

* 当服务器正常启动时，用于打开每个分区的日志片段。
* 当服务器发生崩溃并重启时，用于检查和截短每个分区的日志片段。
* 当服务器正常关闭时，用于关闭日志片段。

在默认情况下，每个日志目录只使用一个线程。因为这些线程只在服务器启动和关闭时使用，所以可以多设置一些线程来实现并行操作。特别是对包含大量分区的服务器来说，一旦发生崩溃，在从错误中恢复时可以通过并行操作省下数小时的时间。***需要注意的是，这个参数对应的是`log.dirs`中的一个目录，也就是说，如果`num.recovery.threads.per.data.dir`被设为8，并且`log.dirs`指定了3条路径，那么总共需要24个线程***。

### `auto.create.topics.enable`

默认值：`true`

在默认情况下，`Kafka`会在如下几种情形中自动创建主题：

* 在默认情况下，Kafka会在如下几种情形中自动创建主题。
* 在默认情况下，Kafka会在如下几种情形中自动创建主题。
* 在默认情况下，Kafka会在如下几种情形中自动创建主题。

根据`Kafka`协议，如果一个主题不提前被创建，则无法知道它是否存在。如果你选择通过手动的方式或其他配置系统来显式地创建主题，那么可以把`auto.create.topics.enable`设为`false`。

### `auto.leader.rebalance.enable`

默认值：`true`

为了确保主题的所有权不会集中在一台`broker`上，可以将这个参数设置为`true`，让主题的所有权尽可能地在集群中保持均衡。如果启用了这个功能，那么就会有一个后台线程定期检查分区的分布情况（这个时间间隔可以通过`leader.imbalance.check.interval.seconds`来配置）。如果不均衡的所有权超出了`leader.imbalance.per.broker.percentage`指定的百分比，则会启动一次分区首领再均衡。

### `delete.topic.enable`

默认值：`true`

根据应用场景和数据保留策略的不同，你可能希望将集群锁定，以防主题被随意删除。把这个参数设置为`false`就可以禁用主题删除功能。

## 主题的默认配置

`Kafka`为新创建的主题提供了很多默认的配置参数。

### `num.partitions`

默认值：1

`num.partitions`参数指定了新创建的主题将包含多少个分区，特别是如果启用了主题自动创建功能（默认是启用的），那么主题的分区数就是这个参数指定的值。**默认是1个分区**。需要注意的是，***可以增加主题的分区数，但不能减少***。所以，如果要让一个主题的分区数小于`num.partitions`指定的值，则需要手动创建主题。

### `default.replication.factor`

默认值：1

如果启用了自动创建主题功能，那么这个参数的值就是新创建主题的复制系数。

### `log.retention.ms`

默认值：`null`

`Kafka`通常根据配置的时间长短来决定数据可以被保留多久。我们使用`log.retention.hours`参数来配置时间，默认为168小时，也就是1周。除此以外，还有另外两个参数`log.retention.minutes`和`log.retention.ms`。这3个参数的作用是一样的（**都是用于确定消息将在多久以后被删除**），不过还是推荐使用`log.retention.ms`，因为如果指定了不止一个参数，那么`Kafka`会优先使用具有最小单位值的那个。

### `log.retention.bytes`

默认值：-1

另一种数据保留策略是通过计算已保留的消息的字节总数来判断旧消息是否过期。这个字节总数阈值通过参数`log.retention.bytes`来指定，对应的是每一个分区。也就是说，如果一个主题包含8个分区，并且`log.retention.bytes`被设置为1 GB，那么这个主题最多可以保留8 GB的数据。需要注意的是，**所有保留都是针对单个分区而不是主题执行的**。

所以，如果配置了这个参数，那么当主题增加了新分区，整个主题可以保留的数据也会随之增加。***如果这个值被设置为–1，那么分区就可以无限期地保留数据***。

> 如果同时指定了`log.retention.bytes`和`log.retention.ms`（或另一个按时间保留的参数），那么只要任意一个条件得到满足，消息就会被删除。假设`log.retention.ms`被设置为86 400 000（也就是1天），`log.retention.bytes`被设置为1 000 000 000（也就是1 GB），如果消息字节总数不到一天就超过了1 GB，那么旧数据就会被删除。相反，如果消息字节总数小于1 GB，那么一天之后这些消息也会被删除，尽管分区的数据总量小于1 GB。为简单起见，建议只选择其中的一种保留策略，要么基于数据大小，要么基于时间，或者两种都不选择，以防发生意外的数据丢失。不过，对于复杂的场景，可以两种都使用。

### `log.segment.bytes`

默认值：1073741824 (1GB)

前面介绍的参数都作用在日志片段而不是单条消息上。当消息到达`broker`时，它们会被追加到分区的当前日志片段上。当日志片段大小达到`log.segment.bytes`指定的上限（默认是1 GB）时，当前日志片段会被关闭，一个新的日志片段会被打开。一旦日志片段被关闭，就可以开始进入过期倒计时。这个参数的值越小，关闭和分配新文件就会越频繁，从而降低整体的磁盘写入效率。

如果主题的消息量不是很大，那么如何设置这个参数就变得尤为重要。如果一个主题每天只接收100 MB的消息，并且`log.segment.bytes`使用了默认设置，那么填满一个日志片段将需要10天。因为在日志片段被关闭之前消息是不会过期的，所以如果`log.retention.ms`被设为604 800 000（也就是1周），那么日志片段最多需要17天才会过期。这是因为关闭日志片段需要10天，而根据配置的过期时间，还需要再保留数据7天（要等到日志片段的最后一条消息过期才能将其删除）。

### `log.roll.ms`

默认值：`null`

另一个可用于控制日志片段关闭时间的参数是`log.roll.ms`，它指定了多长时间之后日志片段可以被关闭。就像`log.retention.bytes`和`log.retention.ms`一样，`log.segment.bytes`和`log.roll.ms`并不互斥。日志片段会在大小或时间达到上限时被关闭，就看哪个条件先得到满足。在默认情况下，`log.roll.ms`没有设定值，所以使用`log.roll.hours`设定的默认值——168小时，也就是7天。

> 在使用基于时间的日志片段时，需要考虑并行关闭多个日志片段对磁盘性能的影响。如果多个分区的日志片段一直未达到大小的上限，就会出现这种情况。这是因为`broker`在启动时会计算日志片段的过期时间，一旦满足条件，就会并行关闭它们，尽管它们的数据量可能很少。

### `min.insync.replicas`

默认值：1

为了提升集群的数据持久性，可以将`min.insync.replicas`设置为2，确保至少有两个副本跟生产者保持“同步”。生产者需要配合将`ack`设置为`all`，这样就可以确保至少有两个副本（首领和另一个副本）确认写入成功，从而防止在以下情况下丢失数据：首领确认写入，然后发生停机，所有权被转移到一个副本，但这个副本没有写入成功。如果没有这些配置，则生产者会认为已经写入成功，但实际上消息丢失了。不过，这样做是有副作用的，因为需要额外的开销，所以效率会有所降低。因此，对于能够容忍偶尔消息丢失的高吞吐量集群，不建议修改这个参数的默认值。

### `message.max.bytes`

默认值：1048576 (1MB)

`broker`通过设置`message.max.bytes`参数来限制单条消息的大小，默认值是1 000000，也就是1 MB。如果生产者尝试发送超过这个大小的消息，那么不仅消息不会被`broker`接收，还会收到`broker`返回的错误信息。与其他`broker`配置参数一样，这个参数指的是压缩后的消息大小，也就是说，消息的实际大小可以远大于`message.max.bytes`，只要压缩后小于这个值即可。

这个参数对性能有显著的影响。值越大，负责处理网络连接和请求的线程用在处理请求上的时间就越长。它还会增加磁盘写入块的大小，从而影响`I/O`吞吐量。其他的存储解决方案，比如大对象存储或分层存储，可能也是解决大磁盘写入问题的一种方案。

消费者客户端设置的`fetch.max.bytes`需要与服务器端设置的消息大小保持一致。如果这个参数的值比`message.max.bytes`小，那么消费者就无法读取比较大的消息，进而造成阻塞，无法继续处理消息。在配置`broker`的`replica.fetch.max.bytes`参数时，也遵循同样的原则。

Kafka消费者配置

> *提示：本文基于Kafka 3.7.0写成*

`Kafka`生产者有很多可配置的参数，在[Kafka生产者配置文档](https://kafka.apache.org/documentation/#producerconfigs)中都有说明。它们大部分有合理的默认值，没有必要进行修改。不过有几个参数在内存使用、性能和可靠性方面对生产者影响比较大，本文将详细介绍它们。

## `client.id`

`client.id`是客户端标识符，它的值可以是任意字符串，`broker`用它来识别从客户端发送过来的消息。`client.id`可以被用在日志、指标和配额中。选择一个好的客户端标识符可以让故障诊断变得更容易些。

## `acks`

默认值：`all`

`acks`指定了生产者在多少个分区副本收到消息的情况下才会认为消息写入成功。在默认情况下，`Kafka`会在首领副本收到消息后向客户端回应消息写入成功。这个参数对写入消息的持久性有重大影响，对于不同的场景，使用默认值可能不是最好的选择。

* `acks=0`

  如果`acks=0`，则生产者不会等待任何来自`broker`的响应。也就是说，如果`broker`因为某些问题没有收到消息，那么生产者便无从得知，消息也就丢失了。不过，因为生产者不需要等待`broker`返回响应，所以它们能够以网络可支持的最大速度发送消息，从而达到很高的吞吐量。

* `acks=1`

  如果`acks=1`，那么只要集群的首领副本收到消息，生产者就会收到消息成功写入的响应。如果消息无法到达首领副本（比如首领副本发生崩溃，新首领还未选举出来），那么生产者会收到一个错误响应。为了避免数据丢失，生产者会尝试重发消息。不过，在首领副本发生崩溃的情况下，如果消息还没有被复制到新的首领副本，则消息还是有可能丢失。

* `acks=all`或`acks=-1`

  如果`acks=all`，那么只有当所有副本全部收到消息时，生产者才会收到消息成功写入的响应。这种模式是最安全的，它可以保证不止一个`broker`收到消息，就算有个别`broker`发生崩溃，整个集群仍然可以运行。不过，它的延迟比`acks=1`高，因为生产者需要等待不止一个`broker`确认收到消息。

> 从上面可以看到，为`acks`设置的值越小，生产者发送消息的速度就越快。也就是说，我们通过牺牲可靠性来换取较低的生产者延迟。不过，端到端延迟是指从消息生成到可供消费者读取的时间，这对3种配置来说都是一样的。这是因为为了保持一致性，在消息被写入所有同步副本之前，`Kafka`不允许消费者读取它们。因此，如果你关心的是端到端延迟，而不是生产者延迟，那么就不需要在可靠性和低延迟之间做权衡了：你可以选择最可靠的配置，但仍然可以获得相同的端到端延迟。

## 消息传递时间

### `max.block.ms`

默认值: 60000 (1分钟)

这个参数用于控制在调用`send()`或通过`partitionsFor()`显式地请求元数据时生产者可以发生阻塞的时间。当生产者的发送缓冲区被填满或元数据不可用时，这些方法就可能发生阻塞。当达到`max.block.ms`配置的时间时，就会抛出一个超时异常。

### `delivery.timeout.ms`

默认值: 540000(9分钟)

这个参数用于控制从消息准备好发送（`send()`方法成功返回并将消息放入批次中）到`broker`响应或客户端放弃发送（包括重试）所花费的时间。如果配置的时间不满足这一点，则会抛出异常。通常，成功发送消息的速度要比`delivery.timeout.ms`快得多。如果生产者在重试时超出了`delivery.timeout.ms`，那么将执行回调，并会将`broker`之前返回的错误传给它。如果消息批次还没有发送完毕就超出了`delivery.timeout.ms`，那么也将执行回调，并会将超时异常传给它。**这个时间应该大于或等于`linger.ms`和`request.timeout.ms`**。

### `request.timeout.ms`

默认值：30000 (30秒)

这个参数用于控制生产者在发送消息时等待服务器响应的时间。需要注意的是，这是指生产者在放弃之前等待每个请求的时间，不包括重试、发送之前所花费的时间等。如果设置的值已触及，但服务器没有响应，那么生产者将重试发送，或者执行回调，并传给它一个`TimeoutException`。

## `linger.ms`

默认值：0

这个参数指定了生产者在发送消息批次之前等待更多消息加入批次的时间。生产者会在批次被填满或等待时间达到`linger.ms`时把消息批次发送出去。在默认情况下，只要有可用的发送者线程，生产者都会直接把批次发送出去，就算批次中只有一条消息。把`linger.ms`设置成比0大的数，可以让生产者在将批次发送给服务器之前等待一会儿，以使更多的消息加入批次中。虽然这样会增加一点儿延迟，但也极大地提升了吞吐量。这是因为一次性发送的消息越多，每条消息的开销就越小，如果启用了压缩，则计算量也更少了。

## `retries` 和`retry.backoff.ms`

`retries`默认值：2147483647

`retry.backoff.ms`默认值：1000 (1秒)

当生产者收到来自服务器的错误消息时，这个错误有可能是暂时的（例如，一个分区没有首领）。在这种情况下，`retries`参数可用于控制生产者在放弃发送并向客户端宣告失败之前可以重试多少次。在默认情况下，重试时间间隔是`100`毫秒，但可以通过`retry.backoff.ms`参数来控制重试时间间隔。

并不建议在当前版本的`Kafka`中使用这些参数。相反，你可以测试一下`broker`在发生崩溃之后需要多长时间恢复（也就是直到所有分区都有了首领副本），并设置合理的`delivery.timeout.ms`，让重试时间大于`Kafka`集群从崩溃中恢复的时间，以免生产者过早放弃重试。

生产者并不会重试所有的错误。有些错误不是暂时的，生产者就不会进行重试（例如，“消息太大”错误）。通常，对于可重试的错误，生产者会自动进行重试，所以不需要在应用程序中处理重试逻辑。你要做的是集中精力处理不可重试的错误或者当重试次数达到上限时的情况。

**如果想完全禁用重试，那么唯一可行的方法是将retries设置为0。**

## `buffer.memory`

默认值：33554432

这个参数用来设置生产者要发送给服务器的消息的内存缓冲区大小。如果应用程序调用`send()`方法的速度超过生产者将消息发送给服务器的速度，那么生产者的缓冲空间可能会被耗尽，后续的`send()`方法调用会等待内存空间被释放，如果在`max.block.ms`之后还没有可用空间，就抛出异常。需要注意的是，这个异常与其他异常不一样，它是`send()`方法而不是`Future`对象抛出来的。

## `compression.type`

默认值：`none`

在默认情况下，生产者发送的消息是未经压缩的。这个参数可以被设置为`none`, `gzip`、 `snappy`、`lz4`、或`zstd`，这指定了消息被发送给`broker`之前使用哪一种压缩算法。`snappy`压缩算法由谷歌发明，虽然占用较少的`CPU`时间，但能提供较好的性能和相当可观的压缩比。如果同时有性能和网络带宽方面的考虑，那么可以使用这种算法。`gzip`压缩算法通常会占用较多的`CPU`时间，但提供了更高的压缩比。如果网络带宽比较有限，则可以使用这种算法。使用压缩可以降低网络传输和存储开销，而这些往往是向Kafka发送消息的瓶颈所在。

## `batch.size`

默认值：16384

当有多条消息被发送给同一个分区时，生产者会把它们放在同一个批次里。这个参数指定了一个批次可以使用的内存大小。需要注意的是，该参数是按照字节数而不是消息条数来计算的。当批次被填满时，批次里所有的消息都将被发送出去。但是生产者并不一定都会等到批次被填满时才将其发送出去。那些未填满的批次，甚至只包含一条消息的批次也有可能被发送出去。所以，就算把批次大小设置得很大，也不会导致延迟，只是会占用更多的内存而已。但如果把批次大小设置得太小，则会增加一些额外的开销，因为生产者需要更频繁地发送消息。

## `max.in.flight.requests.per.connection`

默认值：5

这个参数指定了生产者在收到服务器响应之前可以发送多少个消息批次。它的值越大，占用的内存就越多，不过吞吐量也会得到提升。实验数据表明，在单数据中心环境中，该参数被设置为2时可以获得最佳的吞吐量，但使用默认值5也可以获得差不多的性能。

**`Kafka`可以保证同一个分区中的消息是有序的**。也就是说，如果生产者按照一定的顺序发送消息，那么`broker`会按照这个顺序把它们写入分区，消费者也会按照同样的顺序读取它们。在某些情况下，顺序是非常重要的。

假设我们把`retries`设置为非零的整数，并把`max.in.flight.requests.per.connection`设置为比1大的数。如果第一个批次写入失败，第二个批次写入成功，那么`broker`会重试写入第一个批次，等到第一个批次也写入成功，两个批次的顺序就反过来了。

我们希望至少有2个正在处理中的请求（出于性能方面的考虑），并且可以进行多次重试（出于可靠性方面的考虑），这个时候，最好的解决方案是将`enable.idempotence`设置为`true`。这样就可以在最多有5个正在处理中的请求的情况下保证顺序，并且可以保证重试不会引入重复消息。

## `enable.idempotence`

默认值：`true`

从0.11版本开始，`Kafka`支持精确一次性`exactly once`语义。幂等生产者是它的一个简单且重要的组成部分。

假设为了最大限度地提升可靠性，你将生产者的`acks`设置为`all`，并将`delivery.timeout.ms`设置为一个比较大的数，允许进行尽可能多的重试。这些配置可以确保每条消息被写入`Kafka`至少一次。但在某些情况下，消息有可能被写入`Kafka`不止一次。假设一个`broker`收到了生产者发送的消息，然后消息被写入本地磁盘并成功复制给了其他`broker`。此时，这个`broker`还没有向生产者发送响应就发生了崩溃。而生产者将一直等待，直到达到`request.timeout.ms`，然后进行重试。重试发送的消息将被发送给新的首领，而这个首领已经有这条消息的副本，因为之前写入的消息已经被成功复制给它了。现在，你就有了一条重复的消息。

为了避免这种情况，可以将`enable.idempotence`设置为`true`。当幂等生产者被启用时，生产者将给发送的每一条消息都加上一个序列号。如果`broker`收到具有相同序列号的消息，那么它就会拒绝第二个副本，而生产者则会收到`DuplicateSequenceException`，这个异常对生产者来说是无害的。

如果要启用幂等性，那么`max.in.flight.requests.per.connection`应小于或等于5、`retries`应大于0，并且acks被设置为all。如果设置了不恰当的值，则会抛出`ConfigException`异常。

## `max.request.size`

默认值：1048576

这个参数用于控制生产者发送的请求的大小。它限制了可发送的单条最大消息的大小和单个请求的消息总量的大小。假设这个参数的值为`1 MB`，那么可发送的单条最大消息就是`1 MB`，或者生产者最多可以在单个请求里发送一条包含`1024`个大小为`1 KB`的消息。另外，`broker`对可接收的最大消息也有限制`message.max.bytes`，其两边的配置最好是匹配的，以免生产者发送的消息被`broker`拒绝。

## `receive.buffer.bytes`和`send.buffer.bytes`

`receive.buffer.bytes`默认值：65536

`send.buffer.bytes`默认值：131072

这两个参数分别指定了`TCP socket`接收和发送数据包的缓冲区大小。如果它们被设为`–1`，就使用操作系统默认值。如果生产者或消费者与`broker`位于不同的数据中心，则可以适当加大它们的值，因为跨数据中心网络的延迟一般都比较高，而带宽又比较低。

AI

如何使用 Workers AI 构建强大的 AI 应用程序

最近发布

PostgreSQL修改表结构

PostgreSQL开启远程访问

Kafka消费者配置

Kafka生产者配置

分类

算法精解公众号

鬼畜动图小程序