kafka集群安装搭建(springBoo集成kafka)

kafka介绍

根据官网的介绍，ApacheKafka®是一个分布式流媒体平台，它主要有3种功能：

• 发布和订阅消息流，这个功能相似于消息队列，这也是kafka归类为消息队列框架的缘由
• 以容错的方式记录消息流，kafka以文件的方式来存储消息流
• 能够再消息发布的时候进行处理

使用场景

• 在系统或应用程序之间构建可靠的用于传输实时数据的管道，消息队列功能
• 建实时的流数据处理程序来变换或处理数据流，数据处理功能

Kafka目前主要做为一个分布式的发布订阅式的消息系统使用 下图为消息传输流程

• Producer即生产者，向Kafka集群发送消息，在发送消息以前，会对消息进行分类，即Topic，上图展现了两个producer发送了分类为topic1的消息，另一个发送了topic2的消息。
• Topic即主题，经过对消息指定主题能够将消息分类，消费者能够只关注本身须要的Topic中的消息
• Consumer即消费者，消费者经过与kafka集群创建长链接的方式，不断地从集群中拉取消息，而后能够对这些消息进行处理。

下载

在kafka官网 http://kafka.apache.org/downloads下载到最新的kafka安装包，选择下载二进制版本的tgz文件

安装

• 首先确保你的机器上安装了jdk，kafka须要java运行环境，之前的kafka还须要zookeeper，新版的kafka已经内置了一个zookeeper环境，因此咱们能够直接使用。
• 若是只须要进行最简单的尝试的话咱们只须要解压到任意目录便可，这里咱们将kafka压缩包解压到/home目录

kafka解压目录下下有一个config的文件夹，里面放置的是咱们的配置文件

consumer.properites 消费者配置

producer.properties 生产者配置

server.properties kafka服务器的配置，此配置文件用来配置kafka服务器 目前仅介绍几个最基础的配置

• broker.id 申明当前kafka服务器在集群中的惟一ID，需配置为integer,而且集群中的每个kafka服务器的id都应是惟一的，咱们这里采用默认配置便可
• listeners 申明此kafka服务器须要监听的端口号，若是是在本机上跑虚拟机运行能够不用配置本项，默认会使用localhost的地址，若是是在远程服务器上运行则必须配置，例如：　listeners=PLAINTEXT:// 192.168.180.128:9092。并确保服务器的9092端口可以访问
• zookeeper.connect 申明kafka所链接的zookeeper的地址 ，需配置为zookeeper的地址，因为本次使用的是kafka高版本中自带zookeeper，使用默认配置便可 zookeeper.connect=localhost:2181

运行

启动zookeeper

#前台启动
[root@CentOS124 home]# cd kafka2.11/
[root@CentOS124 kafka2.11]# bin/zookeeper-server-start.sh config/zookeeper.properties

#后台启动
[root@CentOS124 kafka2.11]# bin/zookeeper-server-start.sh config/zookeeper.properties 1>/dev/null 2>&1 &
[1] 18466

#查看是否启动成功
[root@CentOS124 ~]#  ps -ef|grep kafka

启动kafka

[root@CentOS124 kafka2.11]# bin/kafka-server-start.sh config/server.properties

#后台启动
[root@CentOS124 kafka2.11]# bin/kafka-server-start.sh config/server.properties 1>/dev/null 2>&1 &

#建立 topic
[root@CentOS124 kafka2.11]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test


#查看Kafka 中的 topic 列表
bin/kafka-topics.sh --list --zookeeper localhost:2181

配置kafka集群

• 条件有限，在同一个机器上启动三个broker来模拟kafka集群，三个broker使用另外安装的同一个zookeeper服务（实际集群中，每一个broker一般在不一样的机器上，也会使用不一样host的zookeeper）

#复制server.properties配置文件为三份，分别起名为server.properties，server-2.properties，server-3.properties

三份配置中都要修改如下
#三个配置中分别修改成0,2,3
broker.id=0

#三个配置中分别修改成9092,9093,9094
port=9092

#kafka-logs，kafka-logs-2，kafka-logs-3
log.dirs=/tmp/kafka-logs

#都设置为3，即每一个topic默认三个partition
num.partitions=3

#zookeeper集群地址，外部能够配置，这里环境有限  使用默认既可
zookeeper.connect=localhost:2181

#分别进入kafka目录下 执行以下命令启动服务控制台输出日子完成了
bin/kafka-server-start.sh config/server.properties
bin/kafka-server-start.sh config/server-2.properties
bin/kafka-server-start.sh config/server-3.properties

springBoot中如何使用kafka

首先建立一个springBoot项目 引入spring-kafka

application.properties 配置

server.port=8080

#kafka地址 brokers集群地址用,隔开
spring.kafka.bootstrap-servers=127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094

#生产者的配置，大部分咱们可使用默认的，这里列出几个比较重要的属性
#每批次发送消息的数量
spring.kafka.producer.batch-size=16
#发送失败重试次数
spring.kafka.producer.retries=0
#即32MB的批处理缓冲区
spring.kafka.producer.buffer-memory=33554432
#key序列化方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#消费者的配置
##Kafka中没有初始偏移或若是当前偏移在服务器上再也不存在时,默认区最新 ，有三个选项 【latest, earliest, none】
spring.kafka.consumer.auto-offset-reset=latest
#是否开启自动提交
spring.kafka.consumer.enable-auto-commit=true
#自动提交的时间间隔
spring.kafka.consumer.auto-commit-interval=100
#key的解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#value的解码方式
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#在kafka/config文件的consumer.properties中有配置
spring.kafka.consumer.group-id=test-consumer-group

建立Producer生产者

package com.example.modules;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
 * 〈生产者〉
 * @author qinxuewu
 * @create 18/8/4下午11:56
 * @since 1.0.0
 */
@Component
public class Producer {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    //发送消息方法
    public void send() {
        JSONObject obj=new JSONObject();
        obj.put("id",System.currentTimeMillis());
        obj.put("name","生产者发送消息");
        obj.put("date",new Date());
        //这个 topic 在 Java 程序中是不须要提早在 Kafka 中设置的，由于它会在发送的时候自动建立你设置的 topic
        kafkaTemplate.send("qxw",obj.toString());
    }
}

建立消费者

@Component
public class Consumer {
    private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
    /**
     *  同时监听两个 topic 的消息了，可同时监听多个topic
     * @param record
     * @throws Exception
     */
    @KafkaListener(topics = {"test","qxw"})
    public void listen (ConsumerRecord<?, ?> record) throws Exception {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            logger.info("消费者开始消费message：" + message);
        }
    }
}

运行后就能够看到控制台输出了

@RunWith(SpringRunner.class)                  
@SpringBootTest                               
public class KafkaDemoApplicationTests {      
    @Autowired                                
    private Producer producer;                
    @Test                                     
    public void contextLoads() {              
        for (int i = 0; i <3 ; i++) {         
            producer.send();                  
            try {                             
                Thread.sleep(1000);           
            } catch (InterruptedException e) {
                e.printStackTrace();          
            }                                 
        }                                     
    }                                         
}

kafka 配置文件参数详解

https://www.cnblogs.com/alan319/p/8651434.html kafka的配置分为 broker、producter、consumer三个不一样的配置