Fork me on GitHub

分布式消息系统MQ之kafka部署

字数统计: 2.9k字   |   阅读时长≈ 14分

原理

kafka介绍

1
2
3
Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。

类似的消息系统还有RbbitMQ、ActiveMQ、ZeroMQ，最主要的优势是其具备分布式功能、并且结合zookeeper 可以实现动态扩容。

Apache Kafka与传统消息系统相比，有以下不同

1
2
3
4
它被设计为一个分布式系统，易于向外扩展；
它同时为发布和订阅提供高吞吐量；
它支持多订阅者，当失败时能自动平衡消费者；
它将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。

kafka基本组件

1
2
3
4
5
- 话题（Topic）: 是特定类型的消息流。
  消息是字节的有效负载（Payload），话题是消息的分类名或种子（Feed）名。
- 生产者（Producer）: 是能够发布消息到话题的任何对象。
- 已发布的消息保存在一组服务器中，它们被称为代理（Broker）或Kafka集群。
- 消费者可以订阅一个或多个话题，并从Broker拉数据，从而消费这些已发布的消息。

kafka整体架构

kafka整体架构

zookeeper

1
2
3
4
5
6
7
8
9
提到kafka就必须说说zookeeper集群，zookeeper集群特性：

zookeeper集群是否可用的依据是可用节点数量是否超过了集群的一半，zk集群中可用节点如果超过zookeeper集群节点总数量的一半，集群就认为是可用的(可用节点若等于集群的一半集群不可用)。

假如有2台服务器做了zookeeper集群，只要任何一台故障或宕机，整个集群就不可用，因为可用的节点没有大于集群一半的数量；但是假如集群中有3个节点，那么当1个节点故障，还剩2个节点，大于3个节点的一半，所以一个节点故障集群是正常运行的。但是再坏1个节点就只剩一个节点集群就不可用了。

假如有4台服务器做了zookeeper集群，最多只能损坏一个节点，若同时损坏两个，可用节点等于集群总节点的一半，此时集群是不可用的。

所以这也就是为什么一个zookeeper集群的节点数建议是奇数的原因，建议5台机器做zookeeper机器，业务量小3台也行。

部署

部署zookeeper集群(3台zookeeper节点)

1
2
3
4
5
6
7
8
9
10
11
12
官网: https://zookeeper.apache.org
下载最新稳定版zookeeper
wget https://archive.apache.org/dist/zookeeper/stable/zookeeper-3.4.10.tar.gz

解压
tar -xf zookeeper-3.4.10.tar.gz -C /usr/local/
cd /usr/local/
mv zookeeper-3.4.10 zookeeper
useradd zookeeper -s /sbin/nologin -M
chown -R zookeeper:zookeeper zookeeper
cd zookeeper
cp conf/zoo_sample.cfg conf/zoo.cfg
1
2
3
4
5
6
配置zookeeper环境变量
tail -1 /etc/profile
export PATH=$PATH:/usr/local/zookeeper/bin

source /etc/profile
env |grep zookeeper

配置zookeeper集群

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
创建日志和数据存放目录并授权
[root@localhost conf]# mkdir /var/log/zookeeper -p
[root@localhost conf]# chown -R zookeeper:zookeeper /var/log/zookeeper
[root@localhost zookeeper]# mkdir /usr/local/zookeeper/data -p
[root@localhost zookeeper]# chown -R zookeeper.zookeeper /usr/local/zookeeper/data

cd /usr/local/zookeeper/conf/
编辑配置文件zoo.cfg
[root@localhost conf]# egrep -v "^#|^$" zoo.cfg 
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper/data
clientPort=2181
dataLogDir=/var/log/zookeeper
server.1= 10.0.10.21:2888:3888
server.2= 10.0.10.22:2888:3888
server.3= 10.0.10.23:2888:3888

创建ServerID标识

1
2
3
4
5
6
7
8
9
10
注：zookeeper集群模式下还要配置一个myid文件,这个文件需要放在data目录下。

(1)在10.0.10.21服务器上创建myid文件,并设置值为1,同时需要与zoo.cfg文件里面的server.1对应
   echo "1" > /usr/local/zookeeper/data/myid
   
