选择与其他mq相比1.2、应用场景1.2.1、流量消峰1.2.2、解耦1.2.3、异步通讯

1.3、消息队列的两种模式1.3.1、点对点模式1.3.2、发布/订阅模式

1.4、Kafka

二、安装部署2.1、安装包方式2.2、docker安装方式2.3、docker安装kafka-map图形化管理工具

三、Kafka

命令行操作3.1、主题命令行操作3.2、生产者命令行操作3.3、消费者命令行操作

四、Springboot整合Kafka4.1、依赖配置4.2、代码示例4.2.1、简单的生产、消费4.2.2、指定分区生产、消费4.2.3、消息确认

一、简介

2、Kafka与其他消息队列MQ如ActiveMQ、RabbitMQ等相比有以下几个区别

磁盘存储Kafka将所有消息都保存在磁盘上并使用内存映射文件进行读写。

这种存储方式可以支持大量的消息数据而且数据还可以保留很长时间比如几个月甚至几年。

而其他MQ的存储方式多是基于内存不适合存储大量的数据。

分布式设计Kafka是作为分布式系统设计的可以在多个节点之间实现消息的高效传输和处理。

其他MQ也支持分布式部署但Kafka在这方面更加优秀。

发布/订阅模式Kafka采用发布/订阅模式允许多个消费者同时订阅同一个主题而且Kafka消费者可以自定义从哪个位置开始消费消息。

其他MQ中消费者一般需要通过消费者组来进行负载均衡而且其他MQ消费者只能从当前位置开始消费。

大数据处理Kafka最初是为大数据处理而设计的它可以非常高效地处理海量数据适合用于数据仓库、日志处理、统计分析等场景。

其他MQ则更多用于异步通信、任务调度、实时通知等领域。

Connect、Kafka

Streams等工具和框架可以方便地与大数据处理平台如Hadoop、Spark、Flink等进行集成。

其他MQ的生态系统相对较小。

1.2、应用场景

如双十一秒杀期间参与用户10亿人/s但是我们的系统只能支持处理能力1千万人/s为了避免服务挂掉或者请求超时等等问题我们可以将10亿的请求都写入到消息队列中我们系统再去取消息队列上的消息消费达到流量消峰的效果。

1.2.2、解耦

我们的数据源来源可能非常多不可能都全部去集成。

例如场景我们要去买东西不用去知道他是在哪里进货的我们只需要去超市买就可以消息队列也是如此那么多的数据让他们全部写入到kafka消息队列中即可我们再去消息队列中获取我们的数据。

通常我们写代码如注册完要发短信如果同步处理等到发短信成功后再返回结果给用户这样请求时间太久了。

消息队列的方式可以再注册的时候发送给队列我们这时候就可以返回给用户注册成功了然后消费者再去消费发送短信的队列达到异步的效果。

很多人可能会问多线程的方式不是也能实现到底选择多线程还是消息队列呢

当需要进行任务处理并且任务处理之间没有明显的依赖关系时使用消息队列更适合。

将任务发送到消息队列中由消费者进行消费这样可以实现解耦、提高可靠性和系统的扩展性。

例如一个网站需要生成大量的报告将报告生成请求通过消息队列进行异步处理是一个不错的选择。

当需要对任务进行精细控制并且任务处理之间存在明显的依赖关系时使用多线程更适合。

多线程可以实现更加细粒度的任务处理可以控制任务的执行顺序、进行资源的共享等。

例如一个电商网站需要实时监控库存的变化需要在某个商品的库存下降到一定数量时进行补货在这种情况下使用多线程处理更加合适。

总之消息队列和多线程都有自己的优势和劣势要根据具体的场景选择合适的方式才能更好地提高系统效率和可靠性。

1.3.1、点对点模式

可以有多个topic主题浏览、点赞、收藏、评论等消费者消费数据之后不删除数据(可以控制什么时候删除)每个消费者相互独立都可以消费到数据

1.4、Kafka

1、海量数据分而置之为方便扩展并提高吞吐量一个topic分为多个partition分区。

如100T的数据我可以分成3个区每个区三十几G,可以提高吞吐量。

一个分区的数据只能由一个消费者处理如有两个消费者消费不知道由第一个还是第二个来消费。

3、为提高可用性为每个partition增加若干副本防止一个分区挂了类似NameNode

副本分为leader和fallower之分follower不做被消费只是为了防止leader副本挂了后follower有条件成为leader提高可用性。

