kafka客户端实现---java版本

下载jar包

maven版本的直接找到相关的pom配置即可。
https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients/2.0.0 我们这里使用的是2.0.0版本
如下:

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.0.0</version>
</dependency>

编写消息生产类MsgProducer

注意，不要与kafka-client中的类重名，例如不要用KafkaProducer。

package com.biologic.util;

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

	public static void main(String[] args) {
		Properties props = new Properties();
		props.put("bootstrap.servers", "192.168.11.90:9092");
		props.put("acks", "all");
		props.put("retries", 0);
		props.put("batch.size", 16384);
		props.put("linger.ms", 1);
		props.put("buffer.memory", 33554432);
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

		Producer<String, String> producer = new KafkaProducer<String, String>(props);
		producer.send(new ProducerRecord<String, String>("test", "标题", "值"));
		producer.close();

	}

}

在kafka的服务器安装目录中运行监听consumer

bin/kafka-console-consumer.sh --bootstrap-server 192.168.11.90:9092 --topic test --from-beginning

运行main方法后接收到消息如下:
[外链图片转存失败(img-fJHRQIVm-1565856887677)(http://image.525.life/FuSEmUfx7ezaS-Cw5O-EmaU9pT-k)]

编写消息消费类MsgConsumer

注意，不要与kafka-client中的类重名，例如不要用KafkaConsumer。

package com.biologic.util;

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

	public static void main(String[] args) {
		 Properties props = new Properties();
	     props.put("bootstrap.servers", "192.168.11.90:9092");
	     props.put("group.id", "joe");
	     props.put("enable.auto.commit", "true");
	     props.put("auto.commit.interval.ms", "1000");
	     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
	     consumer.subscribe(Arrays.asList("test"));
	     while (true) {
	         ConsumerRecords<String, String> records = consumer.poll(100);
	         for (ConsumerRecord<String, String> record : records)
	             System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
	     }
	}
	
}

运行main方法后开始监听消息

在kafka的服务器安装目录中运行producer

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

输入如下:
[外链图片转存失败(img-nAhTzEbg-1565856887678)(http://image.525.life/FgsDWC29ylKxiaKsq8rM3yABa08R)]

java控制台输出如下:
[外链图片转存失败(img-swRJMAOI-1565856887679)(http://image.525.life/FuM95HM5_DQ3WByznteq-aeEIwHy)]

相关参数说明(kafka全部参数详见consumerconfigs):
bootstrap.servers
用于初始化时建立链接到kafka集群，以 host:port形式，多个以逗号分隔host1:port1,host2:port2；

group.id
消费者的组id
kafka使用消费者分组的概念来允许多个消费者共同消费和处理同一个topic中的消息。分组中消费者成员是动态维护的，如果一个消费者处理失败了，那么之前分配给它的partition将被重新分配给分组中其他消费者；同样，如果分组中加入了新的消费者，也将触发整个partition的重新分配，每个消费者将尽可能的分配到相同数目的partition，以达到新的均衡状态；

enable.auto.commit
用于配置是否自动的提交消费进度；

auto.commit.interval.ms
用于配置自动提交消费进度的时间；

session.timeout.ms
会话超时时长，客户端需要周期性的发送“心跳”到broker，这样broker端就可以判断消费者的状态，如果消费在会话周期时长内未发送心跳，那么该消费者将被判定为dead，那么它之前所消费的partition将会被重新的分配给其他存活的消费者；

key.serializer,value.serializer
说明了使用何种序列化方式将用户提供的key和vaule值序列化成字节。