美文网首页
Java实现Kafka生产者和消费者的示例

Java实现Kafka生产者和消费者的示例

作者: 万猫学社 | 来源:发表于2022-03-18 11:24 被阅读0次

    Kafka简介

    Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写。Kafka的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。

    方式一:kafka-clients

    引入依赖

    在pom.xml文件中,引入kafka-clients依赖:

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.3.1</version>
    </dependency>
    

    生产者

    创建一个KafkaProducer的生产者实例:

    @Configuration
    public class Config {
    
        public final static String bootstrapServers = "127.0.0.1:9092";
    
        @Bean(destroyMethod = "close")
        public KafkaProducer<String, String> kafkaProducer() {
            Properties props = new Properties();
            //设置Kafka服务器地址
            props.put("bootstrap.servers", bootstrapServers);
            //设置数据key的序列化处理类
            props.put("key.serializer", StringSerializer.class.getName());
            //设置数据value的序列化处理类
            props.put("value.serializer", StringSerializer.class.getName());
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);
            return producer;
        }
    }
    

    在Controller中进行使用:

    @RestController
    @Slf4j
    public class Controller {
    
        @Autowired
        private KafkaProducer<String, String> kafkaProducer;
    
        @RequestMapping("/kafkaClientsSend")
        public String send() {
            String uuid = UUID.randomUUID().toString();
            RecordMetadata recordMetadata = null;
            try {
                //将消息发送到Kafka服务器的名称为“one-more-topic”的Topic中
                recordMetadata = kafkaProducer.send(new ProducerRecord<>("one-more-topic", uuid)).get();
                log.info("recordMetadata: {}", recordMetadata);
                log.info("uuid: {}", uuid);
            } catch (Exception e) {
                log.error("send fail, uuid: {}", uuid, e);
            }
            return uuid;
        }
    }
    

    消费者

    创建一个KafkaConsumer的消费者实例:

    @Configuration
    public class Config {
    
        public final static String groupId = "kafka-clients-group";
        public final static String bootstrapServers = "127.0.0.1:9092";
    
        @Bean(destroyMethod = "close")
        public KafkaConsumer<String, String> kafkaConsumer() {
            Properties props = new Properties();
            //设置Kafka服务器地址
            props.put("bootstrap.servers", bootstrapServers);
            //设置消费组
            props.put("group.id", groupId);
            //设置数据key的反序列化处理类
            props.put("key.deserializer", StringDeserializer.class.getName());
            //设置数据value的反序列化处理类
            props.put("value.deserializer", StringDeserializer.class.getName());
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
            //订阅名称为“one-more-topic”的Topic的消息
            kafkaConsumer.subscribe(Arrays.asList("one-more-topic"));
            return kafkaConsumer;
        }
    }
    

    在Controller中进行使用:

    @RestController
    @Slf4j
    public class Controller {
    
        @Autowired
        private KafkaConsumer<String, String> kafkaConsumer;
    
        @RequestMapping("/receive")
        public List<String> receive() {
            ////从Kafka服务器中的名称为“one-more-topic”的Topic中消费消息
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));
            List<String> messages = new ArrayList<>(records.count());
            for (ConsumerRecord<String, String> record : records.records("one-more-topic")) {
                String message = record.value();
                log.info("message: {}", message);
                messages.add(message);
            }
            return messages;
        }
    }
    

    方式二:spring-kafka

    使用kafka-clients需要我们自己创建生产者或者消费者的bean,如果我们的项目基于SpringBoot构建,那么使用spring-kafka就方便多了。

    引入依赖

    在pom.xml文件中,引入spring-kafka依赖:

    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.3.12.RELEASE</version>
    </dependency>
    

    生产者

    在application.yml文件中增加配置:

    spring:
      kafka:
        #Kafka服务器地址
        bootstrap-servers: 127.0.0.1:9092
        producer:
          #设置数据value的序列化处理类
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
    

    在Controller中注入KafkaTemplate就可以直接使用了,代码如下:

    @RestController
    @Slf4j
    public class Controller {
    
        @Autowired
        private KafkaTemplate<String, String> template;
    
        @RequestMapping("/springKafkaSend")
        public String send() {
            String uuid = UUID.randomUUID().toString();
            //将消息发送到Kafka服务器的名称为“one-more-topic”的Topic中
            this.template.send("one-more-topic", uuid);
            log.info("uuid: {}", uuid);
            return uuid;
        }
    }
    

    消费者

    在application.yml文件中增加配置:

    spring:
      kafka:
        #Kafka服务器地址
        bootstrap-servers: 127.0.0.1:9092
        consumer:
          #设置数据value的反序列化处理类
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    

    创建一个可以被Spring框架扫描到的类,并且在方法上加上@KafkaListener注解,就可以消费消息了,代码如下:

    @Component
    @Slf4j
    public class Receiver {
    
        @KafkaListener(topics = "one-more-topic", groupId = "spring-kafka-group")
        public void listen(ConsumerRecord<?, ?> record) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                String message = (String) kafkaMessage.get();
                log.info("message: {}", message);
            }
        }
    }
    

    最后,谢谢你这么帅,还给我点赞关注

    相关文章

      网友评论

          本文标题:Java实现Kafka生产者和消费者的示例

          本文链接:https://www.haomeiwen.com/subject/aeuidrtx.html