基于 Kafka 架构中的 Cosmos DB 一致性级别

2025 年 5 月 16 日 | 阅读 8 分钟

引言

Cosmos DB 是 Microsoft Azure 提供的一项全球分布式、多模型数据库服务。它的一个关键特性是可调的一致性级别，允许开发人员根据应用程序需求在可用性、性能和数据一致性之间取得平衡。当将 Cosmos DB 与 Apache Kafka 集成时，理解一致性级别对于确保跨分布式系统正确、高效、可靠地处理数据至关重要。

为什么一致性级别在基于 Kafka 的架构中很重要

Kafka 是一个分布式事件流平台，其与 Cosmos DB 的集成需要为不同的数据处理模式选择正确的一致性级别。

最终一致性：最适合实时分析，其中数据传播的轻微延迟是可以接受的。
会话一致性：适用于事务性事件处理，其中同一生产者必须读取自己的写入。
强一致性：对于关键任务数据至关重要，例如银行交易或订单履行。

不同一致性级别的 Kafka-Cosmos DB 集成

(a) Kafka Producer → Cosmos DB Sink（最终一致性）

此示例模拟了一个 Kafka producer，它以最终一致性将实时事件发送到 Cosmos DB 进行分析。

Kafka 生产者

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;
import java.util.HashMap;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;

public class KafkaEventProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        ObjectMapper objectMapper = new ObjectMapper();
        for (int i = 1; i <= 5; i++) {
            try {
                Map<String, Object> event = new HashMap<>();
                event.put("event_id", i);
                event.put("status", "processed");
                String message = objectMapper.writeValueAsString(event);
                
                producer.send(new ProducerRecord<>("cosmos-events", message));
                System.out.println("Sent: " + message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.flush();
        producer.close();
    }
}

输出

Cosmos DB Consistency Levels in Kafka-based Architectures

Kafka Consumer（使用最终一致性写入 Cosmos DB）

import com.azure.cosmos.*;
import com.azure.cosmos.models.CosmosContainer;
import com.azure.cosmos.models.CosmosDatabase;
import com.azure.cosmos.models.PartitionKey;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaToCosmosConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "cosmos-consumer-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("cosmos-events"));
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint("https://your-account.documents.azure.com")
                .key("your-key")
                .consistencyLevel(ConsistencyLevel.EVENTUAL)
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase("KafkaDB");
        // Poll messages from Kafka and insert into Cosmos DB
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                container.createItem(record.value(), new PartitionKey("event_id"), null);
                System.out.println("Stored in Cosmos DB: " + record.value());
            }
        }
    }
}

输出

为什么选择最终一致性？

这对于分析来说是最佳的，因为过时读取不会影响结果。
它最大限度地减少了高吞吐量事件的写入延迟。

(b) Kafka Producer → 强一致性的 Cosmos DB

对于关键交易（例如，金融数据），强一致性可确保没有过时读取或不正确的顺序。

在 Cosmos DB 中设置强一致性

修改 Cosmos DB 的一致性设置

{
    "defaultConsistencyLevel": "Strong"
}

Kafka Consumer 使用强一致性写入

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;
import org.apache.kafka.clients.consumer.*;

import java.util.Collections;
import java.util.Properties;

public class KafkaCosmosConsumerStrongConsistency {
    public static void main(String[] args) {
        // Initialize Cosmos DB Client with Strong Consistency
                .endpoint("https://your-account.documents.azure.com")
                .key("your-key")
                .consistencyLevel(ConsistencyLevel.STRONG) // Enforce Strong Consistency
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase("KafkaDB");
        CosmosContainer container = database.getContainer("Events");

        // Configure Kafka Consumer
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "cosmos-consumer-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("cosmos-events"));

        // Poll Kafka messages and write to Cosmos DB with Strong Consistency
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                CosmosItemResponse<Object> response = container.createItem(
                        record.value(),
                        new PartitionKey(record.key()),
                        new CosmosItemRequestOptions().setConsistencyLevel(ConsistencyLevel.STRONG)
                );
                System.out.println("Stored in Cosmos DB: " + record.value());
            }
        }
    }
}

