kafka学习笔记_kafka messagelistener

安装kafka

下载

下载window的kafka地址

window的kafka只是为了方便学习

安装地址：kafka.apache.org/

安装

解压zip为文件夹

启动kafka

kafka服务器的功能相当于RocketMQ中的broker，kafka运行还需要一个类似于命名服务器的服务。在kafka安装目录中自带一个类似于命名服务器的工具，叫做zookeeper，它的作用是注册中心，相关知识请到对应课程中学习。

zookeeper-server-start.bat ..\..\config\zookeeper.properties		
//启动zookeeper

kafka-server-start.bat ..\..\config\server.properties				
//启动kafka

1
2
3
4
5
6

先启动zookeeper在启动kafka

管理员方式打开cmd,启动zookeeper之后启动kafka

运行bin目录下的windows目录下的zookeeper-server-start命令即可启动注册中心，默认对外服务端口2181。

另启动一个cmd窗口启动kafka

运行bin目录下的windows目录下的kafka-server-start命令即可启动kafka服务器，默认对外服务端口9092。

kafka命令操作

**注意根据kafka的版本来决定操作命令 版本，操作系统window或Linux等都决定kafka的命令操作 **

运行bin目录下的windows目录下的zookeeper-server-start命令即可启动注册中心，默认对外服务端口2181。

运行bin目录下的windows目录下的kafka-server-start命令即可启动kafka服务器，默认对外服务端口9092。

和之前操作其他MQ产品相似，kakfa也是基于主题操作，操作之前需要先初始化topic

# 创建topic
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic qil0820
# 查询topic
kafka-topics.bat --zookeeper 127.0.0.1:2181 --list					
# 删除topic
kafka-topics.bat --delete --zookeeper localhost:2181 --topic qil0820

1
2
3
4
5
6
7

windows下的kafka命令操作

创建一个生产者，在订阅一个消费者来进行监听消费

kafka-console-producer.bat --broker-list localhost:9092 --topic wslKafkaTest						
# 测试生产消息
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic wslKafkaTest --from-beginning	
# 测试消息消费

1
2
3
4
5

kafkaTool

Offset Explorer (kafkatool.com)

kafka整合SpringBoot

案例

生产者向Topic中发送消息

消费者通过指定监听Topic来进行消费消息

        <!--kafka的依赖-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
1
2
3
4
5

spring: 
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      group-id: wslKafkaTest // 设置默认的生产者消费者所属组id
1
2
3
4
5

生产者

@Api("Kafka生产者")
@RestController
public class KafkaProduceController {

    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;

    @ApiOperation("测试kafka发送消息")
    @PostMapping("/sendMessage")
    public void sendMessage(String str) {
        System.out.println("待发送短信的订单已纳入处理队列（kafka），id："+str);
        kafkaTemplate.send("wslKafkaTest",str);//使用send方法发送消息，需要传入topic名称
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

消费者

@Component
public class MessageListener {

    @KafkaListener(topics = "wslKafkaTest")
    public void onMessage(ConsumerRecord<String,String> record){
        System.out.println("已完成短信发送业务(kafka)，id："+record.value());
    }

}

1
2
3
4
5
6
7
8
9
10

kafka的常见Boot配置

spring:
  application:
    name: hello-kafka
  kafka:
    listener:
      #设置是否批量消费，默认 single（单条），batch（批量）
      type: single
    # 集群地址
    bootstrap-servers: 127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094
    # 生产者配置
    producer:
      # 重试次数
      retries: 3
      # 应答级别
      # acks=0 把消息发送到kafka就认为发送成功
      # acks=1 把消息发送到kafka leader分区，并且写入磁盘就认为发送成功
      # acks=all 把消息发送到kafka leader分区，并且leader分区的副本follower对消息进行了同步就任务发送成功
      acks: all
      # 批量处理的最大大小 单位 byte
      batch-size: 4096
      # 发送延时,当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
      buffer-memory: 33554432
      # 客户端ID
      client-id: hello-kafka
      # Key 序列化类
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # Value 序列化类
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 消息压缩：none、lz4、gzip、snappy，默认为 none。
      compression-type: gzip
      properties:
        partitioner:
          #指定自定义分区器
          class: top.zysite.hello.kafka.partitioner.MyPartitioner
        linger:
          # 发送延时,当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
          ms: 1000
        max:
          block:
            # KafkaProducer.send() 和 partitionsFor() 方法的最长阻塞时间 单位 ms
            ms: 6000
    # 消费者配置
    consumer:
      # 默认消费者组
      group-id: testGroup
      # 自动提交 offset 默认 true
      enable-auto-commit: false
      # 自动提交的频率 单位 ms
      auto-commit-interval: 1000
      # 批量消费最大数量
      max-poll-records: 100
      # Key 反序列化类
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # Value 反序列化类
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 当kafka中没有初始offset或offset超出范围时将自动重置offset
      # earliest:重置为分区中最小的offset
      # latest:重置为分区中最新的offset(消费分区中新产生的数据)
      # none:只要有一个分区不存在已提交的offset,就抛出异常
      auto-offset-reset: latest
      properties:
        interceptor:
          classes: top.zysite.hello.kafka.interceptor.MyConsumerInterceptor
        session:
          timeout:
            # session超时，超过这个时间consumer没有发送心跳,就会触发rebalance操作
            ms: 120000
        request:
          timeout:
            # 请求超时
            ms: 120000

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

kafka的常见工具类

生产者KafkaTemplate

// 根据生产者工厂构建kafkaTemplate
@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
    KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory);
    return kafkaTemplate;
}

