社区
学习
工具库
AI工具
休闲
搜索
简体中文
首页 Java java教程 如何获取和消费使用 Kafka 传递和订购的消息

如何获取和消费使用 Kafka 传递和订购的消息

Linda Hamilton
Linda Hamilton
Nov 06, 2024 pm 09:08 PM

Como conseguir y la consumición entregada y ordenada de mensajes con Kafka

为了确保事件在 Apache Kafka 中以完全一致的顺序发送和消费，必须了解消息分区和消费者分配的工作原理。

在 Kafka 中使用分区

  1. 主题分区:

    • Kafka 将消息组织到主题内的 分区 中。每个分区都维护它接收到的消息的顺序，这意味着消息按照发送到该分区的顺序进行处理。
    • 为了确保顺序，与同一上下文（例如用户 ID 或事务 ID）相关的所有消息都发送到同一分区至关重要。这是通过在发送消息时使用分区键来实现的。 Kafka 使用此键来确定使用哈希函数将消息发送到哪个分区[1][5]。

  2. 消息键:

    • 发送消息时，可以指定键。具有相同密钥的所有消息将被发送到同一分区，这确保它们按照生成的顺序被消费。例如，如果使用用户 ID 作为键，则与该用户相关的所有事件都将进入同一个分区。

消费者群体

  1. 消费者分配：

    • Kafka 中的消费者分为 消费者组。每个组可以有多个消费者，但每个分区一次只能被组内的一个消费者读取。
    • 这意味着如果您的消费者多于分区，则一些消费者将处于不活动状态。为了维持秩序并最大化效率，建议至少有与组中消费者一样多的分区。

  2. 偏移量管理：

    • Kafka 使用 偏移量 存储每个消费者的读取状态，偏移量是分区中每条消息的增量数字标识符。这使得消费者可以在出现故障时从上次中断的地方继续。

附加策略

  • 避免过载：选择分区键时，重要的是要考虑流量分配，以避免某些分区过载而其他分区未充分利用。
  • 复制和容错：确保为分区配置足够的复制（大于1），这不仅提高可用性，而且提高系统对故障的恢复能力。

使用Avro在Kafka中实现消息生产和消费系统，确保消息按顺序处理并处理可能的故障，这里是一个完整的示例。这包括 Avro 架构的定义、生产者和消费者代码以及处理错误的策略。
Avro 方案
首先，我们为有效负载定义 Avro 架构。我们将创建一个名为 user_signed_up.avsc 的文件，用于描述消息的结构。

{
  "type": "record",
  "name": "UserSignedUp",
  "namespace": "com.example",
  "fields": [
    { "name": "userId", "type": "int" },
    { "name": "userEmail", "type": "string" },
    { "name": "timestamp", "type": "string" } // Formato ISO 8601
  ]
}
登录后复制
登录后复制

密钥生成
为了保证消息生产和消费的顺序，我们将使用结构为 message-type-date 的 key，例如：user-signed-up-2024-11-04。
制片人卡夫卡
以下是使用 Avro 方案向 Kafka 发送消息的 生产者 的代码：

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Properties;

public class AvroProducer {
    private final KafkaProducer<String, byte[]> producer;
    private final Schema schema;

    public AvroProducer(String bootstrapServers) throws IOException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");

// Establecer la propiedad de reintentos, Número de reintentos
        properties.put(ProducerConfig.RETRIES_CONFIG, 3); 
// Asegura que todos los réplicas reconozcan la escritura,
        properties.put(ProducerConfig.ACKS_CONFIG, "all"); 
// Solo un mensaje a la vez
properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); 
// Habilitar idempotencia, no quiero enviar duplicados
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); 

        this.producer = new KafkaProducer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void sendMessage(String topic, int userId, String userEmail) {
        GenericRecord record = new GenericData.Record(schema);
        record.put("userId", userId);
        record.put("userEmail", userEmail);
        record.put("timestamp", java.time.Instant.now().toString());

        String key = String.format("user-signed-up-%s", java.time.LocalDate.now());

        ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>(topic, key, serialize(record));

        producer.send(producerRecord, (metadata, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
**handleFailure(exception, producerRecord);
**            } else {
                System.out.printf("Mensaje enviado a la partición %d con offset %d%n", metadata.partition(), metadata.offset());
            }
        });
    }

