Apache Kafka-通过设置Consumer Group实现广播模式

偏执的太偏执、 2022-10-30 07:25 546阅读 0赞

### 文章目录 ###

*  概述
 *  Code
 *   *  POM依赖
     *  配置文件
     *  生产者
     *  消费者
     *  单元测试
     *  测速结果
 *  源码地址

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcnRpc2FuLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]

--------------------

# 概述 #

传统的消息传递模式有2种：队列( queue) 和（publish-subscribe）

*  queue模式：多个consumer从服务器中读取数据，消息只会到达一个consumer
 *  publish-subscribe模式：消息会被广播给所有的consumer

Kafka基于这2种模式提供了一种consumer的抽象概念： consumer group

*  queue模式：所有的consumer都位于同一个consumer group 下。
 *  publish-subscribe模式：**所有的consumer都有着自己唯一的consumer group**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70]

说明： 由2个broker组成的kafka集群，某个主题总共有4个partition(P0-P3)，分别位于不同的broker上。这个集群由2个Consumer Group消费， A有2个consumer instances ，B有4个。

通常一个topic会有几个consumer group，每个consumer group都是一个逻辑上的订阅者（ logicalsubscriber ）。每个consumer group由多个consumer instance组成，从而达到可扩展和容灾的功能。

--------------------

广播模式的应用 ----> 应用里缓存了数据字典等配置表在内存中，可以通过 Kafka 广播消费，实现每个应用节点都消费消息，刷新本地内存的缓存。

--------------------

# Code #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 1]

## POM依赖 ##

<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.springframework.kafka</groupId>
    			<artifactId>spring-kafka</artifactId>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<scope>test</scope>
    		</dependency>
    	</dependencies>

--------------------

## 配置文件 ##

spring:
      # Kafka 配置项，对应 KafkaProperties 配置类
      kafka:
        bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
        # Kafka Producer 配置项
        producer:
          acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
          retries: 3 # 发送失败时，重试发送的次数
          key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化
          value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
    
        # Kafka Consumer 配置项
        consumer:
          auto-offset-reset: latest # 在广播订阅下，一般情况下，无需消费历史的消息，而是从订阅的 Topic 的队列的尾部开始消费即可
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
          properties:
            spring:
              json:
                trusted:
                  packages: com.artisan.springkafka.domain
        # Kafka Consumer Listener 监听器配置
        listener:
          missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
    
    logging:
      level:
        org:
          springframework:
            kafka: ERROR # spring-kafka
          apache:
            kafka: ERROR # kafka

`auto-offset-reset: latest` 广播模式，一般情况下，无需消费历史的消息，从订阅的 Topic 的队列的尾部开始消费即可

--------------------

## 生产者 ##

package com.artisan.springkafka.producer;
    
    import com.artisan.springkafka.constants.TOPIC;
    import com.artisan.springkafka.domain.MessageMock;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.stereotype.Component;
    import org.springframework.util.concurrent.ListenableFuture;
    
    import java.util.Random;
    import java.util.concurrent.ExecutionException;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:25 * @mark: show me the code , change the world */
    
    @Component
    public class ArtisanProducerMock { 
    
    
        @Autowired
        private KafkaTemplate<Object,Object> kafkaTemplate ;
    
    
        /** * 异步发送 * @return * @throws ExecutionException * @throws InterruptedException */
        public ListenableFuture<SendResult<Object, Object>> sendMsgASync() throws ExecutionException, InterruptedException { 
            // 模拟发送的消息
            Integer id = new Random().nextInt(100);
            MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);
            // 异步发送消息
            ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(TOPIC.TOPIC, messageMock);
            return result ;
    
        }
    
    }

--------------------

## 消费者 ##

package com.artisan.springkafka.consumer;
    
    import com.artisan.springkafka.domain.MessageMock;
    import com.artisan.springkafka.constants.TOPIC;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:33 * @mark: show me the code , change the world */
    
    @Component
    public class ArtisanCosumerMockDiffConsumeGroup { 
    
    
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        private static final String CONSUMER_GROUP_PREFIX = "MOCK-B" ;
    
        @KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC +  "-" + "#{T(java.util.UUID).randomUUID()})")
        public void onMessage(MessageMock messageMock){ 
            logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock);
        }
    
    }

注意: groupId 通过 Spring EL 表达式，在每个消费者分组的名字上配合 UUID 生成其后缀。这样，就能保证每个项目启动的消费者分组不同，从而达到广播消费的目的。

--------------------

## 单元测试 ##

package com.artisan.springkafka.produceTest;
    
    import com.artisan.springkafka.SpringkafkaApplication;
    import com.artisan.springkafka.producer.ArtisanProducerMock;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.test.context.junit4.SpringRunner;
    import org.springframework.util.concurrent.ListenableFuture;
    import org.springframework.util.concurrent.ListenableFutureCallback;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.TimeUnit;
    
    /** * @author 小工匠 * * @version 1.0 * @description: TODO * @date 2021/2/17 22:40 * @mark: show me the code , change the world */
    
    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = SpringkafkaApplication.class)
    public class ProduceMockTest { 
    
        private Logger logger = LoggerFactory.getLogger(getClass());
    
    
        @Autowired
        private ArtisanProducerMock artisanProducerMock;
    
    
        @Test
        public void testAsynSend() throws ExecutionException, InterruptedException { 
            logger.info("开始发送 测试广播模式");
    
            artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() { 
                @Override
                public void onFailure(Throwable throwable) { 
                    logger.info(" 发送异常{}]]", throwable);
    
                }
                @Override
                public void onSuccess(SendResult<Object, Object> objectObjectSendResult) { 
                    logger.info("回调结果 Result = topic:[{}] , partition:[{}], offset:[{}]",
                            objectObjectSendResult.getRecordMetadata().topic(),
                            objectObjectSendResult.getRecordMetadata().partition(),
                            objectObjectSendResult.getRecordMetadata().offset());
                }
            });
    
    
            // 阻塞等待，保证消费
            new CountDownLatch(1).await();
    
        }
    
    }

启动多次单元测试， 观察消息的接受情况

--------------------

## 测速结果 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 2]

可以看到不消费组下的 消费者（目前是一个消费组下一个消费者） 均收到了 这条消息，这就是广播模式

![在这里插入图片描述][20210218113735320.gif_pic_center]

--------------------

# 源码地址 #

[https://github.com/yangshangwei/boot2/tree/master/springkafkaBroadCast][https_github.com_yangshangwei_boot2_tree_master_springkafkaBroadCast]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcnRpc2FuLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20221024/ee9c6ecc3b034737a530a0d7b5e1145a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70]: /images/20221024/9a51e07607294ab9a88bd0952081ced2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 1]: /images/20221024/0ab89e07269e45288742d5e03bcf5802.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 2]: /images/20221024/291f982ce0194e7cb40ea0784a749b4f.png
[20210218113735320.gif_pic_center]: /images/20221024/743747ba82774b20a7518cefcca87c13.png
[https_github.com_yangshangwei_boot2_tree_master_springkafkaBroadCast]: https://github.com/yangshangwei/boot2/tree/master/springkafkaBroadCast