网络网站开发设计,安徽建设工程信息网怎么打不开了,南京中石化第五建设有限公司,用dw做网站怎么换行Apache Kafka是一个开源分布式事件流平台#xff0c;被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序
Kafka 官网#xff1a;https://kafka.apache.org/ Kafka 在2.8版本之后#xff0c;移除了对Zookeeper的依赖#xff0c;将依赖于ZooKeeper的控制器…Apache Kafka是一个开源分布式事件流平台被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序
Kafka 官网https://kafka.apache.org/ Kafka 在2.8版本之后移除了对Zookeeper的依赖将依赖于ZooKeeper的控制器改造成了基于Kafka Raft的Quorm控制器因此可以在不使用ZooKeeper的情况下实现集群
本文讲解 Kafka KRaft 模式集群搭建
笔者使用3台服务器它们的 ip 分别是 192.168.3.232、192.168.2.90、192.168.2.11 目录
1、官网下载 Kafka
2、配置 Kafka
3、创建 KRaft 集群
4、启动 Kafka KRaft 集群
5、关闭 Kafka KRaft 集群
6、测试 KRaft 集群 1、官网下载 Kafka
这里笔者下载最新版3.6.0 下载完成 将kafka分别上传到3台linux 在3台服务器上分别创建 kafka 安装目录
mkdir /usr/local/kafka
在3台服务器上分别将 kafka 安装包解压到新创建的 kafka 目录
tar -xzf kafka_2.13-3.6.0.tgz -C /usr/local/kafka 2、配置 Kafka
进入配置目录
cd /usr/local/kafka/kafka_2.13-3.6.0/config/kraft 编辑配置文件
vi server.properties
server.properties 配置说明 node.id 是kafka的broker节点id
controller.quorum.voters 配置的是 kafka 集群中的其他节点kafka Controller的投票者配置定义了一组Controller节点其中包括它们各自的 id 和网络地址
advertised.listeners 是节点自己的监听地址 192.168.3.232 节点配置
node.id 1 192.168.2.90 节点配置
node.id 2 192.168.2.11节点配置
node.id 3 3、创建 KRaft 集群
生成集群id
在任意一个节点上执行就行笔者使用 192.168.3.232 节点
进入bin 目录
cd /usr/local/kafka/kafka_2.13-3.6.0/bin
执行生成集群 id 命令
./kafka-storage.sh random-uuid 生成后保存生成的字符串 82vqfbdSTO2QzS_M0Su1Bw
然后分别在3台机器上执行下面命令
为方便执行命令先回到 kafka安装目录
cd /usr/local/kafka/kafka_2.13-3.6.0
再执行命令完成集群元数据配置
bin/kafka-storage.sh format -t 82vqfbdSTO2QzS_M0Su1Bw -c config/kraft/server.properties 192.168.3.232 节点 192.168.2.90 节点 192.168.2.11节点 上面命令执行完成后开放防火墙端口
kafka 需要开放 9092 端口和 9093 端口
3台机器上分别开放 9092 和 9093 端口
查看开放端口
firewall-cmd --zonepublic --list-ports 开放9092 端口
firewall-cmd --zonepublic --add-port9092/tcp --permanent 开放9093 端口
firewall-cmd --zonepublic --add-port9093/tcp --permanent
更新防火墙规则无需断开连接动态添加规则
firewall-cmd --reload 4、启动 Kafka KRaft 集群
在3台机器上分别启动
下面2个命令均可启动
bin/kafka-server-start.sh -daemon config/kraft/server.properties
或
bin/kafka-server-start.sh config/kraft/server.properties
笔者使用第二个启动命令 启动效果看下图 当 3 个节点都出现 Kafka Server started集群启动成功 5、关闭 Kafka KRaft 集群
关闭命令
bin/kafka-server-stop.sh
在 3 个节点上分别执行关闭命令 6、测试 KRaft 集群
新建 maven 项目添加 Kafka 依赖
dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion3.6.0/version
/dependency
笔者新建 maven项目 kafka-learn
kafka-learn 项目 pom 文件
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.wsjzzcbq/groupIdartifactIdkafka-learn/artifactIdversion1.0-SNAPSHOT/versiondependenciesdependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion3.6.0/version/dependency/dependenciesbuildpluginspluginartifactIdmaven-compiler-plugin/artifactIdconfigurationsource11/sourcetarget11/target/configuration/plugin/plugins/build
/project
新建生产者 ProducerDemo
package com.wsjzzcbq;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;/*** Demo** author wsjz* date 2023/11/24*/
public class ProducerDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {Properties properties new Properties();//配置集群节点信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.3.232:9092,192.168.2.90:9092,192.168.2.11:9092);//配置序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());ProducerString, String producer new KafkaProducer(properties);//topic 名称是demo_topicProducerRecordString, String producerRecord new ProducerRecord(demo_topic, 明月别枝惊鹊);RecordMetadata recordMetadata producer.send(producerRecord).get();System.out.println(recordMetadata.topic());System.out.println(recordMetadata.partition());System.out.println(recordMetadata.offset());}
}新建消费者 ConsumerDemo
package com.wsjzzcbq;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;/*** ConsumerDemo** author wsjz* date 2023/11/24*/
public class ConsumerDemo {public static void main(String[] args) {Properties properties new Properties();// 配置集群节点信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.3.232:9092,192.168.2.90:9092,192.168.2.11:9092);// 消费分组名properties.put(ConsumerConfig.GROUP_ID_CONFIG, demo_group);// 序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());KafkaConsumerString,String consumer new KafkaConsumer(properties);// 消费者订阅主题consumer.subscribe(Arrays.asList(demo_topic));while (true) {ConsumerRecordsString,String records consumer.poll(Duration.ofMillis(1000));for (ConsumerRecordString,String record:records) {System.out.printf(收到消息partition%d, offset%d, key%s, value%s%n,record.partition(),record.offset(),record.key(),record.value());}}}
}运行测试
效果图 消息成功发送并成功消费 至此完
