首页 > 其他分享 >6. kafka案例--将控制台输入的每一行数据发送到Kafka中进行存储

6. kafka案例--将控制台输入的每一行数据发送到Kafka中进行存储

时间:2022-09-07 22:57:15浏览次数:84  
标签:-- Kafka apache key kafka import prop 数据

案例要求

/**
 * 要求将控制台输入的每一行数据发送到Kafka中进行存储
 *    输入的每一行数据形式如下
 *    s001  zs  20  男
 *    s002  ls  21  女
 *    要求将数据发送到我们的kafka的student主题中  同时要求发送的数据以学生编号为key,以学生信息为value形式进行数据方法
 *    同时要求student主题一共两个分区,其中编号0号分区存放性别为男的数据  编号1号分区存放性别为女的数据
 * 然后创建消费者去消费我们的Kafka主题中数据
 *   消费两个分区的数据即可
 *
 * 生产者的要求:
 *    1、生产者发送的数据是在控制台键盘输入的数据
 *    2、生产者在发送数据的时候自定义分区机制
 *    3、生产者生产的数据键值对类型的数据,以学生编号为key 以学生信息为value的数据
 *
 * 消费者的要求:
 *    正常能消费Kafka的student主题中的数据即可
 */

生产者

package practice;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;

/**
 * 要求将控制台输入的每一行数据发送到Kafka中进行存储
 *    输入的每一行数据形式如下
 *    s001  zs  20  男
 *    s002  ls  21  女
 *    要求将数据发送到我们的kafka的student主题中  同时要求发送的数据以学生编号为key,以学生信息为value形式进行数据方法
 *    同时要求student主题一共两个分区,其中编号0号分区存放性别为男的数据  编号1号分区存放性别为女的数据
 * 然后创建消费者去消费我们的Kafka主题中数据
 *   消费两个分区的数据即可
 *
 * 生产者的要求:
 *    1、生产者发送的数据是在控制台键盘输入的数据
 *    2、生产者在发送数据的时候自定义分区机制
 *    3、生产者生产的数据键值对类型的数据,以学生编号为key 以学生信息为value的数据
 *
 * 消费者的要求:
 *    正常能消费Kafka的student主题中的数据即可
 */
public class Producer {
    public static void main(String[] args) {
        // 1、定义生产者连接Kafka集群的配置项   key-value格式的
        Properties prop = new Properties();
        prop.put("bootstrap.servers", "192.168.200.111:9092,192.168.200.112:9092,192.168.200.113:9092");
        // key序列化
        prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        /**
         * 设置分区类的时候  要求类必须是public权限的(无法通过反射构建)
         * 第二个参数是全限定类名
         */
        prop.put("partitioner.class","practice.CustomPartition");

        // 2. 创建一个生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);


        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
        String line = null;
        try {
            while ((line = reader.readLine()) != null) {
                System.out.println(line);
                // s002  ls  21  女
                String[] str = line.split(" ");
                final String sno = str[0];
                ProducerRecord<String, String> record = new ProducerRecord<String, String>("demo", sno, line);
                producer.send(record, new Callback() {
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        System.out.println("当前这个数据的分区为:"+recordMetadata.partition() + "---offset:" + recordMetadata.offset());
                        System.out.println("当前的key为:" + sno + "-----主题为:" + recordMetadata.topic());
                    }
                });
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

自定义分区

package practice;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

public class CustomPartition implements Partitioner {
    public int partition(String s, Object key, byte[] bytes, Object value, byte[] bytes1, Cluster cluster) {
        String s1 = value.toString();
        String[] s2 = s1.split(" ");
        String sex = s2[s2.length-1];

        if (sex == "男") {
            return 0;
        } else {
            return 1;
        }
    }

    public void close() {

    }

    public void configure(Map<String, ?> map) {

    }
}

消费者

package practice;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class Customer {
    public static void main(String[] args) {
        Properties prop = new Properties();
        // 连接的kafka集群的地址
        prop.setProperty("bootstrap.servers","192.168.200.111:9092,192.168.200.112:9092,192.168.200.113:9092");
        // 设置一些消费者所属的消费者组
        prop.setProperty("group.id", "uek");
        // 是否自动确认offset  消费者再去消费某一个分区的时候,怎么知道消费者消息消费到哪个位置?
        prop.put("enable.auto.commit", "true");
        // 自动确认offset的时间间隔
        prop.put("auto.commit.interval.ms", "1000");
        // key的反序列化类
        prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // value的反序列化类
        prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);
        consumer.subscribe(Arrays.asList("demo"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(2000);
            for (ConsumerRecord<String, String> record: records) {
                System.out.println("当前消息的key为"+record.key());
                System.out.println("当前消息的value值为" + record.value());
                System.out.println("当前消息的分区为" + record.partition());
            }
        }
    }
}

标签:--,Kafka,apache,key,kafka,import,prop,数据
From: https://www.cnblogs.com/jsqup/p/16667602.html

相关文章

  • git之submodule
    当一个项目很大很复杂时,可以将项目分为几个模块分别进行管理;或者,当一个项目引用第三方开源代码,可以将这些第三方开源代码单独进行管理,这样做是为了代码隔离,方便项目维护。......
  • 最长公共子序列
    P1439【模板】最长公共子序列题目链接P1439LIS(LongestIncreasingSubsequence)(最长递增子序列)LCS(LongestCommonSubsequence)(最长公共子序列)简朴的DP求L......
  • 简单理解数据代理Object.defineProperty(obj,变量,{})
    简单理解Object.defineProperty()数据代理:通过一个对象代理对另一个对象中属性的操作(读写)eg:letobj1={x:1}letobj2={y:2}Object.defineProper......
  • sh /bin/bash^M: 坏的解释器: 没有那个文件或目录
    原因:由于windows系统下换行符为 \r\n,linux下换行符为 \n,所以导致在windows下编写的文件会比linux下多回车符号 \r。解决方案:[root@localhost~]#sed's/\r//'-i......
  • 【已解决】wordpress 修改固定链接 伪静态URL出现nginx 404错误
    一、站点设置 打开站点设置,选择伪静态,选择wordpress   二、wordpress设置打开wordpress后台,选择设置---》固定链接 选择一个你喜欢的格式点击保存 之......
  • zabbix自定义监控主从状态延迟与创建用户和组及权限管理
    1:zabbix监控主从部署mysql主从环境用到两台centos8的操作系统一台为master一台为slave[root@localhost~]#hostnamectlset-hostnamemaster[root@localhost~]#ba......
  • Learning Agent Communication under Limited Bandwidth by Message Pruning 记录
    文章介绍了一种提高多智能体之间通信效率的方法(得到最适合的通信带宽)。首先介绍了多智能体强化学习模型ACML: 算法跟MADDPG是有点类似的,增加了信息生成网络和信息协调......
  • 对象
    1.构造器构造器名字必须和类名相同,且必须没有返回值。java中使用new关键字来实例化一个对象。一个类中什么都不写,也会存在一个方法,即一个无参构造器用来实例化一个对象......
  • springboot 集成jimmer
    添加依赖plugins{//第一步:添加ksp插件id("com.google.devtools.ksp")version"1.7.10-1.0.6"...ommitotherplugins...}depdencies{/......
  • X86 assembly guide
    x86AssemblyGuideContents:Registers|MemoryandAddressing|Instructions|CallingConventionThisguidedescribesthebasicsof32-bitx86assemblylangu......