Centos_PHP に Kafka クラスターをインストールするチュートリアル-PHPチュートリアル-php.cn

Centos に Kafka クラスターをインストールします

インストールの準備:
バージョン
Kafka バージョン: kafka_2.11-0.9.0.0
Zookeeper バージョン:zookeeper-3.4.7
Zookeeper クラスター: bjrenrui0001 bjrenrui0002 bjrenrui0003
Zookeeper クラスター構築については、以下を参照してください。 CentOS に ZooKeeper クラスターをインストールします

物理環境
3 つの物理マシンをインストールします:
192.168.100.200 bjrenrui0001 (3 つのブローカーを実行)
192.168.100.201 bjrenrui0002 (2 つのブローカーを実行)
192.168。 100.202 bj renrui0003 (2 つのブローカーを実行)
作成このクラスターは主に 3 つのステップに分かれています。単一ノード単一ブローカー、単一ノード複数ブローカー、複数ノード複数ブローカー

単一ノード単一ブローカー
このセクションでは例として bjrenrui0001 にブローカーを作成します
kafka をダウンロード:
ダウンロードパス: http://kafka.apache.org/downloads.html
cd /mq/
wget http://mirrors.hust.edu.cn/apache/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz
コピーファイル .sh kafka_2.11-0.9.0.0.tgz bjyfnbserver /mq/
tar zxvf kafka_2.11-0.9.0.0.tgz -C /mq/
ln -s /mq/kafka_2.11-0.9.0.0 /mq/ kafka
mkdir /mq/kafka/logs

構成
config/server.properties を変更します
vi /mq/kafka/config/server.properties
broker.id=1
listeners=PLAINTEXT://:9092
port=9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs = /mq/kafka/logs/kafka-logs
num.partitions=10
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log。 Retention .Check.Interval.ms = 300000SLOG.CLEANER.ENABLE =false
ZooKeeper.Connect= bjrenrui0001：2181、bjrenrui0002：2181、bjrenrui0003：2181gg
Zookeeper.Connection.msms=6000

startKafka Service c d /mq/kafka;sh bin/kafka-server-start.sh -daemon config/server.properties
または
sh /mq/kafka/bin/kafka-server-start.sh -daemon /mq/kafka/config/ server .properties

netstat -ntlp|grep -E '2181|9092'
(すべてのプロセスを識別できたわけではありません。所有されていないプロセス情報
は表示されません。すべてを表示するには root になる必要があります。)
tcp6 0 0 :::9092 :::* 聞く 26903/java
tcp6 0 0 :::2181 :::* 聞く 24532/java

トピックの作成:
sh /mq/kafka/bin/kafka-topics.sh - -create --zookeeper bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181 --replication-factor 1 --partitions 1 --topic test

トピックの表示:
sh /mq/kafka/bin/kafka-topics.sh - -list --zookeeper bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181

プロデューサーがメッセージを送信します:
$ sh /mq/kafka/bin/kafka-console-Producer.sh --broker-list bjrenrui0001:9092 --トピックテスト
最初の
メッセージ

コンシューマがメッセージを受信します:
$ sh bin/kafka-console-consumer.sh --zookeeper bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181 --topic test --from-beginning
first
メッセージ
最新のデータが必要な場合は、--from-beginning パラメーターなしで実行できます。

単一ノードの複数のブローカー
前の章のフォルダーをさらに 2 つのコピーにコピーします: kafka_2、kafka_3
cp -r /mq/kafka_2.11-0.9.0.0 /mq/kafka_2.11-0.9.0.0_2
cp - r /mq/kafka_2.11-0.9.0.0 /mq/kafka_2.11-0.9.0.0_3
ln -s /mq/kafka_2.11-0.9.0.0_2 /mq/kafka_2
ln -s /mq /kafka_2。 11-0.9.0.0_3 /mq/kafka_3

一意性を確保するために、kafka_2/config/server.properties ファイルと kafka_3/config/server.properties ファイルの Broker.id 属性と port 属性をそれぞれ変更します
vi /mq /kafka_2/config /server.properties
broker.id=2
listeners=PLAINTEXT://:9093
port=9093
host.name=bjrenrui0001
num.network.threads=3
num.io.threads=8
socket.send。 buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/mq/kafka_2/logs/kafka-logs
num.partitions=10
num.recovery .threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect =bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181
zookeeper.connection.timeout.ms=6000

