我是廖志伟，一名Java开发工程师、Java领域优质创作者、CSDN博客专家、幕后大佬社区创始人。拥有多年一线研发经验，研究过各种常见框架及中间件的底层源码，对于大型分布式、微服务、三高架构（高性能、高并发、高可用）有过实践架构经验。
博主：java_wxid
博主：Java廖志伟
社区：幕后大佬
文章目录
- Java客户端访问Kafka
- 引入maven依赖
- 消息发送端代码
- 消息接收端代码
本文内容： Java客户端访问Kafka
引入maven依赖
消息发送端代码org.apache.kafka kafka-clients2.4.1
package com.sky.kafka.kafkaDemo;
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class MsgProducer {
private final static String TOPIC_NAME = "my-replicated-topic";
public static void main(String[] args) throws InterruptedException, ExecutionException {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
//把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
Producer producer = new KafkaProducer(props);
int msgNum = 5;
final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
for (int i = 1; i <= msgNum; i++) {
Order order = new Order(i, 100 + i, 1, 1000.00);
//指定发送分区
//未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
ProducerRecord producerRecord = new ProducerRecord(TOPIC_NAME
, order.getOrderId().toString(), JSON.toJSonString(order));
//等待消息发送成功的同步阻塞方法
//异步回调方式发送消息
producer.send(producerRecord, new Callback() {
public void onCompletion(Recordmetadata metadata, Exception exception) {
if (exception != null) {
System.err.println("发送消息失败：" + exception.getStackTrace());
}
if (metadata != null) {
System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
+ metadata.partition() + "|offset-" + metadata.offset());
}
countDownLatch.countDown();
}
});
//送积分 TODO
}
countDownLatch.await(5, TimeUnit.SECONDS);
producer.close();
}
}
消息接收端代码
package com.sky.kafka.kafkaDemo;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
public class MsgConsumer {
private final static String TOPIC_NAME = "my-replicated-topic";
private final static String CONSUMER_GROUP_NAME = "testGroup";
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094");
// 消费分组名
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
// 是否自动提交offset，默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 自动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
//props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
//一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaConsumer consumer = new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(TOPIC_NAME));
// 消费指定分区
//consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
//消息回溯消费
//指定offset消费
//从指定时间点开始消费
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
record.offset(), record.key(), record.value());
}
}
}
}
总结
以上就是今天要讲的内容，还希望各位读者大大能够在评论区积极参与讨论，给文章提出一些宝贵的意见或者建议，合理的内容，我会采纳更新博文，重新分享给大家。
四连 关注点赞收藏⭐️留言
感谢大家的支持，用心写博文分享给大家，你的支持（点赞收藏⭐️留言）是对我创作的最大帮助。
微信公众号：南北踏尘
主页地址：java_wxid
社区地址：幕后大佬
给读者大大的话
我本身是一个很普通的程序员，放在人堆里，除了与生俱来的盛世美颜、所剩不多的发量，就剩下180的大高个了。就是我这样的一个人，默默坚持写博文也有好多年了，有句老话说的好，牛逼之前都是傻逼式的坚持。希望自己可以通过大量的作品，时间的积累，个人魅力、运气和时机，可以打造属于自己的技术影响力。同时也希望自己可以成为一个懂技术，懂业务，懂管理的综合型人才，作为项目架构路线的总设计师，掌控全局的团队大脑，技术团队中的绝对核心是我未来几年不断前进的目标。
提示：以下都是资源分享，求个一键三连。
面试资料
福利大放送，欢迎关注点赞收藏⭐️留言，拜托了，这对我真的很重要。
点击：面试资料
提取码：2021
200套PPT模板
福利大放送，欢迎关注点赞收藏⭐️留言，拜托了，这对我真的很重要。
点击：200套PPT模板
提取码：2021
提问的智慧
福利大放送，欢迎关注点赞收藏⭐️留言，拜托了，这对我真的很重要。
点击：提问的智慧
提取码：2021
Java开发学习路线
P5学习路线图
P6学习路线图
P7学习路线图
P8学习路线图
以上四张图详细介绍了作为Java开发工作者所需要具备的知识技能，同学们学废了嘛，有想法系统学习的同学可以私聊我，欢迎关注点赞收藏⭐️留言。
博主：java_wxid
博主：Java廖志伟
社区：幕后大佬
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）