kafka初认识以及相关使用

灰太狼 2022-01-22 13:19 230阅读 0赞

kafka在现在公司都会应用很广泛，性能也是爆炸性的。无论在大数据领域，或者日常开发也常常看到它的身影。

**目录**

kafka与其他MQ的区别

kafka的组成

kafka在持久化与其他MQ的不同

添加日志的过程

kafka如何保证消息的发送成功

acks

kafka监控

kafka实践

pom.xml

配置

producer

consumer

有意思的应用

个人想法

实现想法

优化方案

想法

实现

方案前后对比

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

## kafka与其他MQ的区别 ##

### kafka的组成 ###

kafka被称为有broker暴力简单路由

由producer,consumer,broker组成

举个栗子：

producer:像母鸡一直生产鸡蛋

consumer:像农民一直收获鸡蛋

而borker呢：像个篮子，把消息储存起来消费。

topic：消费主题，消费者通过监听相应的主题来收到消息

可以说kafka的构造相对比较简单，暴力实现消息传递功能，性能也相应提高了。

那其他MQ：除了这些基本的，还会添加相应的Queue等等，进行相应的处理，所以叫做复杂路由

还有一个区别：kafka需要zookeeper的协助，mq可以单独运行。

### kafka在持久化与其他MQ的不同 ###

一般MQ可以持久化到内存（刷盘），数据库等等，kafka是以日志的形式添加到日志文件里面。

### 添加日志的过程 ###

读操作：直接读取日志

写操作：按顺序添加到日志，所以不会乱序

我们可以看到读和写都不会说阻塞，这也是提高性能的一部分。

kafka有两个参数 M S,意义：会在S秒里面强制性的把最后M条消息刷新到日志里面，所以宕机的话最多影响M条消息

### kafka如何保证消息的发送成功 ###

最终目的是保证只发送一次，这个是我们想要的结果。纳秒kafka如何实现？

生产者发送成功之后收到commit消息，来确保broker已经收到消息。如果没有收到确认消息呢？会重新发送一次，那这里有可能会出现重复的消息，kafka会去重。

### acks ###

这里涉及到一个参数acks,配置在发送方*spring*.*kafka*.producer.*acks*

可以参考[这篇文章][Link 1]

总结：acks默认为1

acks为0的时候，只要消息发送出去，就算成功，不管有没有刷新到磁盘

acks为1的时候，kafka leader接收到消息，然后刷新到磁盘之后为收到状态。

acks为all的时候，必须是所有kafka集群收到并刷新到磁盘，为收到状态

为1的时候，如果主机写入完瞬间宕机，其他机器也没有写入，也是白搭。

为all的时候，必须是集群，如果只有一台主机也白搭

## kafka监控 ##

有蛮多kafka监控的，这里使用雅虎kafka manager

下载链接: https://pan.baidu.com/s/10NpnqoGiV7ygToNRut76\_A 提取码: x9i3

它是需要自己编译的，我在网上找到相应的编译完的。需要配置的文件kafka-manager-1.3.3.17\\conf，下面的application.conf

kafka-manager.zkhosts="localhost:2181"

修改这里zk地址

默认端口9000，启动在bin

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70][]

## kafka实践 ##

启动zk，启动kafka，启动kafka manager

个人是通过springboot去整合kafka

### pom.xml ###

<dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>

### 配置 ###

spring:
      kafka:
        bootstrap-servers: localhost:9092
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          group-id: executor
          enable-auto-commit: true
          auto-commit-interval: 1000
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

需要修改zk的配置

### producer ###

@Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
    
        kafkaTemplate.send("test_topic", msg);

发送消息，第一个参数为topic

### consumer ###

@KafkaListener(topics = "test_topic")
        public void listen(ConsumerRecord<?, ?> record) throws Exception {}

通过@KafkaListener来监听相应topic的消息，其中可以配置多个topic

record.topic(), record.offset(), record.value()

对应topic，偏移量，消息内容

### 有意思的应用 ###

本人现在负责公司一个用户量上w的小程序（某完子）开发，可能会遇到一个开发难题：如果需要给上w的用户发送消息，那么这个对cpu等等资源是一个挑战

### 个人想法 ###

这个跟秒杀不同的是：秒杀保证在高并发下不崩，而且可以快速处理。而对于处理大数量任务的，我们的目的是保证在处理期间不会占用电脑大量cpu完成任务，即使时间长点无所谓。

### 实现想法 ###

将任务批量通过kafka发送到消费者，然后通过多线程去消费，注意不要创建太多线程去执行，达到占用cpu不是很大 ，让其他cpu处理其他业务。

生产者

