哪个公司做农村产权交易网站,ps制作网页,怎样创造自己的网站,cpa游戏推广平台Apache Kafka是一个开源分布式事件流平台#xff0c;被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序
Kafka 官网#xff1a;Apache Kafka 关于ZooKeeper的弃用
根据 Kafka官网信息#xff0c;随着Apache Kafka 3.5版本的发布#xff0c;Zookeeper现…Apache Kafka是一个开源分布式事件流平台被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序
Kafka 官网Apache Kafka 关于ZooKeeper的弃用
根据 Kafka官网信息随着Apache Kafka 3.5版本的发布Zookeeper现已被标记为已弃用。未来计划在Apache Kafka4.0版的下一个主要版本中删除ZooKeeper该版本最快将于2024年4月发布。在弃用阶段ZooKeeper仍然支持用于Kafka集群元数据的管理但不建议用于新的部署。新的部署方式使用 KRaft 模式KRaft 模式部署可以看笔者的文章《kafka 集群 KRaft 模式搭建》考虑到一些公司仍然在使用老版本的 Kafka故笔者写这篇文章记录 Kafka 集群Zookeeper 模式搭建
官网信息截图 笔者使用3台服务器它们的 ip 分别是 192.168.3.232、192.168.2.90、192.168.2.11 目录
1、官网下载 Kafka
2、配置 Kafka
3、启动 Kafka 集群
4、关闭 Kafka 集群
5、使用Kafka 可视化工具查看
6、测试Kafka集群 1、官网下载 Kafka
这里笔者下载最新版3.6.0
3.6.0 版本需要至少 java8 及以上版本笔者使用的是 java8 版本
关于 linux 安装 java没安装过的朋友可以参考《linux 系统安装 jdk》 下载完成 将 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 编辑配置文件 server.properties
vi server.properties
配置 broker.idadvertised.listenerszookeeper.connect
broker.id 每个节点的id
advertised.listeners 本机的外网访问地址
zookeeper.connect zookeeper 地址
192.168.3.232 节点配置 advertised.listeners 笔者配置为本机地址 192.168.2.90 节点 192.168.2.11 节点 笔者zookeeper 地址是 192.168.2.130:2181
zookeeper 版本是3.8.3
关于zookeeper单机安装和集群安装可以参考《Linux环境 安装 zookeeper》《windows环境 安装 zookeeper》《linux 使用 nginx 搭建 zookeeper 集群》 3、启动 Kafka 集群
首先启动 zookeeper
然后在3台机器上依次启动 Kafka
进入 kafka 目录
cd /usr/local/kafka/kafka_2.13-3.6.0
下面2个命令皆可
bin/kafka-server-start.sh config/server.properties
或
bin/kafka-server-start.sh -daemon config/server.properties 4、关闭 Kafka 集群
关闭命令
bin/kafka-server-stop.sh
在 3 个节点上分别执行关闭命令 5、使用Kafka 可视化工具查看
下载地址https://www.kafkatool.com/download.html
运行效果 6、测试Kafka集群
新建 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());}}}
}
运行测试
效果图 至此完
