Kafka入门(安装及使用)

Posted dinghaoran

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka入门(安装及使用)相关的知识,希望对你有一定的参考价值。

Kafka是一种分布式的,基于发布/订阅的消息系统。

Kafka的组成包括:

  • Kafka将消息以topic为单位进行归纳。
  • 将向Kafka topic发布消息的程序成为producers.
  • 将预订topics并消费消息的程序成为consumer.
  • Kafka以集群的方式运行,可以由一个或多个服务组成,每个服务叫做一个broker.

Kafka的下载:https://kafka.apache.org/downloads      下载最新版本

解压后修改配置(config/server.properties:broker.id、log.dirs)

vim config/server.properties

broker.id = 1

log.dirs = "日志目录地址"

启动服务:

  • Kafka用到了Zookeeper,所有首先启动Zookper,下面简单的启用一个单实例的Zookkeeper服务。可以在命令的结尾加个&符号,这样就可以启动后离开控制台。
bin/zookeeper-server-start.sh config/zookeeper.properties &
  • 现在启动Kafka
bin/kafka-server-start.sh config/server.properties
  • 创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
  •  发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 
  • 启动consumer
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

 

Springboot中整合Kafka:

  pom文件中:

 

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.2.0.RELEASE</version>
</dependency>
 
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

  application.yml中:

spring:
  # KAFKA
  kafka:
    # ?kafka服务器地址,可以指定多个
    bootstrap-servers: 123.xxx.x.xxx:19092,123.xxx.x.xxx:19093,123.xxx.x.xxx:19094
    #=============== producer生产者配置 =======================
    producer:
      retries: 0
      # 每次批量发送消息的数量
      batch-size: 16384
      # 缓存容量
      buffer-memory: 33554432
      # ?指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    #=============== consumer消费者配置  =======================
    consumer:
      #指定默认消费者的group id
      group-id: test-app
      #earliest
      #当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
      #latest
      #当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
      #none
      #topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
      auto-offset-reset: latest
      enable-auto-commit: true
      auto-commit-interval: 100ms
      #指定消费key和消息体的编解码方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  定义一个生产者类:KafkaSender 负责消息推送:

@Component
 
public class KafkaSender 
 
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    private final Logger logger = LoggerFactory.getLogger(KafkaSender.class);
 
 
    public void send(String topic, String taskid, String jsonStr) 
 
 
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, taskid, jsonStr);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() 
            @Override
            //推送成功
            public void onSuccess(SendResult<String, Object> result) 
                logger.info(topic + " 生产者 发送消息成功:" + result.toString());
 
 
            
 
            @Override
            //推送失败
            public void onFailure(Throwable ex) 
                logger.info(topic + " 生产者 发送消息失败:" + ex.getMessage());
 
 
            
        );
 
 
    
 
 

  定义一个消费者类:KafkaCustomer 用来接收消息

@Component
public class KafkaConsumer  
 
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
//下面的主题是一个数组,可以同时订阅多主题,只需按数组格式即可,也就是用“,”隔开
    @KafkaListener(topics = "testTopic")
    public void receive(ConsumerRecord<?, ?> record)
 
        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());
    
 

  

 

以上是关于Kafka入门(安装及使用)的主要内容,如果未能解决你的问题,请参考以下文章

Kafka 安装及快速入门

Kafka入门系列——Kafka环境安装

Kafka入门 概述部署与API的简单使用

Kafka入门 概述部署与API的简单使用

linux环境kafka安装及配置

kafka集群安装及简单使用