Kafka 和 Cosmos DB 之间异步数据处理

2025 年 5 月 15 日 | 阅读 5 分钟

引言

异步数据流概念

异步数据流是指数据在不同组件之间移动而无需等待即时响应的系统。与同步处理（其中每个步骤都等待前一个步骤完成）不同，异步系统允许多个进程并发执行。这在需要低延迟、高吞吐量和可伸缩性的分布式系统中特别有用。

在 Kafka 和 Cosmos DB 的上下文中，异步数据流确保 Kafka 发布者生成的消息被消费并存储在 Cosmos DB 中，而不会阻塞发布者。消费者独立处理消息，并将它们摄入 Cosmos DB，从而确保数据流顺畅且不中断。

什么是异步处理？

异步处理是一种非阻塞机制，其中任务独立执行，而无需等待先前的任务完成。它允许多个操作并行运行，从而提高性能和效率。

关键特性

非阻塞执行：任务不必等待前一个任务完成。
并发性：多个操作同时执行。
可伸缩性：轻松处理海量数据处理。
提高性能：降低数据流延迟。

异步处理的一个典型示例是事件驱动架构，其中事件触发操作而无需立即响应。

1. 非阻塞执行

在非阻塞系统中，任务在不等待先前任务完成的情况下继续执行。

程序：使用 Java CompletableFuture 进行非阻塞执行

import java.util.concurrent.CompletableFuture;

public class NonBlockingExample {
    public static void main(String[] args) {
        System.out.println("Task started...");

        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(2000);
                System.out.println("Asynchronous Task Completed!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println("Main thread is not blocked!");
    }
}   

预期输出

Asynchronous Data Processing Between Kafka and Cosmos DB

主线程继续执行，而异步任务单独运行。

2. 并发性

并发性允许多个任务同时执行。

程序：并发运行多个任务

 
import java.util.concurrent.*;

public class ConcurrencyExample {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(3);

        Future<String> task1 = executor.submit(() -> {
            Thread.sleep(2000);
            return "Task 1 Completed!";
        });

        Future<String> task2 = executor.submit(() -> {
            Thread.sleep(1000);
            return "Task 2 Completed!";
        });

        Future<String> task3 = executor.submit(() -> {
            Thread.sleep(3000);
            return "Task 3 Completed!";
        });

        System.out.println(task1.get());
        System.out.println(task2.get());
        System.out.println(task3.get());

        executor.shutdown();
    }
}   

预期输出

任务使用多个线程并行执行，提高了效率。

3. 可扩展性

可伸缩性允许高效地处理大量任务。

程序：同时处理多个请求

 
import java.util.concurrent.*;

public class ScalabilityExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        for (int i = 1; i <= 5; i++) {
            int taskId = i;
            executor.execute(() -> {
                System.out.println("Processing task " + taskId + " by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}   

预期输出

多个任务并行处理，提高了可伸缩性。

4. 改进的性能

异步处理通过独立执行任务来降低延迟。

程序：通过异步调用实现更快的响应时间

 
import java.util.concurrent.CompletableFuture;

public class PerformanceExample {
    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();

        CompletableFuture<String> asyncTask = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
                return "Async Task Completed!";
            } catch (InterruptedException e) {
                return "Error!";
            }
        });

        System.out.println("Doing other work while waiting...");

        String result = asyncTask.join(); // Waits for completion but doesn't block execution initially
        System.out.println(result);

        long endTime = System.currentTimeMillis();
        System.out.println("Total execution time: " + (endTime - startTime) + "ms");
    }
}   

预期输出

应用程序在任务异步执行时保持响应。

示例：同步与异步处理

 
// Synchronous Processing Example
public void processSync() {
    String data = fetchData(); // Waits for data to be fetched
    processData(data); // Waits for processing to complete
    storeData(data); // Waits for storing to finish
}

// Asynchronous Processing Example
public CompletableFuture<Void> processAsync() {
    return CompletableFuture.supplyAsync(this::fetchData)
        .thenApplyAsync(this::processData)
        .thenAcceptAsync(this::storeData);
}   

在现代数据架构中的重要性

在实时数据分析和事件驱动应用程序至关重要的时代，异步数据处理可实现可伸缩性、效率和容错能力。

了解 Kafka

Kafka 架构

Kafka 作为分布式发布-订阅消息系统运行。它由以下几部分组成：

发布者：生成消息并将其发送到 Kafka 主题。
代理：管理消息存储和复制。
消费者：从主题读取消息。
ZooKeeper：协调 Kafka 代理。

Kafka 主题、分区和偏移量

Kafka 将消息组织成主题，主题被分成多个分区以实现并行处理。偏移量跟踪消息的消费情况。

示例：创建 Kafka 主题

 
kafka-topics.sh --create --topic music-streams --bootstrap-server localhost:9092 --partitions 3 --replication-factor 2   

Cosmos DB 简介

Cosmos DB 架构

Cosmos DB 是一项全球分布式、多模型数据库服务，支持键值、文档、图和列族数据模型。

分区和缩放

Cosmos DB 使用分区键将数据高效地分布在多个服务器上。

示例：创建 Cosmos DB 容器

 
from azure.cosmos import CosmosClient
client = CosmosClient("<COSMOS_DB_ENDPOINT>", "<COSMOS_DB_KEY>")
database = client.create_database_if_not_exists("MusicDB")
container = database.create_container_if_not_exists(
    id="MusicEvents",
    partition_key=PartitionKey(path="/songId"),
    offer_throughput=400
)   

设置 Kafka

安装并启动 Kafka

 
# Start Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# Start Kafka Broker
bin/kafka-server-start.sh config/server.properties

# Create a Kafka Topic
bin/kafka-topics.sh --create --topic async-events --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1   