zookeeper中记录谁是leader和整个集群中哪些服务器正在工作Kafka2.8.0以后也可以配置不采用ZK安装包已内置ZK

Kafka

组成。

消费者组内每个消费者负责消费不同分区的数据一个分区只能由一个组内消费者消费消费者组之间互不影响。

所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者。

Broker一台

Kafka

topic。

Topic可以理解为一个队列生产者和消费者面向的都是一个

topic

和若干个Follower。

Leader每个分区多个副本的“主”生产者发送数据的对象以及消费者消费数据的对象都是

Leader。

Follower每个分区多个副本中的“从”实时从

Leader

https://kafka.apache.org/downloads

2、将安装包上传服务器并解压

kafka_2.12-3.4.0.tgz进入下的server.properties配置文件

vim

server.properties修改log.dirs的路径这个是kafka存储数据的地方默认放在了临时文件夹里容易被删除我们需改成我们服务器可以存放的目录如我放在home底下

#编辑配置文件

PATH$PATH:$KAFKA_HOME/bin#例如笔者的文件路径

export

KAFKA_HOME/www/wwwroot/kafka_2.12-3.4.0/

export

/etc/profile5、配置config下的zookeeper配置同样也只是修改存储路径

vim

zookeeper.properties将dataDir修改成你自己存储路径如我的dataDir/home/zookeeper

还有下面的advertised放开注释更换ip,zookeeper连接也更换ip6、启动需先启动zookeeper

注意要先有java环境

启动zookeeper直接指定到你们自己的bin和config目录

nohup

/www/wwwroot/kafka_2.12-3.4.0/bin/zookeeper-server-start.sh

-daemon

/www/wwwroot/kafka_2.12-3.4.0/config/zookeeper.properties#

启动kafka

/www/wwwroot/kafka_2.12-3.4.0/bin/kafka-server-start.sh

-daemon

/www/wwwroot/kafka_2.12-3.4.0/config/server.properties7、验证是否启动成功查看端口是有被用或者看日志文件

2.2、docker安装方式

KAFKA_ZOOKEEPER_CONNECT这里换成你的zookeeper地址和端口

KAFKA_ADVERTISED_LISTENERSPLAINTEXT://这里换成你的kafka地址和端口

KAFKA_LISTENERSPLAINTEXT://0.0.0.0:9092

wurstmeister/kafka-----------------------------------------------

发送消息与消费测试

KAFKA_ADVERTISED_LISTENERSkafka发布到zookeeper供客户端使用的服务地址。

KAFKA_LISTENERS允许使用PLAINTEXT侦听器

如图发送和接收都成功

/home/kafka-map/data:/usr/local/kafka-map/data

--restart

dushixiang/kafka-map:latest开发端口访问界面http://ip:9080

图形化工具教程

bin/kafka-topics.shtopic的命名都是有规律的按上面的提示固定前缀

bin/kafka-topics.sh

bin/kafka-console-producer.sh2、发送消息(发送到topic为first)

--bootstrap-server

bin/kafka-console-consumer.sh2、消费消息监听topic为first的

--bootstrap-server

dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependencyyml配置

spring:#

指定消息key和消息体的编解码方式key-serializer:

org.apache.kafka.common.serialization.StringSerializervalue-serializer:

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

写入失败时重试次数。

当leader节点失效一个repli节点会替代成为leader节点此时可能出现写入失败#

当retris为0时produce不会重复。

retirs重发此时repli节点完全成为leader节点不会产生消息丢失。

retries:

每次批量发送消息的数量,produce积累到一定数据一次发送batch-size:

16384#

produce积累数据一次发送缓存大小达到buffer.memory就发送数据buffer-memory:

指定默认消费者

procedure要求leader在考虑完成请求之前收到的确认数用于控制发送记录在服务端的持久化其值可以为如下#

acks

如果设置为零则生产者将不会等待来自服务器的任何确认该记录将立即添加到套接字缓冲区并视为已发送。

在这种情况下无法保证服务器已收到记录并且重试配置将不会生效因为客户端通常不会知道任何故障为每条记录返回的偏移量始终设置为-1。

#

acks

这意味着leader会将记录写入其本地日志但无需等待所有副本服务器的完全确认即可做出回应在这种情况下如果leader在确认记录后立即失败但在将数据复制到所有的副本服务器之前则记录将会丢失。

#

acks

