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类的参数是一系列属性值，下面分析一下所使用到的重要的属性：
发序列化。Consumer把来自Kafka集群的二进制消息反序列化为指定的类型。因本例中的Producer使用的是String类型，所以调用StringDeserializer来反序列化
Consumer订阅了Topic为HelloWorld的消息，Consumer调用poll方法来轮循Kafka集群的消息，其中的参数100是超时时间（Consumer等待直到Kafka集群中没有消息为止）：
在控制台找到对应的 Topic 可以看到消费者信息
文章作者：阿文
版权声明：本博客所有文章除特别声明外，均采用
CC BY-NC-SA 4.0 许可协议。转载请注明来自
阿文的博客！