Linux安装kafka，Java发送kafka消息_java实现给kafka发送消息

kafka官方下载： https://kafka.apache.org/downloads

zookeeper官方下载：https://zookeeper.apache.org/releases.html

首先要保证有Java环境

将压缩包上传到Linux上，自己添加目录

解压缩

tar -zxvf 包名
1

安装zookeeper

1.添加环境变量

#编辑环境文件
vim /etc/profile

#添加内容
export ZOOKEEPER_HOME=zookeeper的安装路径
export export PATH=$PATH:${ZOOKEEPER_HOME}/bin

#完成后，生效环境配置
source /etc/profile
1
2
3
4
5
6
7
8
9

2.启动zookeeper
打开zookeeper的bin目录

zkServer.sh start
1

安装kafka

#编辑环境文件
vim /etc/profile

#添加内容
export KAFKA_HOME=kafka的安装路径
export PATH=$PATH:$KAFKA_HOME/bin

#完成后，生效环境配置
source /etc/profile
1
2
3
4
5
6
7
8
9

2.修改kafka的配置文件（重要）
listeners没配置或配置错误会导致后面Java程序发送的消息，服务器接收不到

#编辑配置文件
 vim config/server.properties

#修改内容
listeners=PLAINTEXT://0.0.0.0（或者本机的局域网IP）:9092
advertised.listeners=PLAINTEXT://公网IP:9092

#kafka信息在zookeeper中存放的位置
zookeeper.connect=localhost:2181/kafka
1
2
3
4
5
6
7
8
9

2.启动kafka
打开kafka的bin目录

./kafka-server-start.sh -daemon config/server.properties 
1

kafka的常用命令

//查看所有topic
kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list

//创建topic，分区（partitions）1，副本（replication）1
kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --topic first --create --partitions 1 --replication-factor 1

//查看topic详情
kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --topic first --describe

//生产者
kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic first

//消费者
kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic first
1
2
3
4
5
6
7
8
9
10
11
12
13
14

Java发送kafka消息

pom依赖

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>3.0.0</version>
</dependency>
1
2
3
4
5

生产者(Java)

package com.test.kafka.producer;

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 CustomProducer {

    public static void main(String[] args) {
        //0 配置
        Properties properties = new Properties();

        //连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");

        //指定对应的key和value的序列化 key.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //1 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        //2 发送数据
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<String, String>("first","test"+i));
        }

        //3 关闭资源
        kafkaProducer.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35