(2)在10.0.10.22服务器上创建myid文件,并设置值为2,同时需要与zoo.cfg文件里面的server.2对应
   echo "2" > /usr/local/zookeeper/data/myid
   
(3)在10.0.10.23服务器上创建myid文件,并设置值为3,同时需要与zoo.cfg文件里面的server.3对应
   echo "3" > /usr/local/zookeeper/data/myid

启动并验证zookeeper集群

1
2
3
4
5
分别启动三台zookeeper服务器服务
zkServer.sh start

在任意一台查看zookeeper集群服务状态(注：需要所有成员服务器均启动服务后才能查看，否则会报错)
zkServer.sh status
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
IP地址: 10.0.10.21/24
[root@localhost ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower

IP地址: 10.0.10.22/24
[root@localhost ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: leader

IP地址: 10.0.10.23/24
[root@localhost ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower

出现以上信息说明：zookeeper集群已经成功部署。

测试连接zookeeper集群

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
对于客户端来说，zookeeper集群是一个整体，连接到zookeeper集群实际上感觉在独享整个集群的服务，所以，你可以在任何一个结点上建立到服务集群的连接。例如：

在其中10.0.10.57节点使用zkCli.sh命令连接zookeeper服务器10.0.10.21:2181,如下
[root@localhost ~]# zkCli.sh -server 10.0.10.21:2181
Connecting to 10.0.10.21:2181
2017-11-28 02:12:42,232 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.10-39d3a4f269333c922ed3db283be479f9deacaa0f, built on 03/23/2017 10:13 GMT
2017-11-28 02:12:42,238 [myid:] - INFO  [main:Environment@100] - Client environment:host.name=localhost

WatchedEvent state:SyncConnected type:None path:null
[zk: 10.0.10.21:2181(CONNECTED) 0] 
[zk: 10.0.10.21:2181(CONNECTED) 0] ls /            
[zookeeper]
[zk: 10.0.10.21:2181(CONNECTED) 1] 
[zk: 10.0.10.21:2181(CONNECTED) 1]

到此有关zookeeper集群搭建就完全结束
注意: 生产环境需要考虑zookeeper自启配置,可参考：https://wuyanteng.github.io/2017/11/28/zookeeper%E9%9B%86%E7%BE%A4%E9%83%A8%E7%BD%B2/

部署kafka集群(3台zookeeper机器分别安装kafka)

1
2
3
4
下载：  https://kafka.apache.org

tar -xf kafka_2.11-1.0.0.tgz -C /usr/local
ln -s kafka_2.11-1.0.0 kafka

配置kafka环境变量

1
2
3
4
5
6
配置kafka环境变量
tail -1 /etc/profile
export PATH=$PATH:/usr/local/kafka/bin

source /etc/profile
env |grep kafka

节点1-kafka配置

1
2
3
4
5
[root@host21 local]# grep "^[a-Z]" /usr/local/kafka/config/server.properties
broker.id=1
listeners=PLAINTEXT://10.0.10.21:9092
log.retention.hours=24    #保留指定小时的日志内容
zookeeper.connect=10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181   #指定所有的kafka节点

节点2-kafka配置

1
2
3
4
5
[root@host21 local]# grep "^[a-Z]" /usr/local/kafka/config/server.properties
broker.id=2
listeners=PLAINTEXT://10.0.10.22:9092
log.retention.hours=24
zookeeper.connect=10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181

节点3-kafka配置

1
2
3
4
5
[root@host21 local]# grep "^[a-Z]" /usr/local/kafka/config/server.properties
broker.id=3
listeners=PLAINTEXT://10.0.10.23:9092
log.retention.hours=24
zookeeper.connect=10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181

在3个节点分别启动kafka

1
2
3
4
5
#以守护进程的方式启动
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

#查看日志
tailf /usr/local/kafka/logs/server.log

分别验证kafka进程

1
2
3
4
[root@host21 ~]# jps
784 Jps
30598 QuorumPeerMain
710 Kafka

测试Topic

创建名为logstashtest，partitions(分区)为3，replication(复制)为3的topic(主题)

1
2
3
4
在任意kafka服务器上操作：
[root@host21 ~]# kafka-topics.sh --create --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --partitions 3 --replication-factor 3 --topic logstashtest

Created topic "logstashtest".

测试获取topic

1
2
3
4
5
6
可以在任意kafka服务器上测试：
[root@host23 local]# kafka-topics.sh --describe --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --topic logstashtest
Topic:logstashtest	PartitionCount:3	ReplicationFactor:3	Configs:
	Topic: logstashtest	Partition: 0	Leader: 0	Replicas: 0,1,2	Isr: 0,1,2
	Topic: logstashtest	Partition: 1	Leader: 1	Replicas: 1,2,0	Isr: 1,2,0
	Topic: logstashtest	Partition: 2	Leader: 2	Replicas: 2,0,1	Isr: 2,0,1
1
2
状态说明：logstashtest这个topic，当前有三个分区分别为0、1、2
         分区0 有三个副本，并且状态都为lsr（ln-sync，表示可以参加选举成为leader）。

删除一个topic

1
2
3
4
5
6
7
8
9
10
11
12
创建一个名称为test的topic,来测试删除
[root@host21 ~]# kafka-topics.sh --create --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --partitions 3 --replication-factor 3 --topic test
Created topic "test".

删除名为test的topic
[root@host21 ~]# kafka-topics.sh --delete --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --topic test
Topic test is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

验证是否删除
[root@host21 ~]# kafka-topics.sh --describe --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --topic test
[root@host21 ~]#

获取所有topic

1
2
3
[root@host21 ~]# kafka-topics.sh --list --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 _consumer_offsets
logstashtest
[root@host21 ~]#

kafka命令测试消息发送

1
2
3
[root@host21 ~]# kafka-topics.sh --create --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --partitions 3 --replication-factor 3 --topic messagetest
Created topic "messagetest".
[root@host21 ~]#

发送消息

1
2
3
4
5
kafka-console-producer.sh --broker-list 10.0.10.21:9092,10.0.10.22:9092,10.0.10.23:9092 --topic logstashtest
>123
>456
>Hello Kafka
>

其他kafka服务器服务器测试获取消息

服务器A获取消息

1
2
3
4
5
[root@host21 ~]# kafka-console-consumer.sh --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --topic logstashtest --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
456
Hello Kafka
123

服务器B获取消息

1
2
3
4
5
[root@host22 local]# kafka-console-consumer.sh --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --topic logstashtest --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
456
Hello Kafka
123

服务器C获取消息

1
2
3
4
5
[root@host23 local]# kafka-console-consumer.sh --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23:2181 --topic logstashtest --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
Hello Kafka
456
123

使用logstash测试向kafka写入数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
在安装有logstash的机器上操作：

vim /etc/logstash/conf.d/logstash-to-kafka.sh
input {
  stdin {}
}


output {
  kafka {
    topic_id => "hello"
    bootstrap_servers => "10.0.10.21:9092"
    batch_size => 5
  }
  stdout {
    codec => rubydebug
  }
}
1
2
3
4
5
验证logstash语法
[root@localhost conf.d]# logstash -f /usr/share/logstash/config/conf.d/logstash-to-kafka.sh -t

Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties
Configuration OK

测试logstash向kafka写入数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
logstash服务器前台启动
[root@localhost conf.d]# logstash -f /usr/share/logstash/config/conf.d/logstash-to-kafka.sh

手动输入测试内容：
hello
{
      "@version" => "1",
          "host" => "localhost.localdomain",
    "@timestamp" => 2017-12-22T13:17:56.632Z,
       "message" => "hello"
}

logstash
{
      "@version" => "1",
          "host" => "localhost.localdomain",
    "@timestamp" => 2017-12-22T13:18:21.852Z,
       "message" => "logstash"
}

1
2
3
4
5
6
7
8
9
在21的kafka机器上运行如下消费命令：
kafka-console-consumer.sh --zookeeper 10.0.10.21:2181,10.0.10.22:2181,10.0.10.23,2181 --topic hello --from-beginning


结果如下：
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].

2017-12-22T13:17:56.632Z localhost.localdomain hello
2017-12-22T13:18:21.852Z localhost.localdomain logstash

– End –

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true


不论我写了什么...
其实都是错的......