输出

为什么选择强一致性？

确保金融交易的严格排序。
防止由于 Kafka 重平衡导致的重复处理等问题。

选择正确的一致性级别

实时分析：最终一致性
事件驱动的微服务：会话一致性或一致性前缀
金融交易：强一致性
全局分布的应用程序：有界陈旧性

Cosmos DB 一致性级别

Cosmos DB 提供五个不同的 C 一致性级别

强一致性
有界陈旧性
会话一致性
一致性前缀
最终一致性

这些级别中的每一个在一致性、可用性和延迟之间都提供了不同的权衡。让我们在基于 Kafka 的架构的上下文中检查每一个。

1. 强一致性

特性

保证线性化，这意味着读取操作始终返回最新的写入。
确保所有区域在任何时候都看到相同版本的数据。
由于严格的同步，会产生最高的延迟。

对基于 Kafka 的架构的影响

生产者：当 Kafka producer 将消息写入 Cosmos DB 时，强一致性可确保所有消费者始终获得最新数据。但是，写入延迟会增加，从而影响实时处理。
消费者：从 Kafka 读取的消费者可以期望一致性，但如果它们依赖 Cosmos DB 作为源，可能会经历更慢的检索。

示例程序：Kafka 中的强一致性

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;

public class CosmosDBStrongConsistency {
    private static final String ENDPOINT = "<your-cosmosdb-endpoint>";
    private static final String KEY = "<your-cosmosdb-key>";
    private static final String DATABASE_NAME = "KafkaData";
    private static final String CONTAINER_NAME = "Messages";

    public static void main(String[] args) {
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint(ENDPOINT)
                .key(KEY)
                .consistencyLevel(ConsistencyLevel.STRONG) // Set Strong Consistency
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase(DATABASE_NAME);
        CosmosContainer container = database.getContainer(CONTAINER_NAME);

        // Insert a strongly consistent message
        insertMessage(container, "1", "Kafka event");
        // Close Cosmos Client
        cosmosClient.close();
    }

    private static void insertMessage(CosmosContainer container, String messageId, String messageText) {
        CosmosItemResponse<Object> response = container.upsertItem(
                new Message(messageId, messageText, System.currentTimeMillis())
        );

        System.out.println("Message inserted with strong consistency: " + response.getItem());
    }

    // Helper class to define the message structure
    static class Message {
        private String id;
        private String message;
        private long timestamp;

        public Message(String id, String message, long timestamp) {
            this.id = id;
            this.message = message;
            this.timestamp = timestamp;
        }

        // Getters & Setters
        public String getId() { return id; }
        public String getMessage() { return message; }
        public long getTimestamp() { return timestamp; }
    }
}

输出

2. 有界陈旧性

特性

确保读取操作滞后于写入操作一个预定义的时间窗口或版本数。
提供写入的全局顺序，使其适用于有序事件处理。

对基于 Kafka 的架构的影响

生产者：存储在 Cosmos DB 中的 Kafka 消息将保持顺序，但会有一个受控的滞后。
消费者：数据可能略有陈旧，但顺序得以保留，适用于事件溯源和分析。

示例程序：Kafka 中的有界陈旧性

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;

import java.util.Iterator;

public class CosmosDBBoundedStalenessReader {
    private static final String ENDPOINT = "https://your-account.documents.azure.com";
    private static final String KEY = "your-key";
    private static final String DATABASE_NAME = "KafkaData";
    private static final String CONTAINER_NAME = "Messages";

    public static void main(String[] args) {
        // Initialize Cosmos DB Client with Bounded Staleness Consistency
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint(ENDPOINT)
                .key(KEY)
                .consistencyLevel(ConsistencyLevel.BOUNDED_STALENESS) // Set Bounded Staleness
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase(DATABASE_NAME);
        CosmosContainer container = database.getContainer(CONTAINER_NAME);

        // Fetch and print latest messages
        fetchLatestMessages(container);

        // Close Cosmos Client
        cosmosClient.close();
    }

