java集成kafka实例代码_Java

java集成kafka

要在 java 项目中集成 apache kafka 以实现消息的生产和消费，步骤如下：

1. 引入 maven 依赖

在您的 pom.xml 文件中添加以下依赖，以包含 kafka 客户端库：

<dependencies>
    <!-- kafka clients -->
    <dependency>
        <groupid>org.apache.kafka</groupid>
        <artifactid>kafka-clients</artifactid>
        <version>2.8.0</version>
    </dependency>
    <!-- 如果使用 spring boot，可添加以下依赖 -->
    <dependency>
        <groupid>org.springframework.kafka</groupid>
        <artifactid>spring-kafka</artifactid>
        <version>2.7.0</version>
    </dependency>
</dependencies>

2. 配置 kafka 生产者

首先，设置生产者的配置属性：

import org.apache.kafka.clients.producer.kafkaproducer;
import org.apache.kafka.clients.producer.producer;
import org.apache.kafka.clients.producer.producerconfig;
import org.apache.kafka.clients.producer.producerrecord;
import org.apache.kafka.common.serialization.stringserializer;

import java.util.properties;

public class kafkaproducerexample {
    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, stringserializer.class.getname());
        props.put(producerconfig.value_serializer_class_config, stringserializer.class.getname());

        // 创建生产者
        producer<string, string> producer = new kafkaproducer<>(props);

        // 发送消息
        for (int i = 0; i < 10; i++) {
            producerrecord<string, string> record = new producerrecord<>("your_topic", "key" + i, "value" + i);
            producer.send(record);
        }

        // 关闭生产者
        producer.close();
    }
}

3. 配置 kafka 消费者

接下来，设置消费者的配置属性，并订阅主题以消费消息：

import org.apache.kafka.clients.consumer.consumer;
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.common.serialization.stringdeserializer;

import java.time.duration;
import java.util.collections;
import java.util.properties;

