1、消费者食用DEMO

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

Properties prop = new Properties();
prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
prop.put(ConsumerConfig.GROUP_ID_CONFIG, "testConsumer");
prop.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumerDemo");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
consumer.subscribe(Collections.singleton("test"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        String key = record.key();
        String value = record.value();
        System.err.println(record.toString());
    }
}

必要的参数

• bootstrap.servers

该参数为 broker 地址，不需要全部都填，因为 kafka 会从当前 broker 中获取其他 broker 信息。不过为了某个 broker 挂掉，一般填多个 broker 地址

• key.serializer

生产者发送消息的整体流程

消息追加器 RecordAccumulator

前面几个组件，在 3.1 的文章中，已经说清楚。现在来看 RecordAccumulator 组件

kafka 应用场景

1. 应用监控
2. 网站用户行为追踪
3. 流数据
4. 持久性日志

基本概念

在说基本概念前，先看一下 kafka 的系统架构

单机版 Docker 搭建

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

version: '2' 
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    depends_on: [ zookeeper ]
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"

参考官网。
当消息已过期或者无法再投递的时候，就会移动到死信队列（Dead Letter Queue）。
在以下情况，消息会重新发送到客户端：

• 使用事务的时候，调用了session的rollback方法
• 使用事务的时候，在会话关闭后再commit
• 使用CLIENT_ACKNOWLEDGE消息确认机制的时候，调用session的recover方法
• 客户端连接超时

当我们需要广播的时候，我们采用发布订阅模式。生产者往broker发送1，2，3，4，5，6，消费者1和消费者2都收到1，2，3，4，5，6。

当我们需要队列的时候，我们采用点对点模式，点对点模式中如果为了加快消息消费，我们就加多个消费者。生产者往broker发送1，2，3，4，5，6，消费者1收到1，3，5，消费者2收到2，4，6。

现在有个场景是这样的，在发布订阅模式中，由于生产者产生消息的速度超过了消费者1消费的速度，使消息一直积压，于是就想能不能既用发布订阅模式分发消息又用点对点模式加快消息消费呢？

activemq提供了一个虚拟topics的功能，首先元素的开头必须是VirtualTopic.开头的，消费端的格式是Consumer.<consumer name>.VirtualTopic.<VirtualTopicName>. ，比如生产者的是VirtualTopic.topic，那消费者A是Consumer.a.VirtualTopic.topic，a表示消费端组的名称，名字一样的话，会类似ptp一样消费。

消费者通过消息选择器来获取自己想要的消息，我们也可以通过另外通配符的方式，来获取自己想要的消息，当然两种还是有区别的。
通配符的方式有三种：

• “.”：用于分隔路径名字
• “*”：用于匹配路径的任何名字
• “>”：用于匹配末尾的名称

示例

消费者1消费a.b.c的消息，消费者2消费a.*.c的消息，所以中间是什么他不在乎，消费者3消费a.开头的消息，所以a.后面是什么他不在乎。生产者往a.b.c和a.b.c发送数据。