Java 连接 kafka
说明
1.本文以网易云 kafka 1.1.0 版本进行讲解,关于产品简介以及如何创建 kafka,请参考官网文档
2.本文bootstrap.servers
地址为 c-m1dvx2wwog.kafka.cn-east-1.internal:9092
3.由于 kafka 只能在可用区 B 创建,如果需要在本地调试,需要先在可用区 B 搭建 VPN 连接。OpenVPN 搭建,参考文档
如无法连接,请检查安全组和系统防火墙,请在对应的 VPC 安全组中放行内网访问 9092端口以及允许外网访问 OpenVPN 端口
关于外网连接 openvpn 无法 解析 kafka 域名的问题
我们连接 openvpn 后 ping c-m1dvx2wwog.kafka.cn-east-1.internal 是不行的,需要先在内网的机器上 ping 这个地址拿到 IP 后
[root@vpn ~]# ping c-m1dvx2wwog.kafka.cn-east-1.internal
PING c-m1dvx2wwog.kafka.cn-east-1.internal (192.168.10.154) 56(84) bytes of data.
64 bytes from 192.168.10.154 (192.168.10.154): icmp_seq=1 ttl=64 time=1.33 ms
64 bytes from 192.168.10.154 (192.168.10.154): icmp_seq=2 ttl=64 time=0.740 ms
64 bytes from 192.168.10.154 (192.168.10.154): icmp_seq=3 ttl=64 time=0.502 ms
^C
--- c-m1dvx2wwog.kafka.cn-east-1.internal ping statistics ---
3 packets transmitted, 3 received, 0% packet loss, time 2002ms
rtt min/avg/max/mdev = 0.502/0.859/1.337/0.352 ms
用这个 IP 去连接 kafka
创建一个 maven 工程
1.创建一个 maven 工程,在pom.xml 中加入
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>1.1.0</version>
</dependency>
</dependencies>
需要注意:kafka-clients的版本必须和kafka安装的版本一致
Producer demo
package KafkaService;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class KafkaProducerService {
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.10.154:9092");
properties.put("acks", "all");
properties.put("retries", 0);
properties.put("batch.size", 16384);
properties.put("linger.ms", 1);
properties.put("buffer.memory", 33554432);
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = null;
try {
producer = new KafkaProducer<String, String>(properties);
for (int i = 0; i < 100; i++) {
String msg = "Message " + i;
producer.send(new ProducerRecord<String, String>("test", msg));
System.out.println("Sent:" + msg);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
producer.close();
}
}
}
可以使用KafkaProducer类的实例来创建一个Producer,KafkaProducer类的参数是一系列属性值,下面分析一下所使用到的重要的属性:
bootstrap.servers
properties.put("bootstrap.servers", "192.168.1.110:9092");
bootstrap.servers
它是Kafka集群的IP地址,如果Broker数量超过1个,则使用逗号分隔,如”192.168.10.110:9092,192.168.10.110:9092”。
key.serializer&value.serializer
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
序列化类型。 Kafka消息是以键值对的形式发送到Kafka集群的,其中Key是可选的,Value可以是任意类型。但是在Message被发送到Kafka集群之前,Producer需要把不同类型的消
息序列化为二进制类型。本例是发送文本消息到Kafka集群,所以使用的是StringSerializer。
发送Message到Kafka集群
for (int i = 0; i < 100; i++) {
String msg = "Message " + i;
producer.send(new ProducerRecord<String, String>("test", msg));
System.out.println("Sent:" + msg);
}
上述代码会发送100个消息到test这个Topic
Topic 要在控制台创建
执行上述程序运行结果如下:
Consumer Demo
package KafkaService;
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class ConsumerDemo {
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.10.154:9092");
properties.put("group.id", "group-1");
properties.put("enable.auto.commit", "true");
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "earliest");
properties.put("session.timeout.ms", "30000");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
kafkaConsumer.subscribe(Arrays.asList("test"));
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, value = %s", record.offset(), record.value());
System.out.println();
}
}
}
}
可以使用KafkaConsumer类的实例来创建一个Consumer,KafkaConsumer类的参数是一系列属性值,下面分析一下所使用到的重要的属性:
bootstrap.servers:和Producer一样,是指向Kafka集群的IP地址,以逗号分隔。
group.id:Consumer分组ID
key.deserializer and value.deserializer
发序列化。Consumer把来自Kafka集群的二进制消息反序列化为指定的类型。因本例中的Producer使用的是String类型,所以调用StringDeserializer来反序列化
Consumer订阅了Topic为HelloWorld的消息,Consumer调用poll方法来轮循Kafka集群的消息,其中的参数100是超时时间(Consumer等待直到Kafka集群中没有消息为止):
在控制台找到对应的 Topic 可以看到消费者信息