// 将一个生产者工厂注册到spring容器中
@Bean
public ProducerFactory<String, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(kafkaProperties.buildProducerProperties());
}

1
2
3
4
5
6
7
8
9
10
11
12
13

KafkaTemplate提供了几个发送消息的接口如下：

topic：指定要发送消息的topic名称

partition：向指定的partition发送消息

key：消息的key

data：消息的data

timestamp：时间信息，一般默认为当前时间

record：ProducerRecord结构，是对key和value的一层封装，直接发送key和value，也会在内部被封装成ProducerRecord然后再发送出去

message：包含消息头（topic、partition、字符集)等信息和消息的封装格式

// 向默认的topic发送消息
public ListenableFuture<SendResult<K, V>> sendDefault(V data);
public ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
// 向指定的topic发送消息
public ListenableFuture<SendResult<K, V>> send(String topic, V data);
public ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
public ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
public ListenableFuture<SendResult<K, V>> send(Message<?> message)

1
2
3
4
5
6
7
8
9
10
11
12
13

SpringKafka消费者

Kafka消息监听器MessageListener

Kafka的消息监听一般可以分为：1.单条数据监听；2.批量数据监听。GenericMessageListener是SpringKafka的消息监听器接口，也是一个函数式接口，利用接口的onMessage方法可以实现消费数据。

public interface GenericMessageListener<T> {
	void onMessage(T data);
    
	default void onMessage(T data, @Nullable Acknowledgment acknowledgment) {
		throw new UnsupportedOperationException("Container should never call this");
	}

	default void onMessage(T data, Consumer<?, ?> consumer) {
		throw new UnsupportedOperationException("Container should never call this");
	}

	default void onMessage(T data, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
		throw new UnsupportedOperationException("Container should never call this");
	}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

基于此接口可以实现单条数据消息监听器接口MessageListenen、多条数据消息监听器接口BatchMessageListener、带ACK机制的消息监听器AcknowledgingMessageListener和BatchAcknowledgingMessageListener

消息监听容器与容器工厂

消息监听器MessageListener是由消费监听器容器MessageListenerContainer接口来承载，使用setupMessageListenner()方法启动一个监听器。其中还有定义了操作消息的resume()、pause()等方法。

public interface MessageListenerContainer extends SmartLifecycle {
    // 启动一个消息监听器
	void setupMessageListener(Object messageListener);
    // 获取消费者的指标信息
    Map<String, Map<MetricName, ? extends Metric>> metrics();
}

1
2
3
4
5
6
7

spring-kafka提供了两个容器KafkaMessageListenerContainer和ConcurrentMessageListenerContainer。

消息监听器容器由容器工厂KafkaListenerContainerFactory统一创建并管理

public interface KafkaListenerContainerFactory<C extends MessageListenerContainer> {
    // 根据endpoint创建监听器容器
    C createListenerContainer(KafkaListenerEndpoint endpoint);
    // 根据topic、partition和offset的配置
	C createContainer(TopicPartitionOffset... topicPartitions);
	// 根据topic创建监听器容器
    C createContainer(String... topics);
    // 根据topic的正则表达式创建监听器容器
	C createContainer(Pattern topicPattern);
}

1
2
3
4
5
6
7
8
9
10
11

spring-kafka提供了监听器容器工厂ConcurrentKafkaListenerContainerFactory，其有两个重要的配置

ContainerProperties和ConsumerFactory

ContainerProperties定义了要消费消息的topic，消息处理的MessageListener等信息。

因此要实现一个消息监听器的流程如下：

非注解式消费监听器

SpringKafka的消费者是由一个消费监听器容器ListenerConatiner去承载的，容器对应一个配置文件为ContainerProperties，ContainerProperties继承自消费者配置类ConsumerProperties，并且承载了消息监听器的设置

首先介绍非注解式的消息监听器，类似于ProducerFactory，消费者需要创建一个ConsumerFactory

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties());
}

