Springboot整合Kafka

川长思鸟来 2022-10-06 01:59 170阅读 0赞

# Springboot整合Kafka #

### 介绍kafka ###

Kafka是由Apache软件基金会开发的一个开源流处理平台，由`Scala和Java编写`。Kafka是`一种高吞吐量的分布式发布订阅消息系统`，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过`集群来提供实时的消息`。

### 特性： ###

*  Kafka 是一种高吞吐量的分布式发布订阅消息系统，有如下特性：通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
 *  高吞吐量 ：即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。
 *  支持通过Kafka服务器和消费机集群来分区消息。
 *  支持Hadoop并行数据加载。

### 整合： ###

1.导入依赖

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

2.配置

spring:
      kafka:
        producer:
          bootstrap-servers: ip:9092
        consumer:
          group-id: springboot-group1,springboot-group2
          auto-offset-reset: earliest

发送定时任务：docker开启kafka+zookeeper

@Component
    @EnableScheduling
    public class KafkaSender { 
        //模板
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        /** * 定时任务 每隔一秒发送 */
        @Scheduled(cron = "00/1 * * * * ?")
        public void send(){ 
            String string = UUID.randomUUID().toString();
            System.out.println(string);
            //发送消息
            ListenableFuture future = kafkaTemplate.send("mytest1", string);
            future.addCallback(o -> System.out.println("成功"+string),throwable -> System.out.println("失败"+throwable));
        }
    }

![在这里插入图片描述][20210610203157700.png]

[20210610203157700.png]: /images/20221005/7856b9db3a0d4517b0880664c24cff84.png