JAVA简单实现Kafka
首先需要先引入pom.xml依赖。<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>0.10.2.0</version></dependency>引入依赖后,实现生产
·
首先需要先引入pom.xml依赖。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.2.0</version>
</dependency>
引入依赖后,实现生产者和消费者代码
生产者
package com.mall.kafka.KafkaDemo;
import java.util.Properties;
import java.util.Random;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
/**
* 生产者
*
* @author AdminMall
*
*/
public class Producer {
public static String topic = "malltest";// 定义主题
public static void main(String[] args) throws InterruptedException {
Properties p = new Properties();
// kafka地址,多个地址用逗号分割
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
// 配置value的序列化类
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// 配置key的序列化类
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// 0表示不确认主服务器是否收到消息,马上返回,低延迟但最弱的持久性,数据可能会丢失
// 1表示确认主服务器收到消息后才返回,持久性稍强,可是如果主服务器死掉,从服务器数据尚未同步,数据可能会丢失
// -1表示确认所有服务器都收到数据,完美!
p.put("request.required.acks", "-1");
// 异步生产,批量存入缓存后再发到服务器去
p.put("producer.type", "async");
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(p);
try {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "测试消息");
kafkaProducer.send(record);
System.out.println("推送成功");
} finally {
// 必须关闭
kafkaProducer.close();
}
}
}
消费者
package com.mall.kafka.KafkaDemo;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.CommitFailedException;
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;
/**
* 消费者
*
* @author AdminMall
*
*/
public class Consumer {
public static void main(String[] args) {
Properties p = new Properties();
p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
// 自动提交秒
// p.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, false);
// 是否开启自动提交
p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
// group 代表一个消费组,加入组里面,消息只能被该组的一个消费者消费
// 如果所有消费者在一个组内,就是传统的队列模式,排队拿消息
// 如果所有的消费者都不在同一个组内,就是发布-订阅模式,消息广播给所有组
// 如果介于两者之间,那么广播的消息在组内也是要排队的
p.put(ConsumerConfig.GROUP_ID_CONFIG, "malltest");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
// 订阅消息;subscribe()
// 方法接受一个主题列表作为参数:也可以接收一个正则表达式,匹配多个主题;consumer.subscribe("test.*");
kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));
while (true) {
// 100 是超时时间(ms),在该时间内 poll 会等待服务器返回数据
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
// poll 返回一个记录列表。
// 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量,以及记录的键值对。
for (ConsumerRecord<String, String> record : records) {
// 主题
System.out.println("主题:" + record.topic());
System.out.println("读取位置:" + record.offset());
System.out.println("Key:" + record.key());
System.out.println("内容:" + record.value());
}
/ 手动提交
try {
// poll 的数据全部处理完提交
kafkaConsumer.commitAsync();
} catch (CommitFailedException e) {
e.printStackTrace();
}
}
}
}
更多推荐
已为社区贡献1条内容
所有评论(0)