public class kafkaconsumerexample {
    public static void main(string[] args) {
        // 配置属性
        properties props = new properties();
        props.put(consumerconfig.bootstrap_servers_config, "localhost:9092");
        props.put(consumerconfig.group_id_config, "your_group_id");
        props.put(consumerconfig.key_deserializer_class_config, stringdeserializer.class.getname());
        props.put(consumerconfig.value_deserializer_class_config, stringdeserializer.class.getname());

        // 创建消费者
        consumer<string, string> consumer = new kafkaconsumer<>(props);

        // 订阅主题
        consumer.subscribe(collections.singletonlist("your_topic"));

        // 持续消费消息
        try {
            while (true) {
                consumerrecords<string, string> records = consumer.poll(duration.ofmillis(100));
                records.foreach(record -> {
                    system.out.printf("consumed message: key = %s, value = %s, offset = %d%n",
                            record.key(), record.value(), record.offset());
                });
            }
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}

4. 使用 spring boot 集成 kafka

如果您使用 spring boot，可以通过配置 kafkatemplate（用于生产消息）和使用 @kafkalistener 注解（用于消费消息）来简化 kafka 的集成。

生产者配置：

import org.apache.kafka.clients.producer.producerconfig;
import org.apache.kafka.common.serialization.stringserializer;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.kafka.core.defaultkafkaproducerfactory;
import org.springframework.kafka.core.kafkatemplate;
import org.springframework.kafka.core.producerfactory;

import java.util.hashmap;
import java.util.map;

@configuration
public class kafkaproducerconfig {

    @bean
    public producerfactory<string, string> producerfactory() {
        map<string, object> configprops = new hashmap<>();
        configprops.put(producerconfig.bootstrap_servers_config, "localhost:9092");
        configprops.put(producerconfig.key_serializer_class_config, stringserializer.class.getname());
        configprops.put(producerconfig.value_serializer_class_config, stringserializer.class.getname());
        return new defaultkafkaproducerfactory<>(configprops);
    }

    @bean
    public kafkatemplate<string, string> kafkatemplate() {
        return new kafkatemplate<>(producerfactory());
    }
}

使用 kafkatemplate 发送消息：

import org.springframework.beans.factory.annotation.autowired;
import org.springframework.kafka.core.kafkatemplate;
import org.springframework.stereotype.service;

@service
public class kafkaproducerservice {

    @autowired
    private kafkatemplate<string, string> kafkatemplate;

    public void sendmessage(string topic, string key, string value) {
        kafkatemplate.send(topic, key, value);
    }
}

消费者配置：

import org.apache.kafka.clients.consumer.consumerconfig;
import org.apache.kafka.common.serialization.stringdeserializer;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.kafka.annotation.enablekafka;
import org.springframework.kafka.config.concurrentkafkalistenercontainerfactory;
import org.springframework.kafka.core.consumerfactory;
import org.springframework.kafka.core.defaultkafkaconsumerfactory;

import java.util.hashmap;
import java.util.map;

@enablekafka
@configuration
public class kafkaconsumerconfig {

    @bean
    public consumerfactory<string, string> consumerfactory() {
        map<string, object> props = new hashmap<>();
        props.put(consumerconfig.bootstrap_servers_config, "localhost:9092");
        props.put(consumerconfig.group_id_config, "your_group_id");
        props.put(consumerconfig.key_deserializer_class_config, stringdeserializer.class.getname());
        props.put(consumerconfig.value_deserializer_class_config, stringdeserializer.class.getname());
        return new defaultkafkaconsumerfactory<>(props);
    }

    @bean
    public concurrentkafkalistenercontainerfactory<string, string> kafkalistenercontainerfactory() {
        concurrentkafkalistenercontainerfactory<string, string> factory = new concurrentkafkalistenercontainerfactory<>();
        factory.setconsumerfactory(consumerfactory());
        return factory;
    }
}

使用 @kafkalistener 消费消息：

在 spring boot 中，@kafkalistener 注解用于监听指定的 kafka 主题，并在收到消息时触发相应的方法。

以下是一个基本示例：

import org.springframework.kafka.annotation.kafkalistener;
import org.springframework.stereotype.service;

@service
public class kafkaconsumerservice {

    @kafkalistener(topics = "your_topic", groupid = "your_group_id")
    public void listen(string message) {
        system.out.println("received message: " + message);
        // 在此处添加处理逻辑
    }
}

在上述代码中：

topics：指定要监听的 kafka 主题。
groupid：指定消费者组 id。

listen 方法：当有新消息发布到指定主题时，该方法会被调用，message 参数包含消息的内容。

批量消费消息

如果希望一次处理多条消息，可以启用批量监听。

首先，需要配置一个支持批量消费的 kafkalistenercontainerfactory：

import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.kafka.annotation.enablekafka;
import org.springframework.kafka.config.concurrentkafkalistenercontainerfactory;
import org.springframework.kafka.core.consumerfactory;

@enablekafka
@configuration
public class kafkaconsumerconfig {

    @bean
    public concurrentkafkalistenercontainerfactory<string, string> kafkalistenercontainerfactory(
            consumerfactory<string, string> consumerfactory) {
        concurrentkafkalistenercontainerfactory<string, string> factory =
                new concurrentkafkalistenercontainerfactory<>();
        factory.setconsumerfactory(consumerfactory);
        factory.setbatchlistener(true); // 启用批量监听
        return factory;
    }
}

然后，在消费者服务中使用 @kafkalistener 注解，并指定使用上述配置的工厂：

import org.springframework.kafka.annotation.kafkalistener;
import org.springframework.stereotype.service;
import java.util.list;

@service
public class kafkabatchconsumerservice {

    @kafkalistener(
        topics = "your_topic",
        groupid = "your_group_id",
        containerfactory = "kafkalistenercontainerfactory"
    )
    public void listen(list<string> messages) {
        system.out.println("received batch messages: " + messages);
        // 在此处添加批量处理逻辑
    }
}

在上述代码中：

containerfactory：指定使用支持批量消费的工厂。

listen 方法的参数类型为 list<string>，用于接收一批消息。

控制消费者的启动和停止

在某些情况下，可能需要在运行时控制 kafka 消费者的启动和停止。

可以通过 kafkalistenerendpointregistry 来实现：

import org.springframework.beans.factory.annotation.autowired;
import org.springframework.kafka.listener.kafkalistenerendpointregistry;
import org.springframework.kafka.listener.messagelistenercontainer;
import org.springframework.scheduling.annotation.scheduled;
import org.springframework.stereotype.service;

@service
public class kafkalistenermanager {

    @autowired
    private kafkalistenerendpointregistry registry;

    // 启动监听器
    public void startlistener(string listenerid) {
        messagelistenercontainer listenercontainer = registry.getlistenercontainer(listenerid);
        if (listenercontainer != null && !listenercontainer.isrunning()) {
            listenercontainer.start();
        }
    }

    // 停止监听器
    public void stoplistener(string listenerid) {
        messagelistenercontainer listenercontainer = registry.getlistenercontainer(listenerid);
        if (listenercontainer != null && listenercontainer.isrunning()) {
            listenercontainer.stop();
        }
    }
}

在上述代码中：

startlistener 方法用于启动指定的监听器。
stoplistener 方法用于停止指定的监听器。
listenerid 对应于 @kafkalistener 注解中的 id 属性。

通过这种方式，可以在应用运行时根据需要动态地控制 kafka 消费者的行为。

通过上述配置和代码示例，可以在 spring boot 项目中有效地集成 kafka，实现消息的生产和消费功能。

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持代码网。

java集成kafka实例代码

2024年12月31日 • Java •我要评论