Kafka实现订单异步提交

阳光穿透心脏的1/2处 2023-10-15 16:58 31阅读 0赞

import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.TopicPartition;
    import org.apache.kafka.common.serialization.*;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Properties;
    
    public class OrderKafkaExample {
        private static final String TOPIC = "order-topic";
        private static final String BOOTSTRAP_SERVERS = "localhost:9092";
        private static final String GROUP_ID = "order-consumer-group";
    
        // 订单类
        public static class Order {
            private String orderId;
            private String product;
            // 省略其他属性和方法
    
            // 构造函数
            public Order(String orderId, String product) {
                this.orderId = orderId;
                this.product = product;
            }
    
            // Getter 和 Setter 方法
            // 省略其他 Getter 和 Setter 方法
        }
    
        // 订单序列化器
        public static class OrderSerializer implements Serializer<Order> {
            @Override
            public byte[] serialize(String topic, Order data) {
                // 将订单对象转换为字节数组
                // 省略具体实现
                return null;
            }
        }
    
        // 订单反序列化器
        public static class OrderDeserializer implements Deserializer<Order> {
            @Override
            public Order deserialize(String topic, byte[] data) {
                // 将字节数组转换为订单对象
                // 省略具体实现
                return null;
            }
        }
    
        public static void main(String[] args) {
            // 创建Kafka生产者以发送订单数据到指定的Topic
            sendOrder(new Order("order-001", "product-001"));
    
            // 创建一个Kafka消费者以处理已提交但未完成的订单
            processOrders();
        }
    
        private static void sendOrder(Order order) {
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, OrderSerializer.class.getName());
    
            // 创建生产者实例
            Producer<String, Order> producer = new KafkaProducer<>(props);
    
            // 创建要发送到Kafka的记录
            ProducerRecord<String, Order> record = new ProducerRecord<>(TOPIC, order.getOrderId(), order);
    
            // 发送记录到Kafka，并在回调函数中处理结果
            producer.send(record, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        exception.printStackTrace();
                    } else {
                        System.out.println("订单成功发送到Kafka，topic: " + metadata.topic() +
                                ", partition: " + metadata.partition() +
                                ", offset: " + metadata.offset());
                    }
                }
            });
    
            // 关闭生产者
            producer.close();
        }
    
        private static void processOrders() {
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, OrderDeserializer.class.getName());
    
            // 创建消费者实例
            KafkaConsumer<String, Order> consumer = new KafkaConsumer<>(props);
    
            // 订阅要消费的Topic
            consumer.subscribe(Collections.singletonList(TOPIC));
    
            while (true) {
                // 从Kafka拉取待处理的消息记录
                ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(100));
    
                for (ConsumerRecord<String, Order> record : records) {
                    // 异步处理订单逻辑
                    processOrder(record.value());
    
                    // 手动提交已处理的消息偏移量
                    TopicPartition topicPartition = new TopicPartition(TOPIC, record.partition());
                    OffsetAndMetadata offset = new OffsetAndMetadata(record.offset() + 1);
                    consumer.commitSync(Collections.singletonMap(topicPartition, offset));
                }
            }
        }
    
        private static void processOrder(Order order) {
            // 异步处理订单逻辑
        }
    }

**processOrder已经实现了异步提交订单，kafka的作用是什么?**

`processOrder` 方法实现了异步提交订单，而 Kafka 的作用是将订单数据持久化到分布式的消息队列中，以便后续其他消费者处理订单数据。

具体来说，`sendOrder` 方法会将订单数据封装成 Kafka 生产者的消息（`ProducerRecord`），并发送到指定的 Kafka Topic 中。而由于 Kafka 具有高吞吐量、持久化和可扩展性等特点，因此能够保证订单数据的高效、可靠和安全传输和存储。

另一方面，`processOrders` 方法通过创建 Kafka 消费者来订阅指定的 Topic，并拉取待处理的订单数据。在拉取到订单数据后，`processOrders` 会异步地调用 `processOrder` 方法来处理订单数据，从而实现了订单处理的异步化。而在处理完订单数据后，Kafka 消费者会手动提交偏移量（offset），以表示已经处理完该条消息记录，避免重复消费和数据丢失等问题。

综上所述，Kafka 作为一个分布式的消息队列，在订单系统中发挥着重要的作用，能够实现订单数据的可靠传输、存储和处理。

**如果不在processOrder方法里实现异步处理订单，通过processOrders能实现订单异步提交吗?**

如果`processOrder`方法本身没有实现异步处理订单，而只是在`processOrders`方法中调用`processOrder`方法的话，并不能实现订单的异步提交。