vi /mq/kafka_3/config/server.properties
broker.id=3
listeners=PLAINT EXT: //: 9094
port=9094
host.name=bjrenrui0001
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
ソケット。 request.max.bytes=104857600
log.dirs=/mq/kafka_3/logs/kafka-logs
num.partitions=10
num.recovery.threads.per.data.dir=1
log.retention=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181
ゾーキーパー。接続。 timeout.ms=6000

開始
他の 2 つのブローカーを開始します:
sh /mq/kafka_2/bin/kafka-server-start.sh -daemon /mq/kafka_2/config/server.properties
sh /mq/kafka_3/bin/kafka-server-start.sh -daemon /mq/kafka_3/config/server.properties

检查端末口:
[dreamjobs@bjrenrui0001 config]$ netstat -ntlp|grep -E '2181| 909[2-9]'|sort -k3
(すべてのプロセスを識別できたわけではありません。所有されていないプロセス情報
は表示されません。すべてを表示するには root になる必要があります。)
tcp6 0 0 ::: 2181 :::* 聞いてください 24532/java
tcp6 0 0 :::9092 :::* 聞いてください 26903/java
tcp6 0 0 :::9093 :::* 聞いてください 28672/java
tcp6 0 0 :::9094 : ::* リッスン 28734/java

创建一レプリケーション係数 3 のトピック:
sh /mq/kafka/bin/kafka-topics.sh --create --zookeeper bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181 - -replication-factor 3 --partitions 1 --topic my-replicated-topic

查見るトピックの状態:
$ sh /mq/kafka/bin/kafka-topics.sh --describe -zookeeper bjrenrui0001:2181,bjrenrui0002: 2181,bjrenrui0003:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic:my-replicated-topic パーティション:0 リーダー:3 レプリカ:3、1、2 Isr: 3,1,2

上の内容からわかるように、このトピックには 1 つのパートが含まれており、レプリケーション係数は 3 であり、Node3 はリーダーです
解释例:
"リーダー" は、すべての読み取りと書き込みを担当するノードです。指定されたパーティション。各ノードは、ランダムに選択されたパーティション部分のリーダーになります。
「レプリカ」は、リーダーであるかどうか、または現在アクティブであるかどうかに関係なく、このパーティションのログを複製するノードのリストです。
「isr」「同期」レプリカのセットです。これは、現在存続しており、リーダーに追いついたレプリカリストのサブセットです。

再び見る次の前に作成されたテストトピック、下の図からレプリケーションの実行が見られません
$ sh /mq/kafka/bin/kafka-topics.sh --describe --zookeeper bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003 :2181 --topic test
Topic:test PartitionCount:1 ReplicationFactor:1 Configs:
Topic:test Partition:0 Leader:1 Replicas:1 Isr:1

多个节点的多个ブローカー
在bjrenrui0002、bjrenrui0003上分别ダウンロードされたファイルを kafka_4、kafka_5、kafka_6 の 2 つのファイルに解凍し、bjrenrui0001 上のserver.properties 構成ファイルをこの 3 つのファイルに保存します
vi /mq/kafka_4/config/server.properties
broker .id=4
listeners=PLAINTEXT://:9095
port=9095
host.name=bjrenrui0002
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/mq/kafka_4/logs/kafka-logs
num.partitions=10
num.recovery.threads.per.data .dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=bjrenrui0001:2181,bjrenrui0002 :2181,bjrenrui0003:2181
zookeeper.connection.timeout.ms=6000

vi /mq/kafka_5/config/server.properties
broker.id=5
listeners=PLAINTEXT://:9096
port=9096
host .name=bjrenrui0002
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/mq/kafka_5/logs/kafka-logs
num.partitions=10
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes= 1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181
zookeeper.connection.timeout.ms=6000