    private static void fetchLatestMessages(CosmosContainer container) {
        CosmosPagedIterable<Object> items = container.readAllItems(new PartitionKey("partition-key"), Object.class);

        Iterator<Object> iterator = items.iterator();
        while (iterator.hasNext()) {
            System.out.println("Message: " + iterator.next());
        }
    }
}

输出

3. 会话一致性

特性

在会话内保证读写一致性。
非常适合单用户或服务特定的交互。

对基于 Kafka 的架构的影响

生产者：每个 producer 实例会立即看到自己的写入，但可能不会立即看到其他实例的写入。
消费者：使用相同会话令牌的消费者将看到一致的数据，从而提高性能。

示例程序：Kafka 中的会话一致性

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;

import java.util.List;

public class CosmosDBSessionConsistency {
    private static final String ENDPOINT = "<your-cosmosdb-endpoint>";
    private static final String KEY = "<your-cosmosdb-key>";
    private static final String DATABASE_NAME = "KafkaData";
    private static final String CONTAINER_NAME = "Messages";

    public static void main(String[] args) {
        // Initialize Cosmos DB Client with Session Consistency
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint(ENDPOINT)
                .key(KEY)
                .consistencyLevel(ConsistencyLevel.SESSION) // Set Session Consistency
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase(DATABASE_NAME);
        CosmosContainer container = database.getContainer(CONTAINER_NAME);

        // Insert and fetch messages within the same session
        insertAndFetchSession(container);

        // Close Cosmos Client
        cosmosClient.close();
    }

    private static void insertAndFetchSession(CosmosContainer container) {
        // Insert an item with session consistency
        CosmosItemResponse<Object> response = container.upsertItem(
                new Message("2", "Session consistency event", System.currentTimeMillis())
        );

        System.out.println("Inserted session-consistent message: " + response.getItem());

        // Fetch and print all messages
        List<CosmosItemProperties> sessionItems = container.readAllItems(
                new CosmosQueryRequestOptions(), CosmosItemProperties.class
        ).stream().toList();

        for (CosmosItemProperties item : sessionItems) {
            System.out.println("Session Read: " + item.getString("message"));
        }
    }

    // Helper class to define the message structure
    static class Message {
        private String id;
        private String message;
        private long timestamp;

        public Message(String id, String message, long timestamp) {
            this.id = id;
            this.message = message;
            this.timestamp = timestamp;
        }

        // Getters & Setters
        public String getId() { return id; }
        public String getMessage() { return message; }
        public long getTimestamp() { return timestamp; }
    }
}

输出

4. 一致性前缀

特性

确保读取操作永远不会返回乱序的写入。
不保证即时可见性，但能保持正确的排序。

对基于 Kafka 的架构的影响

生产者：确保事件流有序，适用于基于日志的存储。
消费者：顺序处理数据的消费者可以依赖正确的排序，但可能会经历轻微的陈旧性。

示例程序：Kafka 中的一致性前缀

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;

public class CosmosDBConsistentPrefix {
    private static final String ENDPOINT = "<your-cosmosdb-endpoint>";
    private static final String KEY = "<your-cosmosdb-key>";
    private static final String DATABASE_NAME = "KafkaData";
    private static final String CONTAINER_NAME = "Messages";

    public static void main(String[] args) {
        // Initialize Cosmos DB Client with Consistent Prefix Consistency
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint(ENDPOINT)
                .key(KEY)
                .consistencyLevel(ConsistencyLevel.CONSISTENT_PREFIX) // Set Consistent Prefix
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase(DATABASE_NAME);
        CosmosContainer container = database.getContainer(CONTAINER_NAME);

        // Insert ordered messages
        insertOrderedMessages(container);

        // Close Cosmos Client
        cosmosClient.close();
    }

    private static void insertOrderedMessages(CosmosContainer container) {
        // Insert ordered events
        container.upsertItem(new Message("3", "Ordered event 1", System.currentTimeMillis()));
        container.upsertItem(new Message("4", "Ordered event 2", System.currentTimeMillis()));

        System.out.println("Ordered events inserted with Consistent Prefix consistency");
    }