在现有的代码逻辑下，如果`processOrder`方法没有实现异步处理订单，那么在`processOrders`中的循环中，每次消费到一条订单消息后，会同步地调用`processOrder`方法处理该订单。也就是说，只有在当前订单处理完毕后，才会继续处理后续的订单。这种情况下，`processOrder`和`processOrders`之间的订单处理是同步执行的。

**如果processOrder方法实现了异步提交订单，不用kafka会出现什么问题?**

1.  可靠性问题：在异步提交订单的过程中，如果发生故障或错误，可能会导致订单数据丢失或处理失败。没有消息队列来持久化和保证消息传递的可靠性，系统无法确保所有订单都得到处理。
2.  数据传输效率降低：如果直接将异步提交的订单通过网络传输给下游服务或应用程序，可能会对网络带宽和延迟造成较大压力。这可能会导致订单处理的效率降低，并可能影响整个系统的性能。
3.  扩展性限制：使用Kafka等消息队列可以实现分布式和水平扩展，从而提高系统的处理能力和吞吐量。而如果不使用消息队列，订单处理可能会受限于单个处理节点的性能和资源，难以满足高并发的订单提交需求。
4.  无法处理消费者负载均衡：使用基于Kafka的消息队列可以很容易地实现多个消费者对订单进行并行处理，并且能够自动进行负载均衡。而如果不使用消息队列，需要自行实现负载均衡机制，以确保订单可以被多个消费者并发处理。

综上所述，使用消息队列（如Kafka）作为中间件可以解决上述问题，提供可靠的消息传递、强大的扩展性和高效的订单处理能力。尽管`processOrder`方法实现了异步提交订单，但没有消息队列可能会带来可靠性、效率和扩展性等方面的挑战。

**kafka中的消息被消掉还会占内存吗?**

在 Kafka 中，一旦消息被消费者成功消费，它将被认为是已处理并且会从 Kafka 的日志（或者称为主题分区）中移除。这意味着一旦消息被消费，它不再占用 Kafka 内存。

Kafka 的消息日志实际上是一种顺序追加的持久化存储机制，而不是一个传统意义上的队列。消费者可以按照自己的速度进行消费，并且消费进度是由消费者组管理和跟踪的。

当一条消息被消费者组中的所有消费者都消费完成后，Kafka 将确定该消息已经被成功处理，并使用相应的删除策略来清理该消息的磁盘存储。这样，即使消息被消费了，也不会在 Kafka 中继续占据内存空间。

需要注意的是，Kafka 的消息存储机制是基于磁盘的，因此在磁盘空间充足的情况下，即使消息被消费了，它仍然会保留在磁盘上一段时间，以便进行数据回溯或重新消费等操作。但是在内存中，已经被成功消费的消息将不再驻留。

总而言之，一旦消息被消费者成功消费，它将从 Kafka 中移除，不再占用内存空间。

**Java 原生的 Serializable 接口实现实体类序列化**

import java.io.Serializable;
    @Data
    public class Person implements Serializable {
        private String name;
        private int age;
    
    }
    
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.ByteArraySerializer;
    
    import java.util.Properties;
    
    public class Producer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", ByteArraySerializer.class.getName());
    
            KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);
    
            Person person = new Person("张三", 20);
            byte[] data = SerializationUtil.serialize(person);
    
            producer.send(new ProducerRecord<>("test_topic", data));
    
            producer.close();
        }
    }
    
    
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.ByteArrayDeserializer;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    import java.util.Collections;
    import java.util.Properties;
    
    public class Consumer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("key.deserializer", StringDeserializer.class.getName());
            props.put("value.deserializer", ByteArrayDeserializer.class.getName());
            props.put("group.id", "test_group");
    
            KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Collections.singleton("test_topic"));
    
            while (true) {
                ConsumerRecords<String, byte[]> records = consumer.poll(1000);
                for (ConsumerRecord<String, byte[]> record : records) {
                    Person person = (Person) SerializationUtil.deserialize(record.value());
                    System.out.println("name: " + person.getName() + ", age: " + person.getAge());
                }
            }
        }
    }
    
    /*在这个示例中，我们创建了一个名为 Person 的自定义类，并让它实现了 Java 原生的
     *Serializable 接口。在 Kafka 生产者端，我们使用了 ByteArraySerializer 将 Person 
     *类对象序列化为字节数组，并将字节数组作为消息发送到 Kafka 中。在消费者端，
     *我们使用了 ByteArrayDeserializer 反序列化接收到的字节数组，然后将其转换为 Person 对象。
     */