vi /mq/kafka_6/config/server.properties
broker.id=6
listeners=PLAINTEXT://:9097
port=9097
host.name=bjrenrui0003
num.network.threads=3
num.io.threads= 8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/mq/kafka_6/logs/kafka-logs
num.partitions =10
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable =false
zookeeper.connect=bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181
zookeeper.connection.timeout.ms=6000

vi /mq/kafka_7/config/server.properties
broker.id=7
listeners= PLAINTEXT://:9098
port=9098
host.name=bjrenrui0003
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes =1048576
socket.request.max.bytes=104857600
log.dirs=/mq/kafka_7/logs/kafka-logs
num.partitions=10
num.recovery.threads.per.data.dir=1
log。 Retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181
Zookeeper.connection.timeout.ms=6000

启アニメーション服务
sh /mq/kafka/bin/kafka-server-start.sh -daemon /mq/kafka/config/server.properties
sh /mq/kafka_2/bin/kafka-server-start.sh -daemon /mq/kafka_2/ config/server.properties
sh /mq/kafka_3/bin/kafka-server-start.sh -daemon /mq/kafka_3/config/server.properties

sh /mq/kafka_4/bin/kafka-server-start.sh -daemon /mq/kafka_4/config/server.properties
sh /mq/kafka_5/bin/kafka-server-start.sh -daemon /mq/kafka_5/config/server.properties

sh /mq/kafka_6/bin/ kafka-server-start.sh -daemon /mq/kafka_6/config/server.properties
sh /mq/kafka_7/bin/kafka-server-start.sh -daemon /mq/kafka_7/config/server.properties

check ：
$ netstat -ntlp|grep -E '2181|909[2-9]'|sort -k3

サービスを停止します:
sh /mq/kafka/bin/kafka-server-stop.sh
を使用する場合停止するスクリプトブローカーサービスは単一ノード上の複数のブローカーサービスを停止するため、注意してください。！！
ps grep -i 'kafka.Kafka' | grep java | grep -v grep | 3 つの物理マシン上で 7 つのブローカーが開始されました。 dreamjobs@bjrenrui0001 bin]$ netstat -ntlp|grep -E '2181|909[2-9]'|sort -k3
(すべてのプロセスを識別できたわけではありません。所有されていないプロセス情報
は表示されません。すべて見るには root になる必要があります。)
tcp6 0 0 :::2181 :::* LISTEN 24532/java
tcp6 0 0 :::9092 :::* LISTEN 33212/java
tcp6 0 0 :: :9093 : ::* 聞いてください 32997/java
tcp6 0 0 :::9094 :::* 聞いてください 33064/java

[dreamjobs@bjrenrui0002 config]$ netstat -ntlp|grep -E '2181|909[2-9 ]'| sort -k3
(すべてのプロセスを識別できるわけではありません。所有されていないプロセス情報
は表示されません。すべてを表示するには root になる必要があります。)
tcp6 0 0 :::2181 :::* LISTEN 6899/ java
tcp6 0 0 :::9095 :::* 聞いてください 33251/java
tcp6 0 0 :::9096 :::* 聞いてください 33279/java

[dreamjobs@bjrenrui0003 config]$ netstat -ntlp|grep -E ' 2181|909[2-9]'|sort -k3
(すべてのプロセスを識別できたわけではありません。所有されていないプロセス情報
は表示されません。すべてを表示するには root になる必要があります。)
tcp 0 0 0.0 .0.0:2181 0.0.0.0:* リッスン 14562/java
tcp 0 0 0.0.0.0:9097 0.0.0.0:* リッスン 23246/java
tcp 0 0 0.0.0.0:9098 0.0.0.0:* リッスン 23 270 /java

プロデューサーがメッセージを送信:
$ sh /mq/kafka/bin/kafka-console-Producer.sh --broker-list bjrenrui0001:9092 --topic my-replicated-topic

コンシューマーがメッセージを受信:
$ sh /mq /kafka_4/bin/kafka-console-consumer.sh --zookeeper bjrenrui0001:2181,bjrenrui0002:2181,bjrenrui0003:2181 --topic my-replicated-topic --from-beginning

http://www.bkjia.com/PHPjc/1091524.html