    // Helper class to define the message structure
    static class Message {
        private String id;
        private String message;
        private long timestamp;

        public Message(String id, String message, long timestamp) {
            this.id = id;
            this.message = message;
            this.timestamp = timestamp;
        }

        // Getters & Setters
        public String getId() { return id; }
        public String getMessage() { return message; }
        public long getTimestamp() { return timestamp; }
    }
}

输出

5. 最终一致性

特性

不对数据的时效性做任何保证；读取可能不一致。
提供最低的延迟和最高的可用性。

对基于 Kafka 的架构的影响

生产者：允许高吞吐量的 Kafka 摄取与 Cosmos DB。
消费者：适用于一致性不是优先级的应用程序，例如推荐引擎。

示例程序：Kafka 中的最终一致性

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;

import java.util.List;

public class CosmosDBEventualConsistency {
    private static final String ENDPOINT = "<your-cosmosdb-endpoint>";
    private static final String KEY = "<your-cosmosdb-key>";
    private static final String DATABASE_NAME = "KafkaData";
    private static final String CONTAINER_NAME = "Messages";

    public static void main(String[] args) {
        // Initialize Cosmos DB Client with Eventual Consistency
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint(ENDPOINT)
                .key(KEY)
                .consistencyLevel(ConsistencyLevel.EVENTUAL) // Set Eventual Consistency
                .buildClient();

        CosmosDatabase database = cosmosClient.getDatabase(DATABASE_NAME);
        CosmosContainer container = database.getContainer(CONTAINER_NAME);

        // Fetch and print messages
        fetchEventualMessages(container);

        // Close Cosmos Client
        cosmosClient.close();
    }

    private static void fetchEventualMessages(CosmosContainer container) {
        // Read all items from the container
        List<CosmosItemProperties> items = container.readAllItems(new PartitionKey("id"), CosmosItemProperties.class).stream().toList();

        System.out.println("Messages fetched with eventual consistency:");
        for (CosmosItemProperties item : items) {
            System.out.println(item.toJson());
        }
    }
}

输出

为 Kafka 选择正确的一致性级别

一致性级别	用例
强	金融交易、关键数据系统
有界陈旧性	具有严格顺序的事件流
Session	具有独立会话保证的微服务
一致性前缀	具有顺序保证的日志处理
最终	高性能应用程序，具有宽松的一致性

下一主题监控和调整 Kafka 到 Cosmos DB 的管道

基于 Kafka 架构中的 Cosmos DB 一致性级别

引言

为什么一致性级别在基于 Kafka 的架构中很重要

不同一致性级别的 Kafka-Cosmos DB 集成

(a) Kafka Producer → Cosmos DB Sink（最终一致性）

(b) Kafka Producer → 强一致性的 Cosmos DB

选择正确的一致性级别

Cosmos DB 一致性级别

1. 强一致性

2. 有界陈旧性

3. 会话一致性

4. 一致性前缀

5. 最终一致性

为 Kafka 选择正确的一致性级别

联系信息

关注我们

教程

面试题

在线编译器

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 一致性级别

引言

为什么一致性级别在基于 Kafka 的架构中很重要

不同一致性级别的 Kafka-Cosmos DB 集成

(a) Kafka Producer → Cosmos DB Sink（最终一致性）

(b) Kafka Producer → 强一致性的 Cosmos DB

选择正确的一致性级别

Cosmos DB 一致性级别

1. 强一致性

2. 有界陈旧性

3. 会话一致性

4. 一致性前缀

5. 最终一致性

为 Kafka 选择正确的一致性级别

相关帖子

Kafka 消费者重新平衡

Kafka 中的数据捕获

Kafka 中的 Zookeeper：详细信息

最佳 Apache Kafka 书籍

从 Kafka 到 Cosmos DB 的变更数据捕获 (CDC)

Kafka API

Kafka 保留策略

Kafka 中的容错性

使用 Kafka 设计容错微服务

使用 Kafka 和 ML 进行自动化异常检测

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器