private void handleFailure(Exception exception, ProducerRecord<String, byte[]> producerRecord) {
        // Log the error for monitoring
        System.err.println("Error sending message: " + exception.getMessage());

        // Implement local persistence as a fallback
        saveToLocalStorage(producerRecord);

        // Optionally: Notify an external monitoring system or alert
    }

    private void saveToLocalStorage(ProducerRecord<String, byte[]> record) {
        try {
            // Persist the failed message to a local file or database for later processing
            Files.write(new File("failed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE, 
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje guardado localmente para reenvío: " + record.key());
        } catch (IOException e) {
            System.err.println("Error saving to local storage: " + e.getMessage());
        }
    }

    private byte[] serialize(GenericRecord record) {
        // Crear un ByteArrayOutputStream para almacenar los bytes serializados
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    // Crear un escritor de datos para el registro Avro
    DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(record.getSchema());

    // Crear un encoder para escribir en el ByteArrayOutputStream
    Encoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);

    try {
        // Escribir el registro en el encoder
        datumWriter.write(record, encoder);
        // Finalizar la escritura
        encoder.flush();
    } catch (IOException e) {
        throw new AvroSerializationException("Error serializing Avro record", e);
    }

    // Devolver los bytes serializados
    return outputStream.toByteArray();
    }

    public void close() {
        producer.close();
    }
}
登录后复制
登录后复制

重试注意事项
**需要注意的是，启用重试时，如果处理不当，可能存在消息重新排序的风险。
为了避免这种情况：
**max.in.flight.requests.per.connection：您可以将此属性设置为 1，以确保一次发送一条消息并按顺序处理。但是，这可能会影响性能。
通过此配置和正确的错误处理，您可以确保您的 Kafka 生产者更加健壮，能够处理消息生产中的故障，同时保持必要的顺序。

**卡夫卡消费者
**读取并处理消息的消费者：

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.DatumReader;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;

public class AvroConsumer {
    private final KafkaConsumer<String, byte[]> consumer;
    private final Schema schema;

    public AvroConsumer(String bootstrapServers, String groupId) throws IOException {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");

        this.consumer = new KafkaConsumer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void consume(String topic) {
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, byte[]> record : records) {
                try {
                    processMessage(record.value());
                } catch (Exception e) {
                    handleProcessingError(e, record);
                }
            }
        }
    }

    private void processMessage(byte[] data) throws IOException {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        var decoder = DecoderFactory.get().binaryDecoder(data, null);
        GenericRecord record = reader.read(null, decoder);

        System.out.printf("Consumido mensaje: %s - %s - %s%n", 
            record.get("userId"), 
            record.get("userEmail"), 
            record.get("timestamp"));
    }

    private void handleProcessingError(Exception e, ConsumerRecord<String, byte[]> record) {
        System.err.println("Error processing message: " + e.getMessage());

        // Implement logic to save failed messages for later processing
        saveFailedMessage(record);
    }

    private void saveFailedMessage(ConsumerRecord<String, byte[]> record) {
        try {
            // Persist the failed message to a local file or database for later processing
            Files.write(new File("failed_consumed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE,
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje consumido guardado localmente para re-procesamiento: " + record.key());
        } catch (IOException e) {
            System.err.println("Error saving consumed message to local storage: " + e.getMessage());
        }
    }

    public void close() {
        consumer.close();
    }
}
登录后复制
登录后复制

按键的现实示例
在具有许多不同事件和许多不同分区的环境中，实际的键可能类似于：

{
  "type": "record",
  "name": "UserSignedUp",
  "namespace": "com.example",
  "fields": [
    { "name": "userId", "type": "int" },
    { "name": "userEmail", "type": "string" },
    { "name": "timestamp", "type": "string" } // Formato ISO 8601
  ]
}
登录后复制
登录后复制

这允许将特定日期与特定类型相关的所有事件发送到同一分区并按顺序处理。此外，如有必要，您可以通过包含更多详细信息（例如会话或交易 ID）来使密钥多样化。
通过这种使用 Avro 处理故障和确保 Kafka 中消息顺序的实现和策略，您可以构建一个健壮且高效的事件管理系统。

现在是一个能力更强的 Kafka 生产者和消费者。

具有断路器、本地持久性和 DLQ 的 Kafka Producer。

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Properties;

public class AvroProducer {
    private final KafkaProducer<String, byte[]> producer;
    private final Schema schema;

    public AvroProducer(String bootstrapServers) throws IOException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");

// Establecer la propiedad de reintentos, Número de reintentos
        properties.put(ProducerConfig.RETRIES_CONFIG, 3); 
// Asegura que todos los réplicas reconozcan la escritura,
        properties.put(ProducerConfig.ACKS_CONFIG, "all"); 
// Solo un mensaje a la vez
properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); 
// Habilitar idempotencia, no quiero enviar duplicados
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); 

        this.producer = new KafkaProducer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void sendMessage(String topic, int userId, String userEmail) {
        GenericRecord record = new GenericData.Record(schema);
        record.put("userId", userId);
        record.put("userEmail", userEmail);
        record.put("timestamp", java.time.Instant.now().toString());

        String key = String.format("user-signed-up-%s", java.time.LocalDate.now());

        ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>(topic, key, serialize(record));

        producer.send(producerRecord, (metadata, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
**handleFailure(exception, producerRecord);
**            } else {
                System.out.printf("Mensaje enviado a la partición %d con offset %d%n", metadata.partition(), metadata.offset());
            }
        });
    }

