kafka template操作kafka
依赖<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-streams</artifactId></dependency><dependency><groupId>org.springframework.kafka<
依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
配置文件
spring.kafka.listener.concurrency=3
在spring-kafka在运行时会启动两类线程,一类是Consumer线程,另一类是Listener线程。前者用来直接调用kafka-client的poll()方法获取消息,后者才是调用我们代码中标有@KafkaListener注解方法的线程。如果直接使用kafka-client的话,那么正常的写法是一个while循环,在循环里面调用poll(),然后处理消息,这在kafka broker看来就是一个Consumer。如果你想用多个Consumer, 除了多启动几个进程以外,也可以在一个进程使用多个线程执行此while()循环。spring-kafka就是这么干的。
对于spring.kafka.listener.concurrency=3这个参数来说,它设置的是每个@KafkaListener的并发个数。每添加一个@KafkaListener, spring-kafka都会启动一条Consumer线程来监听这些topic(注解可以指定监听多个topic), 然后再启动spring.kafka.listener.concurrency条线程来真正执行你的Listener
此段来自:https://blog.csdn.net/dshf_1/article/details/103203920
server:
port: 8080
spring:
kafka:
bootstrap-servers: ip:9092
consumer:
group-id: test-consumer
producer:
batch-size: 16384
buffer-memory: 33554432
key-serializer: org.apache.kafka.common.serialization.StringSerializer
retries: 0
value-serializer: org.apache.kafka.common.serialization.StringSerializer
provder
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* @author: zhangxirui
* @date: 2021-02-07 9:27
* @version: 1.0
*/
@Component
@EnableScheduling
public class KafkaProvder {
@Autowired
private KafkaTemplate kafkaTemplate;
@Scheduled(cron = "* * * * * ?")
public void producerr() {
for (int i = 0; i < 10; i++) {
kafkaTemplate.send(new ProducerRecord<String, String>("my-topic", "hello" + i, "java" + i));
}
}
}
listen
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
* @author: zhangxirui
* @date: 2021-02-07 9:27
* @version: 1.0
*/
@Component
public class KafkaListenerr {
@KafkaListener(topics = {"my-topic"})
public void listen(ConsumerRecord<?, ?> record) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
if (kafkaMessage.isPresent()) {
System.out.println("消息接收成功");
Object message = kafkaMessage.get();
System.out.println("record =" + record);
System.out.println("message =" + message);
}
}
}
更多推荐
所有评论(0)