1、Kafka的定义#

Message Queue(MQ),消息队列中间件;

Kafka是一个分布式、支持分区的(partition)、多副本的 (replica),基于zookeeper协调的分布式消息系统,因其可水平扩展和高吞吐率而被广泛使用

2、在SpringBoot中使用Kafka#

1、引入依赖#

1
<dependency>
2
            <groupId>org.springframework.kafka</groupId>
3
            <artifactId>spring-kafka</artifactId>
4
</dependency>

2、编写配置文件#

1
server:
2
  port: 8081
3
spring:
4
  kafka:
5
    bootstrap-servers: 127.0.0.1:9092
6
    producer:
7
      retries: 3
8
      batch-size: 16384
9
      buffer-memory: 33554432
10
      acks: 1
11
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
12
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
13
    consumer:
14
      group-id: default-group
15
      enable-auto-commit: false
16
      auto-offset-reset: earliest
17
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
18
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
19
      max-poll-records: 500
20
    listener:
21
      # 当每一条记录被消费者监听器(ListenerConsumer)处理之后提交
22
      # RECORD
23
      # 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后提交
24
      # BATCH
25
      # 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后,距离上次提交时间大于TIME时提交
26
      # TIME
27
      # 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后,被处理record数量大于等于COUNT时提交
28
      # COUNT
29
      # TIME | COUNT 有一个条件满足时提交
30
      # COUNT_TIME
31
      # 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后, 手动调用Acknowledgment.acknowledge()后提交
32
      # MANUAL
33
      # 手动调用Acknowledgment.acknowledge()后立即提交,一般使用这种
34
      # MANUAL_IMMEDIATE
35
      ack-mode: MANUAL_IMMEDIATE

3、编写消息生产者#

1
@RestController
2
@RequestMapping("/msg")
3
public class MyKafkaController {
4
    private final static String TOPIC_NAME = "testA";
5

6
    @Autowired
7
    private KafkaTemplate<String,String> kafkaTemplate;
8

9
    @RequestMapping("/send")
10
    public String sendMessage(){
11
        kafkaTemplate.send(TOPIC_NAME,0,"key","this is a message!");
12
        return "send success!";
13
    }
14
}

4、编写消费者#

1
@Component
2
public class MyConsumer {
3
    @KafkaListener(topics = "testA")
4
    /*    @KafkaListener(groupId = "testGroup", topicPartitions = {
5
            @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
6
            @TopicPartition(topic = "topic2", partitions = "0",
7
                    partitionOffsets = @PartitionOffset(partition = "1",
8
                            initialOffset = "100"))
9
    },concurrency = "3")//concurrency就是同组下的消费者个数,就是并发消费数,建议小于等于分区总数*/
10
    public void listenGroup(ConsumerRecord<String, String> record,
11
                            Acknowledgment ack) {
12
        String value = record.value();
13
        System.out.println(value);
14
        System.out.println(record);
15
        //手动提交offset
16
        ack.acknowledge();
17
    }
18
}

音乐

音乐

1、Kafka的定义#

2、在SpringBoot中使用Kafka#

1、引入依赖#

2、编写配置文件#

3、编写消息生产者#

4、编写消费者#

支持与分享

音乐

文章目录

音乐

音乐

Springboot中使用kafka

1、Kafka的定义#

2、在SpringBoot中使用Kafka#

1、引入依赖#

2、编写配置文件#

3、编写消息生产者#

4、编写消费者#

支持与分享

音乐

文章目录