private void handleFailure(Exception exception, ProducerRecord<String, byte[]> producerRecord) {
        // Log the error for monitoring
        System.err.println("Error sending message: " + exception.getMessage());

        // Implement local persistence as a fallback
        saveToLocalStorage(producerRecord);

        // Optionally: Notify an external monitoring system or alert
    }

    private void saveToLocalStorage(ProducerRecord<String, byte[]> record) {
        try {
            // Persist the failed message to a local file or database for later processing
            Files.write(new File("failed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE, 
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje guardado localmente para reenvío: " + record.key());
        } catch (IOException e) {
            System.err.println("Error saving to local storage: " + e.getMessage());
        }
    }

    private byte[] serialize(GenericRecord record) {
        // Crear un ByteArrayOutputStream para almacenar los bytes serializados
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    // Crear un escritor de datos para el registro Avro
    DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(record.getSchema());

    // Crear un encoder para escribir en el ByteArrayOutputStream
    Encoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);

    try {
        // Escribir el registro en el encoder
        datumWriter.write(record, encoder);
        // Finalizar la escritura
        encoder.flush();
    } catch (IOException e) {
        throw new AvroSerializationException("Error serializing Avro record", e);
    }

    // Devolver los bytes serializados
    return outputStream.toByteArray();
    }

    public void close() {
        producer.close();
    }
}
登录后复制
登录后复制

具有 DLQ 管理功能的 Kafka 消费者。

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.DatumReader;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;

public class AvroConsumer {
    private final KafkaConsumer<String, byte[]> consumer;
    private final Schema schema;

