虚拟机中jar包数据到kafka主题ods_base_log

在虚拟机中安装Kafka，并启动Kafka服务。

在虚拟机中创建一个Java项目，将需要发送到Kafka的数据封装成对象。

在Java项目中引入Kafka的依赖，如下所示：

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

在Java项目中编写生产者代码，将数据封装成Kafka消息，发送到Kafka主题。示例代码如下：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class KafkaProducerExample {
    private static final String KAFKA_TOPIC = "ods_base_log";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 将需要发送到Kafka的数据封装成对象
        BaseLog baseLog = new BaseLog();
        baseLog.setTimestamp(System.currentTimeMillis());
        baseLog.setEvent("click");
        baseLog.setPage("home");
        baseLog.setUserId("123");

        // 将数据封装成Kafka消息，发送到Kafka主题
        ProducerRecord<String, String> record = new ProducerRecord<>(KAFKA_TOPIC, baseLog.toString());
        producer.send(record);

        producer.close();
    }
}

在Kafka主题中查看是否已经接收到数据，可以使用Kafka自带的命令行工具来查看，如下所示：

# 查看所有主题
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看指定主题中的消息
bin/kafka-console-consumer.sh --topic ods_base_log --from-beginning --bootstrap-server localhost:9092
``