这意味着leader将等待完整的同步副本集以确认记录这保证了只要至少一个同步副本服务器仍然存活记录就不会丢失这是最强有力的保证这相当于acks

-1,

指定消息key和消息体的编解码方式key-deserializer:

org.apache.kafka.common.serialization.StringDeserializervalue-deserializer:

org.apache.kafka.common.serialization.StringDeserializer4.2、代码示例

生产者

record.value());System.out.println(topic:

record.topic()

record.value());}4.2.2、指定分区生产、消费

Kafka

中生产者可以指定消息的分区和键Key属性。

指定分区和键属性可以带来以下好处

通过指定分区生产者可以控制消息被发送到哪个分区从而控制消息的分发。

例如如果您想要按照时间戳对消息进行排序可以将消息发送到同一个分区中。

Kafka

通过指定键属性生产者可以确保具有相同键的消息被发送到同一个分区中。

这可以确保消息按照顺序被处理从而提高消息的可靠性。

生产者

有keykafkaTemplate.send(two,0,111,

msg0分区);kafkaTemplate.send(two,1,222,

msg1分区);//

没有keykafkaTemplate.send(two,2,null,

msg2分区);//

发送另一个topic消费者测试用kafkaTemplate.send(first,0,111,

msgfirst);

同时监听two和first监听two的0号和1号分区、first的

0号和1号

分区指向1号分区的offset初始值为8**/KafkaListener(topicPartitions

two,

{System.out.println(topic:record.topic()|partition:record.partition()|offset:record.offset()|value:record.value());}//

如果没指定分区就算你生产者发送了多个分区我也全都能接收//KafkaListener(topics

two)//public

record.value());//System.out.println(topic:

record.topic()

#配置监听者#ack-mode参数值如下#MANUAL手动确认模式消费者需要手动调用

Acknowledgment

方法来确认消息。

#MANUAL_IMMEDIATE立即手动确认模式与

MANUAL

模式相同但是消费者在处理完消息后立即提交偏移量而不是等待下一次拉取。

#BATCH批量确认模式消费者将在处理完一批消息后自动提交偏移量。

#RECORD记录确认模式消费者将在处理完每条消息后自动提交偏移量。

ack-mode:

manual

指定消息key和消息体的编解码方式key-serializer:

org.apache.kafka.common.serialization.StringSerializervalue-serializer:

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

写入失败时重试次数。

当leader节点失效一个repli节点会替代成为leader节点此时可能出现写入失败#

当retris为0时produce不会重复。

retirs重发此时repli节点完全成为leader节点不会产生消息丢失。

retries:

每次批量发送消息的数量,produce积累到一定数据一次发送batch-size:

16384#

produce积累数据一次发送缓存大小达到buffer.memory就发送数据buffer-memory:

指定默认消费者

procedure要求leader在考虑完成请求之前收到的确认数用于控制发送记录在服务端的持久化其值可以为如下#

acks

如果设置为零则生产者将不会等待来自服务器的任何确认该记录将立即添加到套接字缓冲区并视为已发送。

在这种情况下无法保证服务器已收到记录并且重试配置将不会生效因为客户端通常不会知道任何故障为每条记录返回的偏移量始终设置为-1。

#

acks

这意味着leader会将记录写入其本地日志但无需等待所有副本服务器的完全确认即可做出回应在这种情况下如果leader在确认记录后立即失败但在将数据复制到所有的副本服务器之前则记录将会丢失。

#

acks

这意味着leader将等待完整的同步副本集以确认记录这保证了只要至少一个同步副本服务器仍然存活记录就不会丢失这是最强有力的保证这相当于acks

-1,

指定消息key和消息体的编解码方式key-deserializer:

org.apache.kafka.common.serialization.StringDeserializervalue-deserializer:

org.apache.kafka.common.serialization.StringDeserializer生产者

/***

kafkaTemplate提供了一个回调方法addCallback我们可以在回调方法中监控消息是否发送成功

失败时做补偿处理*

ListenableFutureCallbackSendResultString,

Object()

{System.out.println(发送消息失败ex.getMessage());}Overridepublic

void

result.getRecordMetadata().topic()

result.getRecordMetadata().partition()

result.getRecordMetadata().offset());}});

}消费者

record.value());System.out.println(topic:

record.topic()

假设出现异常没有走下面的提交//手动提交offsetack.acknowledge();}

catch

}消费者异常了后消息没有被确认我们重启项目时会再次出现那条未确认的消息