前言二、环境说明三、概念四、CentOS7安装kafka1.下载kafka安装包2.下载好后进行解压六、kafka项目集成1️⃣pom引入2️⃣配置kafka3️⃣一个kafka消息发送端4️⃣定义一个kafka消息消费端5️⃣定义一个Controller进行测试6️⃣测试结果如下一、前言

上一期我是带着大家入门了SpringBoot整合WebSocket今天我再来一期kafka的零基础教学吧。

不知道大家对kafka有多少了解反正我就是从搭建开始然后再加一个简单演示这就算是带着大家了个门哈剩下的我再后边慢慢出教程给大家说。

二、环境说明

kafka是linkedin开源的分布式发布-订阅消息系统目前归属于Apache的顶级项目。

主要特点是基于pull模式来处理消息消费追求高吞吐量在一台普通的服务器上既可以达到10W/s的吞吐速率完全的分布式系统。

一开始的目的是日志的收集和传输。

0.8版本开始支持复制不支持事务对消息的丢失重复错误没有严格要求

适用于产生大量数据的互联网服务的数据收集业务。

在廉价的服务器上都能有很高的性能这个主要是基于操作系统底层的pagecache不用内存胜似使用内存。

综上所述kafka是一款开源的消息引擎系统(消息队列/消息中间件)

分布式流处理平台

下载地址https://kafka.apache.org/downloads.html

2.下载好后进行解压

通过ftp将kafka安装包kafka_2.11-0.9.0.1.tgz上传到服务器

执行命令unzip

/opt/monitor/kafka/kafka_2.12-2.2.1

进入kafka目录

/opt/monitor/kafka/kafka_2.12-2.2.11

配置并启动zookeeper

config/zookeeper.properties按一下键盘上的

键进入编辑模式将光标移动到日志文件存放路径配置信息所在行并修改dataDir/opt/monitor/kafka/kafka_2.12-2.2.1/zklogs

dataDir/opt/monitor/kafka/kafka_2.12-2.2.1/zklogs修改好后按下键盘上的Esc

输入wq

执行sh./zookeeper-server-start.sh

命令后台启动zookeeper

注意这里提示报错权限不足使用命令修改权限个人建议把bin的权限全部修改成777

chmod

/opt/monitor/kafka/kafka_2.12-2.2.1/config/zookeeper.properties执行命令ps

-ef

查看zookeeper是否启动成功出现类型如下信息表示成功启动

配置并启动kafka

键进入编辑模式修改advertised.listenersPLAINTEXT://外网IP:9092;

修改log.dirs/opt/monitor/kafka/kafka_2.12-2.2.1/logs该参数为kafka日志文件存放路径

修改每个topic的默认分区参数num.partitions默认是1具体合适的取值需要根据服务器配置进程确定

输入wq

/opt/monitor/kafka/kafka_2.12-2.2.1/bin

#进入kafka启动目录

/opt/monitor/kafka/kafka_2.12-2.2.1/config/server.properties

#启动kafka服务指定配置文件执行命令查看kafka是否启动成功

-ef

dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId

spring:kafka:bootstrap-servers:

0#当有多个消息需要被发送到同一个分区时生产者会把它们放在同一个批次里。

该参数指定了一个批次可以使用的内存大小按照字节数计算。

batch-size:

16384#

org.apache.kafka.common.serialization.StringSerializer#

org.apache.kafka.common.serialization.StringSerializer#

acks0

只要集群的首领节点收到消息生产者就会收到一个来自服务器成功响应。

#

acksall

只有当所有参与复制的节点全部收到消息时生产者才会收到一个来自服务器的成功响应。

acks:

1consumer:#

需要符合特定的格式如1S,1M,2H,5Dauto-commit-interval:

1S#

该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理#

latest默认值在偏移量无效的情况下消费者将从最新的记录开始读取数据在消费者启动之后生成的记录#

earliest

在偏移量无效的情况下消费者将从起始位置读取分区的记录auto-offset-reset:

earliest#

是否自动提交偏移量默认值是true,为了避免出现重复数据和数据丢失可以把它设置为false,然后手动提交偏移量enable-auto-commit:

false#

org.apache.kafka.common.serialization.StringDeserializer#

org.apache.kafka.common.serialization.StringDeserializerlistener:#

5#listner负责ack每调用一次就立即commitack-mode:

manual_immediatemissing-topics-fatal:

dev

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

org.springframework.util.concurrent.ListenableFutureCallback;/***

author

JSONUtil.toJsonStr(obj);log.info(准备发送消息为{},

obj2String);//发送消息ListenableFutureSendResultString,

Object

ListenableFutureCallbackSendResultString,

Object()

throwable.getMessage());}Overridepublic

void

stringObjectSendResult.toString());}});}

4️⃣定义一个kafka消息消费端

org.apache.kafka.clients.consumer.ConsumerRecord;

import

org.springframework.kafka.annotation.KafkaListener;

import

org.springframework.kafka.support.Acknowledgment;

import

org.springframework.kafka.support.KafkaHeaders;

import

org.springframework.messaging.handler.annotation.Header;

import

org.springframework.stereotype.Component;import

author

KafkaProducer.TOPIC_GROUP1)public

void

Header(KafkaHeaders.RECEIVED_TOPIC)

String

Optional.ofNullable(record.value());if

{Object

message.get();log.info(topic.group1

Topic:

msg);ack.acknowledge();}}KafkaListener(topics

groupId

KafkaProducer.TOPIC_GROUP2)public

void

Header(KafkaHeaders.RECEIVED_TOPIC)

String

Optional.ofNullable(record.value());if

{Object

message.get();log.info(topic.group2

Topic:

com.suihao.kafka.KafkaProducer;

import

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.web.bind.annotation.GetMapping;

import

org.springframework.web.bind.annotation.RestController;/***

author

kafkaProducer;GetMapping(/send)public

void

sendMsg(){kafkaProducer.send(------------测试消息-----------);}

6️⃣测试结果如下

https://gitee.com/suihao666/SpringBoot-Kafka

最后送所有正在努力的大家一句话你不一定逆风翻盘但一定要向阳而生。