仿门户网站多功能js相册画廊源码,网站开发容易做吗,wordpress优惠券插件,亚马逊跨境电商入门完整教程安装并配置单机版kafka所需环境
wsl2 环境可用性较高#xff0c;如下介绍在该环境中安装单机版本kafka的详细过程。
启动命令行工具启动wsl#xff1a;wsl --user root --cd ~#xff0c;#xff08;以root用户启动#xff0c;进入wsl后当前路径为~“用户主目录”#…安装并配置单机版kafka所需环境
wsl2 环境可用性较高如下介绍在该环境中安装单机版本kafka的详细过程。
启动命令行工具启动wslwsl --user root --cd ~以root用户启动进入wsl后当前路径为~“用户主目录”安装java进入https://learn.microsoft.com/zh-cn/java/openjdk/download选择相应的java版本下载接口 创建java的安装路径mkdir -p /opt/sdk/java将刚刚下载的javasdk压缩包移动进创建的路径mv /mnt/c/Users/你的用户名/Downloads/microsoft-jdk-21.0.1-linux-aarch64.tar.gz /opt/sdk/java/切换到你java安装路径cd /opt/sdk/java/解压安装tar -zxvf microsoft-jdk-21.0.1-linux-aarch64.tar.gz可以选择性删除你刚刚的压缩包rm -rf microsoft-jdk-21.0.1-linux-aarch64.tar.gz配置环境变量 进入jdk的解压目录cd cd jdk-21.0.112 查看一下当前的绝对路径pwd 复制一下绝对路径的字符串使用vim 打开 /etc/profileGG到文件尾部在文件尾部追加如下内容export JAVA_HOME/opt/sdk/java/jdk-21.0.112
export PATH$PATH:$JAVA_HOME/bin注意JAVA_HOME后边的jdk_xxx是你自己下载的相应版本信息让profile文件生效source /etc/profile查看java是否安装成功java -version显示版本信息则安装成功 rootOphelia:/opt/sdk/java/jdk-21.0.112# java -versionopenjdk version 21.0.1 2023-10-17 LTSOpenJDK Runtime Environment Microsoft-8526870 (build 21.0.112-LTS)OpenJDK 64-Bit Server VM Microsoft-8526870 (build 21.0.112-LTS, mixed mode)安装kafka 在kafka官网下载相应版本kafka我这里选最新版Scala 2.13 - kafka_2.13-3.6.0.tgz (asc, sha512)创建kafka的安装路径mkdir -p /opt/software/kafka将刚刚下载的kafka压缩包移动进创建的路径mv /mnt/c/Users/你的用户名/Downloads/kafka_2.13-3.6.1.tgz /opt/software/kafka/cd /opt/software/kafka/tar -zxvf kafka_2.13-3.6.1.tgzcd kafka_2.13-3.6.1/pwd然后复制绝对路径vim /etc/profile 追加如下内容 # Kafkaexport KAFKA_HOME/opt/software/kafka/kafka_2.13-3.6.1export PATH$PATH:$KAFKA_HOME/binsource /etc/profile配置文件修改 vim /opt/software/kafka/kafka_2.13-3.6.1/config/server.properties 找到log.dirs/tmp/kafka-logs将其修改为log.dirs/opt/software/kafka/kafka-logs如下所示 # log.dirs/tmp/kafka-logslog.dirs/opt/software/kafka/kafka-logsmkdir -p /opt/software/kafka/kafka-logs上述操作目的是防止topic中的数据丢失
启动kafka
启动kafka之前先启动zookeeper Kafka 2.8.0 版本开始内置了zookeeper。
启动zookeeper zookeeper-server-start.sh -daemon /opt/software/kafka/kafka_2.13-3.6.1/config/zookeeper.properties 命令看起来有点长实际上一点儿也不短。 它分为3段[zookeeper-server-start.sh][-daemon][配置文件路径] -daemon参数是以守护进程的方式启动。zookeeper的启动脚本也存储在kafka的环境变量指向的路径中。 启动kafka kafka-server-start.sh -daemon /opt/software/kafka/kafka_2.13-3.6.1/config/server.properties
关闭kafka
先停止kafka后停止zookeeper
rootOphelia:~# kafka-server-stop.sh
rootOphelia:~# zookeeper-server-stop.sh当停用kafka集群的时候也是先停止kafka后停止zookeeper顺序搞反了会导致kafka无法停止只能使用kill去杀死。
kafka-topics.sh 主题相关操作
查看一下kafka-tpics.sh命令都能干什么
rootOphelia:~# kafka-topics.sh
Create, delete, describe, or change a topic.
Option Description
------ -----------
--alter Alter the number of partitions andreplica assignment. Update theconfiguration of an existing topicvia --alter is no longer supportedhere (the kafka-configs CLI supportsaltering topic configs with a --bootstrap-server option).
--at-min-isr-partitions if set when describing topics, onlyshow partitions whose isr count isequal to the configured minimum.
--bootstrap-server String: server to REQUIRED: The Kafka server to connectconnect to to.
--command-config String: command Property file containing configs to beconfig property file passed to Admin Client. This is usedonly with --bootstrap-server optionfor describing and altering brokerconfigs.
--config String: namevalue A topic configuration override for thetopic being created or altered. Thefollowing is a list of validconfigurations:cleanup.policycompression.typedelete.retention.msfile.delete.delay.msflush.messagesflush.msfollower.replication.throttled.replicasindex.interval.bytesleader.replication.throttled.replicaslocal.retention.byteslocal.retention.msmax.compaction.lag.msmax.message.bytesmessage.downconversion.enablemessage.format.versionmessage.timestamp.after.max.msmessage.timestamp.before.max.msmessage.timestamp.difference.max.msmessage.timestamp.typemin.cleanable.dirty.ratiomin.compaction.lag.msmin.insync.replicaspreallocateremote.storage.enableretention.bytesretention.mssegment.bytessegment.index.bytessegment.jitter.mssegment.msunclean.leader.election.enableSee the Kafka documentation for fulldetails on the topic configs. It issupported only in combination with --create if --bootstrap-server optionis used (the kafka-configs CLIsupports altering topic configs witha --bootstrap-server option).
--create Create a new topic.
--delete Delete a topic
--delete-config String: name A topic configuration override to beremoved for an existing topic (seethe list of configurations under the--config option). Not supported withthe --bootstrap-server option.
--describe List details for the given topics.
--exclude-internal exclude internal topics when runninglist or describe command. Theinternal topics will be listed bydefault
--help Print usage information.
--if-exists if set when altering or deleting ordescribing topics, the action willonly execute if the topic exists.
--if-not-exists if set when creating topics, theaction will only execute if thetopic does not already exist.
--list List all available topics.
--partitions Integer: # of partitions The number of partitions for the topicbeing created or altered (WARNING:If partitions are increased for atopic that has a key, the partitionlogic or ordering of the messageswill be affected). If not suppliedfor create, defaults to the clusterdefault.
--replica-assignment String: A list of manual partition-to-brokerbroker_id_for_part1_replica1 : assignments for the topic beingbroker_id_for_part1_replica2 , created or altered.broker_id_for_part2_replica1 :broker_id_for_part2_replica2 , ...
--replication-factor Integer: The replication factor for eachreplication factor partition in the topic beingcreated. If not supplied, defaultsto the cluster default.
--topic String: topic The topic to create, alter, describeor delete. It also accepts a regularexpression, except for --createoption. Put topic name in doublequotes and use the \ prefix toescape regular expression symbols; e.g. test\.topic.
--topic-id String: topic-id The topic-id to describe.This is usedonly with --bootstrap-server optionfor describing topics.
--topics-with-overrides if set when describing topics, onlyshow topics that have overriddenconfigs
--unavailable-partitions if set when describing topics, onlyshow partitions whose leader is notavailable
--under-min-isr-partitions if set when describing topics, onlyshow partitions whose isr count isless than the configured minimum.
--under-replicated-partitions if set when describing topics, onlyshow under replicated partitions
--version Display Kafka version.列出主要命令
选项描述信息–bootstrap-server String: server to connect to连接kafka Broker 主机名:端口号–topic String: topic操作的topic名称–create创建topic–delete删除topic–alter修改topic–list查看所有topic–describe查看topic的详细描述信息–partitions Integer: # of partitions设置分区数–replication-factor Integer: replication factor设置分区副本–config String: namevalue更细系统默认的配置
查看topic
kafka-topics.sh --list --bootstrap-server localhost:9092 或者查看指定的topic kafka-topics.sh --bootstrap-server localhost:9092 --topic 名称 --describe
生产环境中可能有多台机器为了保证高可用性防止访问的kafka挂掉之后无法查询数据可以指定多个实例如kafka-topics.sh --list --bootstrap-server localhost:9092 hostName1:9092 hostName2:9092
创建topic
创建之前先查看一下有多少个topic kafka-topics.sh --list --bootstrap-server localhost:9092 一看啥也木有看了个寂寞这就对溜你不创建topic它就没topic 创建单副本3个分区的 名 为 first的topic kafka-topics.sh --bootstrap-server localhost:9092 --topic first --create --partitions 3 --replication-factor 1
查看一下
rootOphelia:~# kafka-topics.sh --list --bootstrap-server localhost:9092
first或者
kafka-topics.sh --bootstrap-server localhost:9092 --topic first --describe
Topic: first TopicId: Db5XuRoASzi4W2FgQbaW7A PartitionCount: 3 ReplicationFactor: 1 Configs:Topic: first Partition: 0 Leader: 0 Replicas: 0 Isr: 0Topic: first Partition: 1 Leader: 0 Replicas: 0 Isr: 0Topic: first Partition: 2 Leader: 0 Replicas: 0 Isr: 0修改topic
first topic的分区数从3个修改成4个注意分区数在命令行中只能增加不能减少 kafka-topics.sh --bootstrap-server localhost:9092 --topic first --alter --partitions 4
尝试一下会报错的
rootOphelia:~# kafka-topics.sh --bootstrap-server localhost:9092 --topic first --alter --partitions 3
Error while executing topic command : Topic currently has 4 partitions, which is higher than the requested 3.
[2023-12-17 17:55:41,130] ERROR org.apache.kafka.common.errors.InvalidPartitionsException: Topic currently has 4 partitions, which is higher than the requested 3.(kafka.admin.TopicCommand$)删除topic
删除名为first的topic kafka-topics.sh --bootstrap-server localhost:9092 --topic first --delete
kafka-console-producer.sh生产者相关操作
rootOphelia:~# kafka-console-producer.sh
Missing required option(s) [bootstrap-server]
Option Description
------ -----------
--batch-size Integer: size Number of messages to send in a singlebatch if they are not being sentsynchronously. please note that thisoption will be replaced if max-partition-memory-bytes is also set(default: 16384)
--bootstrap-server String: server to REQUIRED unless --broker-listconnect to (deprecated) is specified. The server(s) to connect to. The broker liststring in the form HOST1:PORT1,HOST2:PORT2.
--broker-list String: broker-list DEPRECATED, use --bootstrap-serverinstead; ignored if --bootstrap-server is specified. The brokerlist string in the form HOST1:PORT1,HOST2:PORT2.
--compression-codec [String: The compression codec: either none,compression-codec] gzip, snappy, lz4, or zstd.If specified without value, then itdefaults to gzip
--help Print usage information.
--line-reader String: reader_class The class name of the class to use forreading lines from standard in. Bydefault each line is read as aseparate message. (default: kafka.tools.ConsoleProducer$LineMessageReader)
--max-block-ms Long: max block on The max time that the producer willsend block for during a send request.(default: 60000)
--max-memory-bytes Long: total memory The total memory used by the producerin bytes to buffer records waiting to be sentto the server. This is the option tocontrol buffer.memory in producerconfigs. (default: 33554432)
--max-partition-memory-bytes Integer: The buffer size allocated for amemory in bytes per partition partition. When records are receivedwhich are smaller than this size theproducer will attempt tooptimistically group them togetheruntil this size is reached. This isthe option to control batch.sizein producer configs. (default: 16384)
--message-send-max-retries Integer Brokers can fail receiving the messagefor multiple reasons, and beingunavailable transiently is just oneof them. This property specifies thenumber of retries before theproducer give up and drop thismessage. This is the option tocontrol retries in producerconfigs. (default: 3)
--metadata-expiry-ms Long: metadata The period of time in millisecondsexpiration interval after which we force a refresh ofmetadata even if we havent seen anyleadership changes. This is theoption to control metadata.max.age.ms in producer configs. (default:300000)
--producer-property String: A mechanism to pass user-definedproducer_prop properties in the form keyvalue tothe producer.
--producer.config String: config file Producer config properties file. Notethat [producer-property] takesprecedence over this config.
--property String: prop A mechanism to pass user-definedproperties in the form keyvalue tothe message reader. This allowscustom configuration for a user-defined message reader.Default properties include:parse.keyfalseparse.headersfalseignore.errorfalsekey.separator\theaders.delimiter\theaders.separator,headers.key.separator:null.marker When set, any fields(key, value and headers) equal tothis will be replaced by nullDefault parsing pattern when:parse.headerstrue and parse.keytrue:h1:v1,h2:v2...\tkey\tvalueparse.keytrue:key\tvalueparse.headerstrue:h1:v1,h2:v2...\tvalue
--reader-config String: config file Config properties file for the messagereader. Note that [property] takesprecedence over this config.
--request-required-acks String: The required acks of the producerrequest required acks requests (default: -1)
--request-timeout-ms Integer: request The ack timeout of the producertimeout ms requests. Value must be non-negativeand non-zero. (default: 1500)
--retry-backoff-ms Long Before each retry, the producerrefreshes the metadata of relevanttopics. Since leader election takesa bit of time, this propertyspecifies the amount of time thatthe producer waits before refreshingthe metadata. This is the option tocontrol retry.backoff.ms inproducer configs. (default: 100)
--socket-buffer-size Integer: size The size of the tcp RECV size. This isthe option to control send.buffer.bytes in producer configs.(default: 102400)
--sync If set message send requests to thebrokers are synchronously, one at atime as they arrive.
--timeout Long: timeout_ms If set and the producer is running inasynchronous mode, this gives themaximum amount of time a messagewill queue awaiting sufficient batchsize. The value is given in ms. Thisis the option to control linger.msin producer configs. (default: 1000)
--topic String: topic REQUIRED: The topic id to producemessages to.
--version Display Kafka version.生产消息
用生产者连接broker kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first 再启动一个命令行 wsl --user root --cd ~ 启动一个消费者 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first 生产者发送消息 消费者接收消息
kafka-console-consumer.sh消费者相关操作
rootOphelia:~# kafka-console-consumer.sh
This tool helps to read data from Kafka topics and outputs it to standard output.
Option Description------ -------------bootstrap-server String: server to REQUIRED: The server(s) to connect to.connect to--consumer-property String: A mechanism to pass user-definedconsumer_prop properties in the form keyvalue tothe consumer.--consumer.config String: config file Consumer config properties file. Notethat [consumer-property] takesprecedence over this config.
--enable-systest-events Log lifecycle events of the consumerin addition to logging consumedmessages. (This is specific forsystem tests.)--formatter String: class The name of a class to use forformatting kafka messages fordisplay. (default: kafka.tools.DefaultMessageFormatter)
--formatter-config String: config Config properties file to initializefile the message formatter. Note that[property] takes precedence overthis config.--from-beginning If the consumer does not already havean established offset to consumefrom, start with the earliestmessage present in the log ratherthan the latest message.
--group String: consumer group id The consumer group id of the consumer.
--help Print usage information.
--include String: Java regex (String) Regular expression specifying list oftopics to include for consumption.
--isolation-level String Set to read_committed in order tofilter out transactional messageswhich are not committed. Set toread_uncommitted to read allmessages. (default: read_uncommitted)
--key-deserializer String:deserializer for key--max-messages Integer: num_messages The maximum number of messages toconsume before exiting. If not set,consumption is continual.
--offset String: consume offset The offset to consume from (a non-negative number), or earliestwhich means from beginning, orlatest which means from end(default: latest)
--partition Integer: partition The partition to consume from.Consumption starts from the end ofthe partition unless --offset isspecified.--property String: prop The properties to initialize themessage formatter. Defaultproperties include:print.timestamptrue|falseprint.keytrue|falseprint.offsettrue|falseprint.partitiontrue|falseprint.headerstrue|falseprint.valuetrue|falsekey.separatorkey.separatorline.separatorline.separatorheaders.separatorline.separatornull.literalnull.literalkey.deserializerkey.deserializervalue.deserializervalue.deserializerheader.deserializerheader.deserializerUsers can also pass in customizedproperties for their formatter; morespecifically, users can pass inproperties keyed with key.deserializer., value.deserializer. and headers.deserializer. prefixes to configuretheir deserializers.
--skip-message-on-error If there is an error when processing amessage, skip it instead of halt.
--timeout-ms Integer: timeout_ms If specified, exit if no message isavailable for consumption for thespecified interval.
--topic String: topic The topic to consume on.
--value-deserializer String:deserializer for values--version Display Kafka version.
--whitelist String: Java regex DEPRECATED, use --include instead;(String) ignored if --include specified.Regular expression specifying listof topics to include for consumption.增量接收
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
接收历史全部数据 –from-beginning
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first --from-beginning 使用场景根据情况而定