1
2
3
4
5

然后建立监听器容器工厂ConcurrentKafkaListenerContainerFactory

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(KafkaConsts.DEFAULT_PARTITION_NUM);
    factory.setBatchListener(true);
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
}

1
2
3
4
5
6
7
8
9
10

有了容器工厂之后，就可以通过注册bean的方式生成一个MessageListenerContainer

@Bean
public KafkaMessageListenerContainer<String, String> kafkaMessageListenerContainer(
    ConsumerFactory<String, String> consumerFactory) {
    ContainerProperties containerProperties = new ContainerProperties("numb");
    containerProperties.setMessageListener(
        (MessageListener<String, String>) data -> System.out.println("收到消息： " + data.value()));
    return new KafkaMessageListenerContainer(consumerFactory, containerProperties);
}

1
2
3
4
5
6
7
8
9

在这个kafkaMessageListenerContainer中，通过ContainerProperties配置了消费的topic和messageListener。之后启动项目后，spring会将kafkaMessageListenerContainer注册到ConcurrentKafkaListenerContainerFactory中，这样获取到数据后会自动调用消息监听器进行数据处理。

测试消费者消费数据

@Test
public void test_send_and_consume() {
    ExecutorService threadPool = Executors.newCachedThreadPool();
    threadPool.submit(() -> {
        while (true) {
            kafkaTemplate.send(KafkaConsts.TOPIC_TEST, UUID.randomUUID().toString(), "kv");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("发送完成");
        }
    });
    while (true);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

输出：

发送完成
发送成功
收到消息： kv
发送完成
发送成功
收到消息： kv

1
2
3
4
5
6
7

注解式消费监听器@KafkaListener

之前配置了容器监听器工厂ConcurrentKafkaListenerContainerFactory之后，还需要用代码配置MessageListenerContainer， 指定消费的topic、消息监听器处理等。其实上面这步完全可以通过注解@KafkaListener实现。

@Component
@Slf4j
public class MessageHandler {
    @KafkaListener(topics = KafkaConsts.TOPIC_TEST, containerFactory = "kafkaListenerContainerFactory", id = "consumer_numb"
        // , topicPartitions = { @TopicPartition(topic = "numb", partitionOffsets = {@PartitionOffset(partition = "0", initialOffset="1")})}
        )
    public void handleMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            String message = (String) record.value();
            log.info("收到消息: {}", message);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            // 手动提交 offset
            acknowledgment.acknowledge();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

@KafkaListener的主要属性

• id：监听器的id - groupId：消费组id - idIsGroup：是否用id作为groupId，如果置为false，并指定groupId时，消费组ID使用groupId；否则默认为true，会使用监听器的id作为groupId - topics：指定要监听哪些topic(与topicPattern、topicPartitions 三选一) - topicPattern： 匹配Topic进行监听(与topics、topicPartitions 三选一) - topicPartitions： 显式分区分配，可以为监听器配置明确的主题和分区（以及可选的初始偏移量）

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })

1
2
3
4
5
6

• containerFactory：指定监听器容器工厂 - errorHandler： 监听异常处理器，配置BeanName - beanRef：真实监听容器的BeanName，需要在 BeanName前加 “__” - clientIdPrefix：消费者Id前缀 - concurrency： 覆盖容器工厂containerFactory的并发配置