如何用Rust构建一个简单的Kafka生产者和消费者应用程序

484次阅读  |  发布于1年以前

在这篇文章中，我们将构建一个简单的Rust应用程序，由生产者和消费者组成，使用来自Hacker News的搜索数据。

先决条件

建立一个新的Rust项目

在本例中，我们将利用Cargo的工作区特性。工作区包含多个应用程序或库，但可以从顶层目录编译。

让我们开始吧，创建包含所有源代码的项目目录。

$ mkdir kafka_hn_processing
$ cd kafka_hn_processing

进入新目录，设置我们的工作空间：

$ cat > Cargo.toml
[workspace]
members = [
    "producer"
]

接下来，我们创建“producer”项目：

$ cargo new producer
$ cd producer

要使这个应用程序工作，我们需要安装一些依赖项。为此我们使用了cargo add(注意：cargo add至少需要Rust 1.62)。cargo add下载依赖项并将其添加到项目的Cargo.toml中。

$ cargo add kafka -p producer 
$ cargo add reqwest --features=json -p producer 
$ cargo add tokio -p producer --features=full
$ cargo add serde --features=derive -p producer
$ cargo add serde_json  -p producer
$ cargo add urlencoding -p producer

我们添加了如下依赖项：

设置一个开发容器

在开始编写代码之前，我们需要设置Docker环境。在项目的根目录下，创建docker-compose.yml：

---
 version: '3.8'
 services:
   rust-log-processing:
     image: mcr.microsoft.com/devcontainers/rust:0-1-bullseye
     volumes:
       - ..:/workspaces:cached
     cap_add:
       - SYS_PTRACE
     security_opt:
       - seccomp:unconfined
     command: /bin/sh -c "while sleep 1000; do :; done"
   zookeeper:
     image: confluentinc/cp-zookeeper:7.3.0
     container_name: zookeeper
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
   broker:
     image: confluentinc/cp-kafka:7.3.0
     container_name: broker
     ports:
     # To learn about configuring Kafka for access across networks see
     # https://www.confluent.io/blog/kafka-client-cannot-connect-to-broker-on-aws-on-docker-etc/
       - "9092:9092"
     depends_on:
       - zookeeper
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://broker:29092
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

我们不会直接在本地机器上运行Rust代码，而是在Docker容器中运行。VSCode提供了devcontainer特性，允许我们在Docker容器中运行工作空间代码。

要开始使用Dev Container，我们需要一个.devcontainer文件夹来进行VSCode设置。在根级别目录下创建.devcontainer。文件夹结构：

$ mkdir .devcontainer
$ ls -l
 drwxr-xr-x   7 Justin  staff    224  2 24 17:02 .
drwxr-xr-x  13 Justin  staff    416  2 24 16:38 ..
drwxr-xr-x   2 Justin  staff     64  2 24 16:59 .devcontainer
-rw-r--r--   1 Justin  staff  27259  2 24 16:47 Cargo.lock
-rw-r--r--   1 Justin  staff     41  2 24 16:43 Cargo.toml
-rw-r--r--   1 Justin  staff   1225  2 24 16:53 docker-compose.yml
drwxr-xr-x   6 Justin  staff    192  2 24 17:02 producer

然后，在.devcontainer中创建.devcontainer/devcontainer.json配置文件：

{
  "name": "Rust",
  "service": "rust-log-processing",
  "dockerComposeFile": "../docker-compose.yml",
  "features": {
       "ghcr.io/devcontainers/features/rust:1.0.11": {}
   },
  "workspaceFolder": "/workspaces/${localWorkspaceFolderBasename}",
  "shutdownAction": "stopCompose"
}

我们的配置更高级，因为我们也利用了Docker Compose。默认情况下，VSCode在一个容器中运行所有内容。因为我们需要Kafka来测试我们的代码，需要让它知道我们的docker-compose.yml文件。

有了这些文件，我们就可以开始编写代码了。在VSCode中打开项目：

code .

重要提示：确保在容器中打开项目。VSCode通常会提示你这样做，但如果没有，请单击左下角的小图标，并在命令菜单中选择“重新打开容器”。

增加HN搜索代码

首先，我们添加了一些代码，以允许我们与Hacker News API进行通信。我们选择Hacker News是因为它的API不需要认证，并且提供了大量我们可以处理的数据。

创建一个新的文件producer/src/hn.rs：

use serde::{Serialize, Deserialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct HackerNewsResponse {
    pub hits: Vec<HNSearchResult>,
}

#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub struct HNSearchResult {
    pub author: String,
    #[serde(alias = "objectID")]
    pub id: String,
    pub title: String,
    url: Option<String>,
    pub story_text: Option<String>,
    #[serde(alias = "_tags")]
    pub tags: Option<Vec<String>>,
    pub points: u32,
}

pub async fn fetch_hn_stories(search_term: String, search_result_limit: u32) -> Result<HackerNewsResponse, reqwest::Error> {
    let url_encoded_search_term = urlencoding::encode(&search_term);
    let url_str= format!("https://hn.algolia.com/api/v1/search_by_date?query={}&tags=story&hitsPerPage={}", url_encoded_search_term, search_result_limit);
    let client = reqwest::Client::new();
    let request = client.get(url_str)
            .build().unwrap();
    let json_response = client.execute(request)
            .await?
            .json::<HackerNewsResponse>()
            .await?;

    Ok(json_response)
}