@RestController
    @RequestMapping("kafka")
    public class TestKafkaProducerController {
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        static AtomicLong i = new AtomicLong(0);
    
        @RequestMapping("send")
        public String send(String msg) throws InterruptedException {
            long time = System.nanoTime();
            for (int i = 0; i < 10000; i++) {
                kafkaTemplate.send("test_topic", msg);
                if (i == 800) {
                    Thread.sleep(200);
                }
            }
            while (true) {
                if (i.doubleValue() >= 10000) {
                    break;
                }
            }
            long time1 = System.nanoTime();
            System.out.println("耗时" + (time1 - time));
            return "success";
        }
    
    }

在生产消息的时候，sleep一会，这里可以再进行优化，目的是批量发送，不会一下子消息爆炸

消费者

@KafkaListener(topics = "test_topic", id = "executor")
        public void listen(ConsumerRecord<?, ?> record) throws Exception {
    
            executor.execute(new Thread(new Runnable() {
                @Override
                public void run() {
                    //模拟处理业务
                    try {
                        Thread.sleep(500);
                        System.out.printf("处理消息：topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
                        TestKafkaProducerController.i.incrementAndGet();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }));
    
        }

模拟处理业务0.5秒，计算消耗时间

耗时1670896719700纳秒  
27.8482787    分(min)

最后耗时27分钟，有点时间长，我们再进行优化

## 优化方案 ##

### 想法 ###

多个消费者同时消费消息，那么会出现消费同样的消息，这里需要给消息唯一标识，进行去重，防止重复消费。然后调高线程池的核心线程参数，不然你线程多了，但是线程池核心线程就那么多，多少个消费者都没有用。

但是线程池的核心数大小不宜过大，有相应的算法，为啥？多线程处理方式和cpu的使用是一样的，就是你用一哈我用一哈，你用完我去用的时候，就会出现上下文切换，这个会耗时间。

### 实现 ###

@Component
    public class TestConsumer {
    
        static int size = 1;
    
        static {
            int corePoolSize = Runtime.getRuntime().availableProcessors();
            if (corePoolSize > 1) {
                size = corePoolSize - 1;
            }
        }
    
        public static void main(String[] args) {
            System.out.println(Runtime.getRuntime().availableProcessors());
        }
    
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 20, 10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(10000));
    
        private ConcurrentHashMap concurrentHashMap=new ConcurrentHashMap(10010);
    
        @KafkaListener(topics = "test_topic")
        public void listen1(ConsumerRecord<?, ?> record) throws Exception {
    
            if(concurrentHashMap.get(record.offset())!=null){
                return;
            }else {
                concurrentHashMap.put(record.offset(),1);
                executor.execute(new Thread(new Runnable() {
                    @Override
                    public void run() {
                        //模拟处理业务
                        try {
                            Thread.sleep(500);
                            System.out.printf("处理消息：topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
                            TestKafkaProducerController.i.incrementAndGet();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }));
            }
    
        }
    
        @KafkaListener(topics = "test_topic")
        public void listen2(ConsumerRecord<?, ?> record) throws Exception {
    
            if(concurrentHashMap.get(record.offset())!=null){
                return;
            }else {
                concurrentHashMap.put(record.offset(),1);
                executor.execute(new Thread(new Runnable() {
                    @Override
                    public void run() {
                        //模拟处理业务
                        try {
                            Thread.sleep(500);
                            System.out.printf("处理消息：topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
                            TestKafkaProducerController.i.incrementAndGet();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }));
            }
    
        }
    
        @KafkaListener(topics = "test_topic")
        public void listen3(ConsumerRecord<?, ?> record) throws Exception {
    
            if(concurrentHashMap.get(record.offset())!=null){
                return;
            }else {
                concurrentHashMap.put(record.offset(),1);
                executor.execute(new Thread(new Runnable() {
                    @Override
                    public void run() {
                        //模拟处理业务
                        try {
                            Thread.sleep(500);
                            System.out.printf("处理消息：topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
                            TestKafkaProducerController.i.incrementAndGet();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }));
            }
    
        }
    
    }

由于这里是在本地进行消费，所以使用Map来去重。如果是在分布式环境可以使用redis等等缓存来保证数据的一致性

耗时501649666499纳秒  
8.3608278    分(min)

## 方案前后对比 ##

后者提高的不是一丢丢哦，时间减少了很多，通过查看cpu占用率，后者会高一些，前者在30%左右，后者在50%左右，用于在windows系统，检测只是大概的。

[Link 1]: https://mp.weixin.qq.com/s?__biz=MzA3ODQ0Mzg2OA==&mid=2649049755&idx=2&sn=acee50f2bbf9eecc5d2fb48c497674bb&chksm=87534ea8b024c7be0d9f7fa3775979a047610bb4405d8ea4cffd6a1b759c994ad146902d0976&scene=0&xtrack=1#rd
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]: /images/20220122/65a3b2dda53544628a005f01298bcfc5.png