怎么做二维码进入公司网站,数据库网站 建设,做网站服务费税率,我想在泉州做网站本篇文章的目的是帮助Kafka初学者快速搭建一个Kafka集群#xff0c;以及怎么在SpringBoot项目中使用Kafka。
kafka集群环境包地址#xff1a;百度网盘 请输入提取码 提取码#xff1a;x9yn 一、Kafka集群搭建
1、准备环境
#xff08;1#xff09;准备三台…本篇文章的目的是帮助Kafka初学者快速搭建一个Kafka集群以及怎么在SpringBoot项目中使用Kafka。
kafka集群环境包地址百度网盘 请输入提取码 提取码x9yn 一、Kafka集群搭建
1、准备环境
1准备三台LINUX服务器
xxx.xxx.xxx.1
xxx.xxx.xxx.2
xxx.xxx.xxx.3
2jdk版本大于1.8即可我是1.8.0_181
3在三台服务器上创建用户admin将环境放到admin用户下嫌麻烦的同学也可以直接使用root用户安装真实生产上不建议这么做 tipsLINUX怎么给普通用户赋文件夹操作权限 切换到root用户使用chown -R admin:admin /home/admin命令执行su - admin命令就可以切换用户并定位到/home/admin下 4一定要关闭三台服务器的防火墙不然安装肯定会出问题切记这个真的很重要
2、搭建Zookeeper集群
1解压zookeeper-3.4.12.tar.gz进入zookeeper文件夹 2进入conf文件夹
1复制zoo.cfg文件
cp zoo.cfg zoo_sample.cfg2修改zoo.cfg文件
vim zoo.cfg 这里的3个IP的作用如下 2181对cline端提供服务 3888选举leader使用 2888集群内机器通讯使用Leader监听此端口 3进入data文件夹若没有自己创建一个 在data文件夹下创建myid文件三台机器分别填入server对应的ID这里我是1、2、3 4启动zookeeper集群
1. 启动ZK服务: sh bin/zkServer.sh start2. 查看ZK服务状态: sh bin/zkServer.sh status3. 停止ZK服务: sh bin/zkServer.sh stop4. 重启ZK服务: sh bin/zkServer.sh restart
5三台机器都需要重复上述操作注意myid中的ID要对应 3、搭建Kafka集群
1解压kafka_2.12-2.5.0.tgz进入kafka文件夹 2进入config文件夹修改 server.properties内容
# Kafka使用唯一的一个整数来标识每个broker该参数默认是-1。如果不指定kafka会自动生成一个唯一值
broker.id1
# broker监听器的CSV列表格式是[协议]://[主机名]:[端口]。
listenersPLAINTEXT://xxx.xxx.xxx.1:9092
# 非常重要的参数该参数指定了kafka持久化消息的目录。该参数可以设置多个目录以逗号分隔比如/home/kafka1,/home/kafka2,多目录的做法是推荐的
log.dirs/tmp/kafka-logs
# 同样是很重要的参数这个参数完全没有默认值是必须要自己设置的
zookeeper.connectxxx.xxx.xxx.1:2181,xxx.xxx.xxx.2:2181,xxx.xxx.xxx.3:2181
# 是否开启unclean leader选举。由于开始可能不能保证数据一致性所以设置为false
unclean.leader.election.enablefalse
# topic 在当前 broker 上的分区个数
num.partitions1
# 用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir1
# segment文件保留的最长时间超时将被删除
log.retention.hours16
# 删除 topic 功能使能 ( 允许删除数据 ) ( 手动指定 )
delete.topic.enabletrue
# 处理网络请求的线程数量
num.network.threads3
# 用来处理磁盘 IO 的线程数量
num.io.threads8
# 发送套接字的缓冲区大小
socket.send.buffer.bytes102400
# 接收套接字的缓冲区大小
socket.receive.buffer.bytes102400
# 请求套接字的缓冲区大小
socket.request.max.bytes10485760
3配置环境变量
vim ~/.bash_profile
# KAFKA_HOME
export KAFKA_HOME/export/servers/kafka_2.11-0.11.0.0
export PATH$PATH:$KAFKA_HOME/bin
4启动kafka集群 启动 bin/kafka-server-start.sh config/server.properties 关闭 bin/kafka-server-stop.sh stop
二、使用kafkatool工具操作Kafka
这里提供一篇详细操作https://www.cnblogs.com/frankdeng/p/9452982.html 三、Kafka与SpringBoot集成
1、pom.xml导入
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.3.0.RELEASE/versionrelativePath/ !-- lookup parent from repository --/parentgroupIdcom.sunyard.bigdata/groupIdartifactIdspringbootkafka/artifactIdversion0.0.1-SNAPSHOT/versionnamespringbootkafka/namedescriptionDemo project for Spring Boot/descriptionpropertiesjava.version1.8/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scopeexclusionsexclusiongroupIdorg.junit.vintage/groupIdartifactIdjunit-vintage-engine/artifactId/exclusion/exclusions/dependencydependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka-test/artifactIdscopetest/scope/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project
2、application.properties配置
server.port9001
spring.application.namekafka#### kafka配置生产者 begin ####
# 指定kafka server的地址集群配多个中间逗号隔开
spring.kafka.bootstrap-serversxxx.xxx.xxx.1:9092,xxx.xxx.xxx.2:9092,xxx.xxx.xxx.3:9092
# 写入失败时重试次数。当leader失效一个repli节点会替代成为leader节点此时可能出现写入失败
# 当retris为0时produce不会重复。retirs重发此时repli节点完全成为leader节点不会产生消息丢失。
spring.kafka.producer.retries0
# 每次批量发送消息的数量,produce积累到一定数据一次发送
spring.kafka.producer.batch-size16384
# produce积累数据一次发送缓存大小达到buffer.memory就发送数据
spring.kafka.producer.buffer-memory33554432
#procedure要求leader在考虑完成请求之前收到的确认数用于控制发送记录在服务端的持久化其值可以为如下
#acks 0 如果设置为零则生产者将不会等待来自服务器的任何确认该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下无法保证服务器已收到记录并且重试配置将不会生效因为客户端通常不会知道任何故障为每条记录返回的偏移量始终设置为-1。
#acks 1 这意味着leader会将记录写入其本地日志但无需等待所有副本服务器的完全确认即可做出回应在这种情况下如果leader在确认记录后立即失败但在将数据复制到所有的副本服务器之前则记录将会丢失。
#acks all 这意味着leader将等待完整的同步副本集以确认记录这保证了只要至少一个同步副本服务器仍然存活记录就不会丢失这是最强有力的保证这相当于acks -1的设置。
#可以设置的值为all, -1, 0, 1
spring.kafka.producer.acks1
# 指定消息key和消息体的编解码方式
spring.kafka.producer.key-serializerorg.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializerorg.apache.kafka.common.serialization.StringSerializer
#### kafka配置生产者 end ######## kafka配置消费者 start ####
# 指定默认消费者group id -- 由于在kafka中同一组中的consumer不会读取到同一个消息依靠groud.id设置组名
spring.kafka.consumer.group-idtest1
# smallest和largest才有效如果smallest重新0开始读取如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
spring.kafka.consumer.auto-offset-resetearliest
# enable.auto.commit:true -- 设置自动提交offset
spring.kafka.consumer.enable-auto-committrue
#如果enable.auto.commit为true则消费者偏移自动提交给Kafka的频率以毫秒为单位默认值为5000。
spring.kafka.consumer.auto-commit-interval1000
# 指定消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializerorg.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializerorg.apache.kafka.common.serialization.StringDeserializer
#### kafka配置消费者 end ####
3、启动类代码
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;SpringBootApplication
EnableKafka
public class SpringbootkafkaApplication {public static void main(String[] args) {SpringApplication.run(SpringbootkafkaApplication.class, args);}
}
4、生产者代码
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;RestController
RequestMapping(/api/kafka/)
public class KafkaController {Autowiredprivate KafkaTemplateString, Object kafkaTemplate;GetMapping(send)ResponseBodypublic boolean send(RequestParam String message) {try {kafkaTemplate.send(test-topic, message);kafkaTemplate.send(test-topic2, message);System.out.println(消息发送成功...);} catch (Exception e) {e.printStackTrace();}return true;}GetMapping(test)ResponseBodypublic String test() {System.out.println(hello world!);return ok;}
}
5、消费者代码
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;Component
public class ConsumerListener {KafkaListener(topics test-topic)public void onMessage1(String message) {System.out.println(我是第一个消费者: message);}KafkaListener(topics test-topic2)public void onMessage2(String message) {System.out.println(我是第二个消费者: message);}
}