我们将使用这段代码通过搜索API从Hacker News获取故事，为Kafka生成内容。在我们的main函数中，我们将调用fetch_hn_stories，其中包含一个搜索词和一个限制，表示我们最多需要多少个结果。

生产者(Producer)

在main.rs中添加一个新函数：

use kafka::producer::{Producer, Record, RequiredAcks};
use std::time::Duration;
use crate::hn::HNSearchResult;

mod hn;

fn send_to_kafka(host: &str, topic: &str, payload: Vec<HNSearchResult>) {
     let mut producer = Producer::from_hosts(vec![host.to_owned()])
         .with_ack_timeout(Duration::from_secs(1))
         .with_required_acks(RequiredAcks::One)
         .create()
         .unwrap();

     for search_result in payload {
         let buffer = serde_json::to_string(&search_result).unwrap();

         producer
             .send(&Record::from_value(topic, buffer.as_bytes()))
             .unwrap();
     }
}

fn main() {
   //...
}

send_to_kafka包含了与Kafka broker通信所需的最小设置。我们配置了一个超时(.with_ack_timeout)以及我们至少需要多少ack才能继续(.with_required_acks)。由于我们的Dev Setup只使用一个broker，所以我们将其设置为1(这个设置可能在生产中有所不同，这取决于你的用例和可用broker的数量)。

在topic中，Kafka以字节的形式存储有效负载。因此，我们需要将它序列化为一个字节数组(buffer.as_bytes)。

send_to_kafka函数就绪后，让我们调用main()：

//producer/src/main.rs
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let stories = hn::fetch_hn_stories("Ruby".into(), 100).await?;
     println!("Fetched {} stories", stories.hits.len());
     send_to_kafka("broker:9092", "hnstories", stories.hits);
     Ok(())
}

我们用任意的搜索词(比如“Ruby”)获取100个故事，然后把它发送给kafka。

创建Kafka topic

在运行代码之前，我们需要创建一个Kafka主题。默认情况下，Kafka安装附带命令行实用程序来帮助完成这些维护任务。在本地计算机的终端上运行以下命令：

$ docker exec broker kafka-topics --bootstrap-server broker:9092 --create --topic hnstories

Created topic hnstories.

消费者(Consumer)

生产者就绪后，让我们创建消费者，读取来自主题的Hacker News搜索结果。在我们运行cargo new之前，打开Cargo.toml，在项目的根目录下添加一个新项目：

[workspace]
members = [
    "producer"
    "consumer"
]

将“consumer”添加到成员列表中，保存并关闭文件。在dev容器中，运行以下命令创建一个新项目：

cargo new consumer

为consumer添加以下依赖项：

$ cargo add serde --features=derive -p consumer
$ cargo add serde_json -p consumer
$ cargo add kafka -p consumer

在main.rs中添加以下代码：

use kafka::consumer::{Consumer, FetchOffset};

fn main() {
    let mut consumer =
       Consumer::from_hosts(vec!("broker:9092".to_owned()))
          .with_topic("hnstories".to_owned())
          .with_fallback_offset(FetchOffset::Earliest)
          .create()
          .unwrap();

    loop {
      for ms in consumer.poll().unwrap().iter() {
        for m in ms.messages() {
          let str = String::from_utf8_lossy(m.value);
          println!("{:?}",str);
        }
        let _ = consumer.consume_messageset(ms);
      }
      consumer.commit_consumed().unwrap();
    }
}

让我们一步一步地研究这段代码。

首先，我们创建一个新的消费者：

let mut consumer =
   Consumer::from_hosts(vec!("broker:9092".to_owned()))
      .with_topic("hnstories".to_owned())
      .with_fallback_offset(FetchOffset::Earliest)
      .create()
      .unwrap();

我们连接到单个broker(broker:9092)。在本例中，broker是由Docker compose管理的域名。我们正在监听单个topic(hnstories)并配置一个回退偏移量。

回退偏移量允许使用者从topic的开头开始读取消息。如果省略此配置，一旦启动并运行，它将不考虑以前的事件。

是时候运行代码了，在VSCode中，打开终端在dev容器的上下文中运行命令。

在VSCode中打开两个终端会话。然后，运行代码：

cargo run -p producer

在第二个会话中，运行：

cargo run -p consumer

这个命令启动consumer应用程序，consumer轮询topic以获取内容，并在接收到新消息时打印出消息。

结语

我们只需要几个步骤就可以开始使用Kafka构建应用程序。我们依赖于Docker的Compose配置，它支持Kafka和Zookeeper的单个实例。有了这些，我们就需要一个生产者和消费者。生产者在主题上写入新数据，而消费者则读取它。为了保持所有东西都是自包含的，我们使用了VSCode的开发容器。

Copyright© 2013-2020

All Rights Reserved 京ICP备2023019179号-8