SpringBoot实践之---集成Kafka

朱雀 2023-10-22 07:46 457阅读 0赞

1.maven依赖

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.12</artifactId>
        <version>2.1.0</version>
    </dependency>

2.application.properties配置

#kafka
    metadata.broker.list=172.16.3.31:6667,172.16.3.32:6667,172.16.3.33:6667
    request.required.acks=1
    kafka.topic.overdue=diplomat_userinfo_update

3.Kafka发送消息类

public class KafkaProducerUtils {
     
        /** kafka生产者对象 */
        private static Producer<String, String> producer;
     
        //属性初始化
        static {
            Properties prop = new Properties();
            prop.setProperty("serializer.class", "kafka.serializer.StringEncoder");
            prop.setProperty("metadata.broker.list", PropertyReaderUtils.getProValue("metadata.broker.list"));
            prop.setProperty("request.required.acks", PropertyReaderUtils.getProValue("request.required.acks"));
            ProducerConfig config = new ProducerConfig(prop);
            producer = new Producer(config);
        }
     
        /**
         * 发送单条kafka消息
         *
         * @param topic 消息topic枚举值
         * @param message json格式的消息内容
         */
        public static void sendMsg(EKafkaTopic topic, String message) {
            KeyedMessage<String, String> keyedMessage = new KeyedMessage<>(topic.getTopic(), message);
            producer.send(keyedMessage);
        }
     
        /**
         * 批量发送kafka消息
         *
         * @param topic 消息topic枚举值
         * @param messages json格式的消息内容列表
         */
        public static void batchSendMsg(EKafkaTopic topic, List<String> messages) {
            List<KeyedMessage<String, String>> keyedMessages = new ArrayList<>();
            for (String message : messages) {
                KeyedMessage<String, String> keyedMessage = new KeyedMessage<>(topic.getTopic(), message);
                keyedMessages.add(keyedMessage);
            }
            producer.send(keyedMessages);
        }
    }

4.Kafka消费消息类

import java.util.Arrays;
    import java.util.Properties;
     
    import org.apache.kafka.clients.consumer.Consumer;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
     
    public class ConsumerTest {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "master:9092,master:9093");
            props.put("group.id", "test");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     
            Consumer<String,String> consumer = new KafkaConsumer<String,String>(props);
            consumer.subscribe(Arrays.asList("message"));
            while(true){
                ConsumerRecords<String,String> records = consumer.poll(10);
                for(ConsumerRecord<String,String> record : records){
                    System.out.println("offset=" + record.offset() + ",--key=" + record.key() + ",--value=" + record.value());
                }
            }
        }
    }

[原文][Link 1]

[Link 1]: https://blog.csdn.net/luckykapok918/article/details/79374654