时间:2021-05-20
Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写。Kafka的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。
引入依赖
在pom.xml文件中,引入kafka-clients依赖:
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.3.1</version></dependency>生产者
创建一个KafkaProducer的生产者实例:
@Configurationpublic class Config { public final static String bootstrapServers = "127.0.0.1:9092"; @Bean(destroyMethod = "close") public KafkaProducer<String, String> kafkaProducer() { Properties props = new Properties(); //设置Kafka服务器地址 props.put("bootstrap.servers", bootstrapServers); //设置数据key的序列化处理类 props.put("key.serializer", StringSerializer.class.getName()); //设置数据value的序列化处理类 props.put("value.serializer", StringSerializer.class.getName()); KafkaProducer<String, String> producer = new KafkaProducer<>(props); return producer; }}在Controller中进行使用:
@RestController@Slf4jpublic class Controller { @Autowired private KafkaProducer<String, String> kafkaProducer; @RequestMapping("/kafkaClientsSend") public String send() { String uuid = UUID.randomUUID().toString(); RecordMetadata recordMetadata = null; try { //将消息发送到Kafka服务器的名称为“one-more-topic”的Topic中 recordMetadata = kafkaProducer.send(new ProducerRecord<>("one-more-topic", uuid)).get(); log.info("recordMetadata: {}", recordMetadata); log.info("uuid: {}", uuid); } catch (Exception e) { log.error("send fail, uuid: {}", uuid, e); } return uuid; }}消费者
创建一个KafkaConsumer的消费者实例:
@Configurationpublic class Config { public final static String groupId = "kafka-clients-group"; public final static String bootstrapServers = "127.0.0.1:9092"; @Bean(destroyMethod = "close") public KafkaConsumer<String, String> kafkaConsumer() { Properties props = new Properties(); //设置Kafka服务器地址 props.put("bootstrap.servers", bootstrapServers); //设置消费组 props.put("group.id", groupId); //设置数据key的反序列化处理类 props.put("key.deserializer", StringDeserializer.class.getName()); //设置数据value的反序列化处理类 props.put("value.deserializer", StringDeserializer.class.getName()); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props); //订阅名称为“one-more-topic”的Topic的消息 kafkaConsumer.subscribe(Arrays.asList("one-more-topic")); return kafkaConsumer; }}在Controller中进行使用:
@RestController@Slf4jpublic class Controller { @Autowired private KafkaConsumer<String, String> kafkaConsumer; @RequestMapping("/receive") public List<String> receive() { 从Kafka服务器中的名称为“one-more-topic”的Topic中消费消息 ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1)); List<String> messages = new ArrayList<>(records.count()); for (ConsumerRecord<String, String> record : records.records("one-more-topic")) { String message = record.value(); log.info("message: {}", message); messages.add(message); } return messages; }}使用kafka-clients需要我们自己创建生产者或者消费者的bean,如果我们的项目基于SpringBoot构建,那么使用spring-kafka就方便多了。
引入依赖
在pom.xml文件中,引入spring-kafka依赖:
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.3.12.RELEASE</version></dependency>生产者
在application.yml文件中增加配置:
spring: kafka: #Kafka服务器地址 bootstrap-servers: 127.0.0.1:9092 producer: #设置数据value的序列化处理类 value-serializer: org.apache.kafka.common.serialization.StringSerializer在Controller中注入KafkaTemplate就可以直接使用了,代码如下:
@RestController@Slf4jpublic class Controller { @Autowired private KafkaTemplate<String, String> template; @RequestMapping("/springKafkaSend") public String send() { String uuid = UUID.randomUUID().toString(); //将消息发送到Kafka服务器的名称为“one-more-topic”的Topic中 this.template.send("one-more-topic", uuid); log.info("uuid: {}", uuid); return uuid; }}消费者
在application.yml文件中增加配置:
spring: kafka: #Kafka服务器地址 bootstrap-servers: 127.0.0.1:9092 consumer: #设置数据value的反序列化处理类 value-deserializer: org.apache.kafka.common.serialization.StringDeserializer创建一个可以被Spring框架扫描到的类,并且在方法上加上@KafkaListener注解,就可以消费消息了,代码如下:
@Component@Slf4jpublic class Receiver { @KafkaListener(topics = "one-more-topic", groupId = "spring-kafka-group") public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { String message = (String) kafkaMessage.get(); log.info("message: {}", message); } }}到此这篇关于Java实现Kafka生产者和消费者的示例的文章就介绍到这了,更多相关Java Kafka生产者和消费者 内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
Kafka的结构与RabbitMQ类似,消息生产者向Kafka服务器发送消息,Kafka接收消息后,再投递给消费者。生产者的消费会被发送到Topic中,Topi
1、生产者消费者模型作用和示例如下:1)通过平衡生产者的生产能力和消费者的消费能力来提升整个系统的运行效率,这是生产者消费者模型最重要的作用2)解耦,这是生产者
今天,我们给大家带来一篇如何利用Kafka的API进行客户端编程的文章,这篇文章很简单,就是利用Kafka的API创建一个生产者和消费者,生产者不断向Kafka
生产者消费者模型生产者:生产任务的个体;消费者:消费任务的个体;缓冲区:是生产者和消费者之间的媒介,对生产者和消费者解耦。当缓冲区元素为满,生产者无法生产,消费
本文实例讲述了Java生产者消费者模式。分享给大家供大家参考,具体如下:java的生产者消费者模式,有三个部分组成,一个是生产者,一个是消费者,一个是缓存。这么