Centos-Kafka 消息队列

常用的消息队列有Kafka、RabbitMQ、ActiveMQ、metaq等

kafka有什么?

producer 消息的生成者,即发布消息
consumer 消息的消费者,即订阅消息
broker Kafka以集群的方式运行,可以由一个或多个服务组成,服务即broker
zookeeper 协调转发

1、下载kafka

cd /usr/nacp/
wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka_2.12-1.0.0.tgz

由于kafka_2.12-1.0.0.tgz是可执行程序,里面集成了zookeeper
tar -xzvf kafka_2.12-1.0.0.tgz #解压

2、运行与测试,以下都是终端命令

cd kafka_2.12-1.0.0

启动Zookeeper,如果无报错则说明启动成功。daemon以及nohup &是实现在后台启动,即守护进程。

方法1 ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
方法2 nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties&

启动Kafka,如果无报错则说明启动成功。daemon以及nohup &是实现在后台启动,即守护进程。

方法1 ./bin/kafka-server-start.sh -daemon config/server.properties
方法2 nohup ./bin/kafka-server-start.sh config/server.properties&

3、查看进程是否正常运行,通过检测2181与9092端口,2181是zookeeper,9092是kafka

netstat -tunlp|egrep "(2181|9092)"

4、创建主题,例如test

./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

Centos-Kafka 消息队列

./bin/kafka-topics.sh --create --zookeeper 192.168.1.62:2181,192.168.1.62:2182,192.168.1.62:2183 --replication-factor 3 --partitions 3 --topic emqtest

--zookeeper : zookeeper集群列表,用英文逗号分隔。可以不用指定zookeeper整个集群内的节点列表,只指定某个或某几个zookeeper节点列表也是可以的
replication-factor : 复制数目,提供failover机制;1代表只在一个broker上有数据记录,一般值都大于1,代表一份数据会自动同步到其他的多个broker,防止某个broker宕机后数据丢失。
partitions : 一个topic可以被切分成多个partitions,一个消费者可以消费多个partitions,但一个partitions只能被一个消费者消费,所以增加partitions可以增加消费者的吞吐量。kafka只保证一个partitions内的消息是有序的,多个一个partitions之间的数据是无序的。

注意,
(1)使用集群时,必须手动先创建topic,指定zookeeper节点列表;单机也应该要创建。否则客户端会报错。
(2)主题名不建议使用"."和"_"字符。
(3)建议创建主题时,指定分区数为kafka broker节点数目的整数倍,例如--partitions 3。创建完主题,可以在路径log.dirs看到主题和分区的文件夹。详情参见《Kafka入门与实践.牟大恩》3.6.1节描述

5、查看主题

./bin/kafka-topics.sh --describe --zookeeper  localhost:2181 --topic test
./bin/kafka-topics.sh --list --zookeeper localhost:2181

例如:broker集群,有3个节点。同时给主题mynewt分配了3个分区。每个分区都有leader和follow。

6、开启一个终端,发送消息,生产者的消息要发往kafka

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

7、另起一个终端,消费消息,消费者的消息来自zookeeper(协调转发)

./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

将来新版本需要写成./bin/kafka-console-consumer.sh --bootstrap-server localhost:2181 --topic test --from-beginning

8、结束进程

./bin/kafka-server-stop.sh
./bin/zookeeper-server-stop.sh

kafka启动时先启动zookeeper,再启动kafka;关闭时相反,先关闭kafka,再关闭zookeeper。

9、 测试发送消息

Kafka提供了一个命令行工具,可以从输入文件或者命令行中读取消息并发送给Kafka集群,每一行是一条消息。运行producer,然后在控制台输入几条消息到服务器

# bin/kafka-console-producer.sh --broker-list 192.168.1.62:9092 --topic test 
This is a message
This is another message

Centos-Kafka 消息队列

10、 测试消费消息
Kafka也提供了一个消费消息的命令行工具

# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
This is a message
This is another message

Centos-Kafka 消息队列

备注

1、配置文件与消息持久化路径
/config/server.properties

[html] view plain copy
# A comma seperated list of directories under which to store log     files  
log.dirs=/tmp/kafka-logs  
例如客户端发送test为主题的消息,会持久化在这个文件:/tmp/kafka-logs/test-0/00000000000000000000.log

/config/zookeeper.properties

[html] view plain copy
# the directory where the snapshot is stored.  
dataDir=/tmp/zookeeper  
# the port at which the clients will connect  
clientPort=2181  
建议修改kafka的日志目录和zookeeper数据目录,因为这两项默认放在tmp目录,而tmp目录中内容会随重启而丢失:

server.properties:

log.dirs=/tmp/kafka-logs
修改为
log.dirs=/usr/local/kafka/logs

zookeeper.properties

dataDir=/tmp/zookeeper
修改为
dataDir=/usr/local/zookeeper/data

2、配置文件bootstrap.servers 和 broker.list区别:
本以为是两个参数,其实是实现一个功能,查看源代码后发现broker.list是旧版本命令

上一篇:Centos-mongodb主从复制与副本集


下一篇:国家发改委公示了38个大数据发展重大工程拟支持项目