    public AvroConsumer(String bootstrapServers, String groupId) throws IOException {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");

        this.consumer = new KafkaConsumer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void consume(String topic) {
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, byte[]> record : records) {
                try {
                    processMessage(record.value());
                } catch (Exception e) {
                    handleProcessingError(e, record);
                }
            }
        }
    }

    private void processMessage(byte[] data) throws IOException {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        var decoder = DecoderFactory.get().binaryDecoder(data, null);
        GenericRecord record = reader.read(null, decoder);

        System.out.printf("Consumido mensaje: %s - %s - %s%n", 
            record.get("userId"), 
            record.get("userEmail"), 
            record.get("timestamp"));
    }

    private void handleProcessingError(Exception e, ConsumerRecord<String, byte[]> record) {
        System.err.println("Error processing message: " + e.getMessage());

        // Implement logic to save failed messages for later processing
        saveFailedMessage(record);
    }

    private void saveFailedMessage(ConsumerRecord<String, byte[]> record) {
        try {
            // Persist the failed message to a local file or database for later processing
            Files.write(new File("failed_consumed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE,
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje consumido guardado localmente para re-procesamiento: " + record.key());
        } catch (IOException e) {
            System.err.println("Error saving consumed message to local storage: " + e.getMessage());
        }
    }

    public void close() {
        consumer.close();
    }
}
登录后复制
登录后复制
user-signed-up-2024-11-04
order-created-2024-11-04
payment-processed-2024-11-04
登录后复制

代码说明
断路器：
Resilience4j 用于管理生产者的断路器电路。配置失败率阈值和打开状态下的等待时间。
本地持久化和DLQ：
失败的消息会保存到本地文件 (failed_messages.log) 和错误队列 (dead_letter_queue.log)。
错误处理：
在生产者和消费者中，错误被适当处理并记录。
DLQ 处理：
消费者在消费完主主题的消息后，还会处理存储在 DLQ 中的消息。
记录：
System.err 和 System.out 消息用于记录错误和重要事件。
最终考虑因素
通过此实现：
它确保消息以弹性方式发送和处理。
针对临时或持续错误提供适当的处理。
逻辑允许使用死信队列进行有效恢复。
断路器电路有助于防止系统在长时间故障时变得饱和。
这种方法创建了一个强大的系统，可以处理物理和逻辑问题，同时保持 Kafka 中消息的有序传递。

以上是如何获取和消费使用 Kafka 传递和订购的消息的详细内容。更多信息请关注PHP中文网其他相关文章！

本站声明
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序，用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

Video Face Swap

Video Face Swap

使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸！

显示更多

热门文章

如何修复KB5055612无法在Windows 10中安装？
4 周前 By DDD
<🎜>：泡泡胶模拟器无穷大 - 如何获取和使用皇家钥匙
4 周前 By 尊渡假赌尊渡假赌尊渡假赌
<🎜>：种植花园 - 完整的突变指南
3 周前 By DDD
北端：融合系统，解释
4 周前 By 尊渡假赌尊渡假赌尊渡假赌
Mandragora：巫婆树的耳语 - 如何解锁抓钩
3 周前 By 尊渡假赌尊渡假赌尊渡假赌
显示更多

热工具

记事本++7.3.1

记事本++7.3.1

好用且免费的代码编辑器

SublimeText3汉化版

SublimeText3汉化版

中文版，非常好用

禅工作室 13.0.1

禅工作室 13.0.1

功能强大的PHP集成开发环境

Dreamweaver CS6

Dreamweaver CS6

视觉化网页开发工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

显示更多

热门话题

Java教程
1676
14
CakePHP 教程
1429
52
Laravel 教程
1333
25
PHP教程
1278
29
C# 教程
1257
24
显示更多
Related knowledge
公司安全软件导致应用无法运行？如何排查和解决？ 公司安全软件导致应用无法运行？如何排查和解决？ Apr 19, 2025 pm 04:51 PM

公司安全软件导致部分应用无法正常运行的排查与解决方法许多公司为了保障内部网络安全，会部署安全软件。...

如何将姓名转换为数字以实现排序并保持群组中的一致性？ 如何将姓名转换为数字以实现排序并保持群组中的一致性？ Apr 19, 2025 pm 11:30 PM

将姓名转换为数字以实现排序的解决方案在许多应用场景中，用户可能需要在群组中进行排序，尤其是在一个用...

如何使用MapStruct简化系统对接中的字段映射问题？ 如何使用MapStruct简化系统对接中的字段映射问题？ Apr 19, 2025 pm 06:21 PM

系统对接中的字段映射处理在进行系统对接时，常常会遇到一个棘手的问题：如何将A系统的接口字段有效地映�...

如何优雅地获取实体类变量名构建数据库查询条件？ 如何优雅地获取实体类变量名构建数据库查询条件？ Apr 19, 2025 pm 11:42 PM

在使用MyBatis-Plus或其他ORM框架进行数据库操作时，经常需要根据实体类的属性名构造查询条件。如果每次都手动...

IntelliJ IDEA是如何在不输出日志的情况下识别Spring Boot项目的端口号的？ IntelliJ IDEA是如何在不输出日志的情况下识别Spring Boot项目的端口号的？ Apr 19, 2025 pm 11:45 PM

在使用IntelliJIDEAUltimate版本启动Spring...

Java对象如何安全地转换为数组？ Java对象如何安全地转换为数组？ Apr 19, 2025 pm 11:33 PM

Java对象与数组的转换：深入探讨强制类型转换的风险与正确方法很多Java初学者会遇到将一个对象转换成数组的�...

电商平台SKU和SPU数据库设计：如何兼顾用户自定义属性和无属性商品？ 电商平台SKU和SPU数据库设计：如何兼顾用户自定义属性和无属性商品？ Apr 19, 2025 pm 11:27 PM

电商平台SKU和SPU表设计详解本文将探讨电商平台中SKU和SPU的数据库设计问题，特别是如何处理用户自定义销售属...

如何利用Redis缓存方案高效实现产品排行榜列表的需求？ 如何利用Redis缓存方案高效实现产品排行榜列表的需求？ Apr 19, 2025 pm 11:36 PM

Redis缓存方案如何实现产品排行榜列表的需求？在开发过程中，我们常常需要处理排行榜的需求，例如展示一个�...

See all articles