您的位置:首页 > 娱乐 > 八卦 > 如何开自己的网购商城_在线设计的网站_游戏推广员一个月能赚多少_sem和seo的区别

如何开自己的网购商城_在线设计的网站_游戏推广员一个月能赚多少_sem和seo的区别

2024/10/5 19:15:10 来源:https://blog.csdn.net/qq_45115959/article/details/142497105  浏览:    关键词:如何开自己的网购商城_在线设计的网站_游戏推广员一个月能赚多少_sem和seo的区别
如何开自己的网购商城_在线设计的网站_游戏推广员一个月能赚多少_sem和seo的区别

目录

Kafka 生产与消费数据详解

 生产数据

 命令行操作

 工具操作

Java API

 消费数据

 命令行操作

Java API


Kafka 生产与消费数据详解

 生产数据

一旦消息主题创建完成,就可以通过Kafka客户端向Kafka服务器的主题中发送消息。Kafka生产者客户端是一套API接口,任何能够通过这些接口连接Kafka并发送数据的组件都可以称为Kafka生产者。以下是几种不同的生产数据方式:

 命令行操作
  1. 打开DOS窗口,进入E:\kafka_2.12-3.6.1\bin\windows目录。
  2. 在DOS窗口输入指令,进入生产者控制台。Kafka通过kafka-console-producer.bat文件进行消息生产者操作。必须传递的参数包括:
    • --bootstrap-server: 连接Kafka服务器,默认端口为9092,参数值为localhost:9092
    • --topic: 已经创建好的主题名称。
    指令如下:
    kafka-console-producer.bat --bootstrap-server localhost:9092 --topic test
  3. 控制台生产数据时,每次输入的数据需要回车确认才能发送到Kafka服务器。
 工具操作

对于需要更直观操作的情况,可以使用专门的工具进行快速访问,例如kafkatool_64bit.exe

  1. 安装该工具后,打开工具。
  2. 点击左上角按钮File -> Add New Connection...建立连接。
  3. 点击Test按钮测试连接。
  4. 增加连接后,按照工具中的步骤生产数据。
  5. 增加成功后,点击绿色箭头按钮进行查询,工具会显示当前数据。
Java API

通常,也可通过Java程序来生产数据。以下是在IDEA中使用Kafka Java API来生产数据的示例:

  1. 创建Kafka项目。
  2. 修改pom.xml文件,增加Maven依赖。
    <dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.1</version></dependency>
    </dependencies>
  3. 创建com.lzl.kafka.test.KafkaProducerTest类,并添加main方法以及生产者代码。
    package com.lzl.kafka.test;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import java.util.HashMap;
    import java.util.Map;public class KafkaProducerTest {public static void main(String[] args) {// 配置属性集合Map<String, Object> configMap = new HashMap<>();// 配置属性:Kafka服务器集群地址configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 配置属性:Kafka生产的数据为KV对,所以在生产数据进行传输前需要分别对K,V进行对应的序列化操作configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");// 创建Kafka生产者对象,建立Kafka连接KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);// 准备数据,定义泛型// 构造对象时需要传递 【Topic主题名称】,【Key】,【Value】三个参数ProducerRecord<String, String> record = new ProducerRecord<>("test", "key1", "value1");// 生产(发送)数据producer.send(record);// 关闭生产者连接producer.close();}
    }

 消费数据

一旦消息通过生产者客户端发送到了Kafka服务器,就可以通过消费者客户端对特定主题的消息进行消费。

 命令行操作
  1. 打开DOS窗口,进入E:\kafka_2.12-3.6.1\bin\windows目录。
  2. 输入指令,进入消费者控制台。使用kafka-console-consumer.bat文件进行消息消费者操作。必需的参数包括:
    • --bootstrap-server: 连接Kafka服务器,默认端口为9092。
    • --topic: 主题名称。
    • --from-beginning: 标记参数,确保从第一条数据开始消费。
    指令如下:
    kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
Java API

同样地,可以通过Java程序来消费数据。以下是在IDEA中使用Kafka Java API消费数据的示例:

  1. 创建Maven项目并增加Kafka依赖。
  2. 创建com.lzl.kafka.test.KafkaConsumerTest类,并添加main方法以及消费者代码。
    package com.lzl.kafka.test;
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import java.time.Duration;
    import java.util.Arrays;
    import java.util.Collections;
    import java.util.HashMap;
    import java.util.Map;public class KafkaConsumerTest {public static void main(String[] args) {// 配置属性集合Map<String, Object> configMap = new HashMap<String, Object>();// 配置属性:Kafka集群地址configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 配置属性: Kafka传输的数据为KV对,所以需要对获取的数据分别进行反序列化configMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");configMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");// 配置属性: 读取数据的位置 ,取值为earliest(最早),latest(最晚)configMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// 配置属性: 消费者组configMap.put("group.id", "lzl");// 配置属性: 自动提交偏移量configMap.put("enable.auto.commit", "true");// 创建消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configMap);// 消费者订阅指定主题的数据consumer.subscribe(Collections.singletonList("test"));while (true) {// 每隔100毫秒,抓取一次数据ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));// 打印抓取的数据for (ConsumerRecord<String, String> record : records) {System.out.println("K = " + record.key() + ", V = " + record.value());}}}
    }

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com