鯤鵬麒麟安裝Kafka-v1.1.1

因項目需要在鯤鵬麒麟服務器上安裝Kafka v1.1.1，因此這里將安裝配置過程記錄下來。

環境說明

# 查看系統相關詳細信息
[root@test kafka_2.12-1.1.1]# uname -a
Linux test.novalocal 4.19.148+ #1 SMP Mon Oct 5 22:04:46 EDT 2020 aarch64 aarch64 aarch64 GNU/Linux
# 查看操作系統版本信息
[root@test kafka_2.12-1.1.1]# cat /etc/kylin-release 
Kylin Linux Advanced Server release V10 (Tercel)
# 查看邏輯CPU數量
[root@test kafka_2.12-1.1.1]# cat /proc/cpuinfo| grep "processor"| wc -l
32
# 查看CPU信息
[root@test kafka_2.12-1.1.1]# lscpu
Architecture:                    aarch64
CPU op-mode(s):                  64-bit
Byte Order:                      Little Endian
CPU(s):                          32
On-line CPU(s) list:             0-31
Thread(s) per core:              1
Core(s) per socket:              16
Socket(s):                       2
NUMA node(s):                    2
Vendor ID:                       HiSilicon
Model:                           0
Model name:                      Kunpeng-920
Stepping:                        0x1
CPU max MHz:                     2400.0000
CPU min MHz:                     2400.0000
BogoMIPS:                        200.00
L1d cache:                       2 MiB
L1i cache:                       2 MiB
L2 cache:                        16 MiB
L3 cache:                        64 MiB
NUMA node0 CPU(s):               0-15
NUMA node1 CPU(s):               16-31
Vulnerability Itlb multihit:     Not affected
Vulnerability L1tf:              Not affected
Vulnerability Mds:               Not affected
Vulnerability Meltdown:          Not affected
Vulnerability Spec store bypass: Not affected
Vulnerability Spectre v1:        Mitigation; __user pointer sanitization
Vulnerability Spectre v2:        Not affected
Vulnerability Srbds:             Not affected
Vulnerability Tsx async abort:   Not affected
Flags:                           fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma dcpop asimddp asimdfhm
# 查看Java的版本信息
[root@test kafka_2.12-1.1.1]# java -version
openjdk version "1.8.0_242"
OpenJDK Runtime Environment (build 1.8.0_242-b08)
OpenJDK 64-Bit Server VM (build 25.242-b08, mixed mode)

下載

從apache官網下載相應的版本信息，地址：https://kafka.apache.org/downloads，找到1.1.1版本下載即可，這里下載kafka_2.12-1.1.1.tgz，下載地址：https://archive.apache.org/dist/kafka/1.1.1/kafka_2.12-1.1.1.tgz
這里的版本號有兩段，前一段為Scala的版本號，后一段為Kafka的版本，現在最新版本已經到3.X，這里用1.1.1，主要是項目特殊需要，建議還是用高版本的。

Scala 2.11 和 Scala 2.12 是 Scala 編程語言的兩個主要版本，它們之間存在一些關鍵的區別
1. 性能優化Scala 2.12 在性能方面做了很多優化，特別是在 JVM 上。它引入了值類（Value Classes）的改進，減少了運行時的開銷。Scala 2.11 的性能相對較低，但它的優化主要集中在穩定性和兼容性上。
2. 字符串插值Scala 2.12 引入了新的字符串插值語法，使用 s 前綴，例如 s"Hello, $name"。Scala 2.11 使用的是舊的字符串插值語法，使用 #{}，例如 "Hello, #{name}"。
3. 隱式轉換Scala 2.12 對隱式轉換進行了一些改進，使其更加安全和易于理解。Scala 2.11 的隱式轉換機制相對較為復雜，容易引起混淆。
4. 模塊化Scala 2.12 引入了模塊化系統，允許開發者將代碼分割成多個模塊，便于管理和維護。Scala 2.11 沒有模塊化系統，所有代碼都放在一個項目中。
5. 兼容性Scala 2.12 相對于 Scala 2.11 有一些不兼容的更改，特別是在 API 和庫的使用上。因此，從 Scala 2.11 遷移到 Scala 2.12 可能需要一些工作。Scala 2.11 是一個長期支持（LTS）版本，這意味著它將獲得更長時間的支持和維護。
6. 社區支持Scala 2.12 是目前的主流版本，得到了廣泛的社區支持和庫的更新。Scala 2.11 雖然仍然在使用，但社區支持逐漸減少。
總結如果你正在開發一個新的項目，建議使用 Scala 2.12 或更高版本，以獲得更好的性能和更多的功能。如果你正在維護一個現有的 Scala 2.11 項目，可以考慮在適當的時候遷移到 Scala 2.12，但需要注意兼容性問題。

部署

1、解壓文件執行
tar -zxvf kafka_2.12-1.1.1.tgz
2、解壓后路徑為
[root@test kafka_2.12-1.1.1]# pwd
/data/public/kafka/kafka_2.12-1.1.1
3、修改config/server.properties，注意修改listeners和advertised.listeners即可

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.# see kafka.server.KafkaConfig for additional details and defaults############################# Server Basics ############################## The id of the broker. This must be set to a unique integer for each broker.
broker.id=0############################# Socket Server Settings ############################## The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://192.138.31.100:9092# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://192.138.31.100:9092# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=409600000############################# Log Basics ############################## A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1############################# Log Flush Policy ############################## Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000############################# Log Retention Policy ############################## The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000############################# Zookeeper ############################## Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000############################# Group Coordinator Settings ############################## The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0

啟動

1、先啟動Zookeeper，執行
nohup /data/public/kafka/kafka_2.12-1.1.1/bin/zookeeper-server-start.sh /data/public/kafka/kafka_2.12-1.1.1/config/zookeeper.properties > /dev/null 2>&1 &

2、然后啟動Kafka，執行
nohup /data/public/kafka/kafka_2.12-1.1.1/bin/kafka-server-start.sh /data/public/kafka/kafka_2.12-1.1.1/config/server.properties > /dev/null 2>&1 &

3、檢查是否啟動

[root@test kafka_2.12-1.1.1]# netstat -nltp | grep -E '(2181|9092)'
tcp6       0      0 192.168.31.100:9092     :::*                    LISTEN      970022/java         
tcp6       0      0 :::2181                 :::*                    LISTEN      966577/java

表明Kafka已經啟動起來，其中2181為Zookeeper的端口，9092為Kafka的端口