实现 Kafka 发布者

Kafka 发布者将事件异步写入 async-events 主题。

Kafka 发布者代码（Python）

 
from kafka import KafkaProducer
import json
import time

def json_serializer(data):
    return json.dumps(data).encode('utf-8')

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=json_serializer)

for i in range(10):
    event = {"id": i, "name": f"Song_{i}", "popularity": i * 10}
    producer.send("async-events", event)
    print(f"Produced: {event}")
    time.sleep(1)

producer.flush()   

输出

用于 Cosmos DB 的 Kafka 消费者

消费者异步地从 Kafka 读取事件，并将它们插入 **Azure Cosmos DB**。

消费者代码

 
from kafka import KafkaConsumer
from azure.cosmos import CosmosClient
import json

# Cosmos DB Configuration
COSMOS_ENDPOINT = "https://your-cosmos-db.documents.azure.com:443/"
COSMOS_KEY = "your-primary-key"
DATABASE_NAME = "MusicData"
CONTAINER_NAME = "Songs"

# Initialize Cosmos DB Client
client = CosmosClient(COSMOS_ENDPOINT, COSMOS_KEY)
database = client.create_database_if_not_exists(DATABASE_NAME)
container = database.create_container_if_not_exists(
    id=CONTAINER_NAME,
    partition_key=partition_key("/id")
)

consumer = KafkaConsumer(
    "async-events",
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    event = message.value
    container.upsert_item(event)
    print(f"Inserted into Cosmos DB: {event}")   

输出

处理故障和死信队列（DLQ）

应使用死信队列（DLQ）处理消息处理中的故障。

DLQ 实现

修改消费者以处理错误并将失败的消息存储在单独的 Kafka 主题中

 
from kafka import KafkaProducer

def dlq_producer(event):
    producer = KafkaProducer(
        bootstrap_servers=['localhost:9092'],
        value_serializer=json_serializer
    )
    producer.send("async-events-dlq", event)
    producer.flush()

try:
    container.upsert_item(event)
except Exception as e:
    print(f"Error: {e}. Sending to DLQ.")
    dlq_producer(event)   

输出

优化性能

批量处理以提高效率

修改消费者以批量处理事件

 
batch_size = 10
batch = []

for message in consumer:
    batch.append(message.value)
    if len(batch) >= batch_size:
        container.upsert_item(batch)
        print(f"Inserted batch: {batch}")
        batch = []   

监控和日志记录

使用Prometheus和Grafana监控 Kafka 消费者滞后和 Cosmos DB 摄入速率。

Kafka 消费者滞后监控

 
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-group --describe   

Kafka 与 Cosmos DB 的集成

将 Kafka 与 Cosmos DB 集成涉及设置数据管道，其中 Kafka 作为消息代理，Cosmos DB 作为目标存储。

步骤：

Kafka 发布者：将事件发布到 Kafka 主题。
Kafka 消费者：从 Kafka 读取事件。
Cosmos DB 写入器：将数据存储在 Cosmos DB 中。

示例：用于 Cosmos DB 的 Kafka 发布者和消费者

Kafka 生产者代码

 
from kafka import KafkaProducer
import json
import time

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8'))
data = {"user_id": 123, "song": "Afrobeat", "timestamp": time.time()}
producer.send('music_topic', data)
producer.flush()
print("Message sent to Kafka!")   

Kafka 消费者代码（摄入到 Cosmos DB）

 
from kafka import KafkaConsumer
from azure.cosmos import CosmosClient
import json
cosmos_client = CosmosClient("COSMOS_DB_URI", "COSMOS_DB_KEY")
database = cosmos_client.get_database_client("MusicDB")
container = database.get_container_client("Plays")
consumer = KafkaConsumer(
    'music_topic',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earlciest',
    enable_auto_commit=True,
    value_deserializer=lambda x: json.loads(x.decode('utf-8')))
for message in consumer:
    container.upsert_item(message.value)
    print(f"Inserted into Cosmos DB: {message.value}")   

预期输出

下一主题构建-复制管道-kafka-到-cosmos-db

Kafka 和 Cosmos DB 之间异步数据处理

引言

异步数据流概念

什么是异步处理？

1. 非阻塞执行

2. 并发性

3. 可扩展性

4. 改进的性能

了解 Kafka

Kafka 架构

Kafka 主题、分区和偏移量

Cosmos DB 简介

Cosmos DB 架构

分区和缩放

设置 Kafka

实现 Kafka 发布者

用于 Cosmos DB 的 Kafka 消费者

处理故障和死信队列（DLQ）

DLQ 实现

优化性能

监控和日志记录

Kafka 与 Cosmos DB 的集成

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 和 Cosmos DB 之间异步数据处理

引言

异步数据流概念

什么是异步处理？

1. 非阻塞执行

2. 并发性

3. 可扩展性

4. 改进的性能

了解 Kafka

Kafka 架构

Kafka 主题、分区和偏移量

Cosmos DB 简介

Cosmos DB 架构

分区和缩放

设置 Kafka

实现 Kafka 发布者

用于 Cosmos DB 的 Kafka 消费者

处理故障和死信队列（DLQ）

DLQ 实现

优化性能

监控和日志记录

Kafka 与 Cosmos DB 的集成

相关帖子

Kafka 用于跨数据中心复制的挑战

Kafka 连接器 BigQuery

Kafka Connect BigQuery 接收器

Kafka 在灾难恢复规划中的应用

Kafka 中的实时欺诈检测管道

节能 Kafka 集群

分区策略：Kafka 主题和 Cosmos DB 容器

Kafka Streams DSL

使用 Kafka 在零售业中实现超个性化

Kafka 中的容错性

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器