快上网专注成都网站设计 成都网站制作 成都网站建设
成都网站建设公司服务热线:028-86922220

网站建设知识

十年网站开发经验 + 多家企业客户 + 靠谱的建站团队

量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决

Java如何实现Kafka生产者消费者

小编这次要给大家分享的是Java如何实现Kafka生产者消费者,文章内容丰富,感兴趣的小伙伴可以来了解一下,希望大家阅读完这篇文章之后能够有所收获。

创新互联公司服务项目包括宣恩网站建设、宣恩网站制作、宣恩网页制作以及宣恩网络营销策划等。多年来,我们专注于互联网行业,利用自身积累的技术优势、行业经验、深度合作伙伴关系等,向广大中小型企业、政府机构等提供互联网行业的解决方案,宣恩网站推广取得了明显的社会效益与经济效益。目前,我们服务的客户以成都为中心已经辐射到宣恩省份的部分城市,未来相信会继续扩大服务区域并继续获得客户的支持与信任!

Kafka的结构与RabbitMQ类似,消息生产者向Kafka服务器发送消息,Kafka接收消息后,再投递给消费者。
生产者的消费会被发送到Topic中,Topic中保存着各类数据,每一条数据都使用键、值进行保存。
每一个Topic中都包含一个或多个物理分区(Partition),分区维护着消息的内容和索引,它们有可能被保存在不同服务器。

新建一个Maven项目,pom.xml 加入依赖:

    
      org.apache.kafka
      kafka-clients
      2.3.0
    

1、编写生产者

将消息投递到Kafka服务器的名称为“topic1”的Topic中

package com.example.kafkatest;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服务器地址
    props.put("bootstrap.servers", "localhost:9092");
    //设置数据key和value的序列化处理类
    props.put("key.serializer", StringSerializer.class);
    props.put("value.serializer", StringSerializer.class);
    //创建生产者实例
    KafkaProducer producer = new KafkaProducer<>(props);
    ProducerRecord record = new ProducerRecord("topic1", "userName", "lc");
    //发送记录
    producer.send(record);
    producer.close();
  }
}

运行后,可打开命令行工具,进入Kafka目录,执行命令查询服务器的Topic:

bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

结果如下:

Java如何实现Kafka生产者消费者

2、编写消费者

本例中,消费者和生产者在同一个项目中,只是使用不同的启动类。

消费者会为自已指定一个消费者组的标识,每一条发布到Topic的记录,都会被交付给消费者组的一个消费者实例。

如果多个消费者实例有相同的消费者组,则这些记录会分配到各个消费者实例上,以达到负载均衡的目录。

如果所有的消费者有不同的消费者组,则每一条记录都会广播到全部的消费者进行处理。

package com.example.rabbittest;

import org.apache.kafka.clients.consumer.ConsumerRecord;
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.Arrays;
import java.util.Properties;

public class Consumer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服务器地址
    props.put("bootstrap.servers", "localhost:9092");
    //必须指定消费者组
    props.put("group.id", "test");
    //设置数据key和value的序列化处理类
    props.put("key.deserializer", StringDeserializer.class);
    props.put("value.deserializer", StringDeserializer.class);
    //创建消息者实例
    KafkaConsumer consumer = new KafkaConsumer<>(props);
    //订阅topic1的消息
    consumer.subscribe(Arrays.asList("topic1"));
    //到服务器中读取记录
    while (true){
      ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
      for(ConsumerRecord record : records){
        System.out.println("key:" + record.key() + "" + ",value:" + record.value());
      }
    }
  }
}

运行后,IDEA控制台其中输出如下:

Java如何实现Kafka生产者消费者

看完这篇关于Java如何实现Kafka生产者消费者的文章,如果觉得文章内容写得不错的话,可以把它分享出去给更多人看到。


分享标题:Java如何实现Kafka生产者消费者
文章源于:http://6mz.cn/article/pdoigg.html

其他资讯