当前位置:首页 > Java技术 > kafka客户端实现---java版本

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

2022年09月16日 13:42:40Java技术4

转载请注明出处: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值序列化成字节。

转载请注明出处:kafka客户端java版本

参考链接

消费者示例

生产者示例

kafka官方文档
https://kafka.apache.org/documentation/#producerapi

参数配置Broker Configs
https://kafka.apache.org/documentation/#brokerconfigs

kafka 2.0.0 API 生产者文档

kafka 2.0.0 API 消费者文档

kafka 1.0.1 API文档

使用阿里云

confluent版本的示例

转载请注明出处:kafka客户端java版本

作者:张小凡vip
来源链接:https://blog.csdn.net/q383965374/article/details/83547169

版权声明:
1、JavaClub(https://www.javaclub.cn)以学习交流为目的,由作者投稿、网友推荐和小编整理收藏优秀的IT技术及相关内容,包括但不限于文字、图片、音频、视频、软件、程序等,其均来自互联网,本站不享有版权,版权归原作者所有。

2、本站提供的内容仅用于个人学习、研究或欣赏,以及其他非商业性或非盈利性用途,但同时应遵守著作权法及其他相关法律的规定,不得侵犯相关权利人及本网站的合法权利。
3、本网站内容原作者如不愿意在本网站刊登内容,请及时通知本站(javaclubcn@163.com),我们将第一时间核实后及时予以删除。


本文链接:https://www.javaclub.cn/java/42022.html

标签: Java版本
分享给朋友:

“kafka客户端实现---java版本” 的相关文章

Java空指针异常解决java.lang.NullPointerException解决心得

Java空指针异常解决java.lang.NullPointerException解决心得

今天做课设的时候运行程序报出以下错误 java.lang.NullPointerException 首先要理解的是此错误并不会在 程序中报错,只会在运行的时候报错。 是由于某个参数(集合,数组等数据)可能出现一个null值而导致后面的程序不能运行时...

Java实现Email发送

一、前言最近将项目的登录密码从图形验证码改为了短信验证码,同时也将忘记密码时长度进行了修改,在修改时,想到了之前在一些国外的网站上,使用过邮箱接收验证码的情况,故想到何妨不自己尝试整合一下Java程序发送邮件信息呢,所以动手整合了Email的发送实例。二、Email发送协议想要在互联网上提供电子邮件...

Java中四种访问修饰符的区别

在java中共有4种访问级别,按访问权限由高到低为:public(公有的)、protected(受保护的)、友好的(没有任何访问权限关键字修饰)和private(私有的)。 类型 类内部 同一个包其...

Java 日志框架详解

Java 日志框架详解

1. JUL学习 JUL全称Java util Logging是java原生的日志框架,使用时不需要另外引用第三方类库,相对其他日志框 架使用方便,学习简单,能够在小型应用中灵活使用。 1.1 架构介绍 Loggers...

冒泡排序的原理,思路,以及算法分析(Java实现)

冒泡排序的原理,思路,以及算法分析(Java实现)

冒泡排序 如果遇到相等的值不进行交换,那这种排序方式是稳定的排序方式。 1.原理:比较两个相邻的元素,将值大的元素交换到右边 2.思路:依次比较相邻的两个数,将比较小的数放在前面,比较大的数放在后面。 (1)第一次比较:首先比较第...

Java实现1到n的倒数的累加和

Java实现1到n的倒数的累加和

从键盘读入一个数,然后进行运算 实现代码: public static void main(String[] args) { Scanner in=new Scanner(System.in); int n ; n=in....

动车上的书摘-java对象流与序列化

动车上的书摘-java对象流与序列化

动车上的书摘-java对象流与序列化摘要: 摘要: 原创出处: http://www.cnblogs.com/Alandre/ 泥沙砖瓦浆木匠 希望转载,保留摘要,谢谢!钢笔不限贵便宜,书法是来自心对手的交流。-泥沙砖瓦浆木匠 一.对象序列化当需要存储相同类型的数据,选...

Java 基础:hashCode方法

Java 基础:hashCode方法

Writer:BYSocket(泥沙砖瓦浆木匠) 微博:BYSocket 豆瓣:BYSocket 一、前言     泥瓦匠最近被项目搞的天昏地暗。发现有些要给自己一些目标,关于技术的目标: 专注...

Java 容器 & 泛型:三、HashSet,TreeSet 和 LinkedHashSet比较

Java 容器 & 泛型:三、HashSet,TreeSet 和 LinkedHashSet比较

Writer:BYSocket(泥沙砖瓦浆木匠) 微博:BYSocket 豆瓣:BYSocket 上一篇总结了下ArrayList 、LinkedList和Vector比较,今天泥瓦匠总结下Hash 、LinkedList和Vector比较。其实大家都是...

Java实现ModbusTCP通信

Java实现ModbusTCP通信

使用ModbusTCP实现和硬件设备通信 有问题可以私信和评论,看到会回复。 一个项目,需要用Java实现使用ModbusTCP和硬件设备通信 视频地址:https://www.bilibili.com/video/BV1cz4y1R7cg...

发表评论

访客

◎欢迎参与讨论,请在这里发表您的看法和观点。