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
./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
10、 测试消费消息
Kafka也提供了一个消费消息的命令行工具
# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning This is a message This is another message
备注
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是旧版本命令
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
【资料下载】Python深度学习第二讲——数据增广
Python深度学习第二讲——数据增广 深度学习是目前的热门话题。而Python作为优美的解释型程序语言,拥有大量第三方库,正是连接深度学习的工具首选。从数据预处理到搭建神经网络,从模型训练到模型实现,python均能发挥其妙用。"授人以鱼不如授之以渔",这一课就带大家走进python深度学习的世界,从常用库的调用到代码的书写讲解python的数据增广方法。希望python能够成为大家研究和应用深度学习的亲密战友。 课程大纲:1、基于python的深度学习应用大观;2、基于python的深度学习数据增广方法;3、python数据增广所用库及代码解析; 讲师简介:猫饼君,香港城市大学理学硕士,曾在南洋理工大学参加过机器学习算法相关项目,现任职计算机视觉算法工程师,专注于计算机视觉算法研究及应用落地。 欢迎扫码进群观看直播及往期回顾 视频链接:
- 下一篇
Java堆内存又溢出了!教你一招必杀技
JAVA堆内存管理是影响性能主要因素之一。堆内存溢出是JAVA项目非常常见的故障,在解决该问题之前,必须先了解下JAVA堆内存是怎么工作的。 先看下JAVA堆内存是如何划分的,如图: JVM内存划分为堆内存和非堆内存,堆内存分为年轻代(Young Generation)、老年代(Old Generation),非堆内存就一个永久代(Permanent Generation)。 年轻代又分为Eden和Survivor区。Survivor区由FromSpace和ToSpace组成。Eden区占大容量,Survivor两个区占小容量,默认比例是8:1:1。 堆内存用途:存放的是对象,垃圾收集器就是收集这些对象,然后根据GC算法回收。 非堆内存用途:永久代,也称为方法区,存储程序运行时长期存活的对象,比如类的元数据、方法、常量、属性等。 在JDK1.8版本废弃了永久代,替代的是元空间(MetaSpace),元空间与永久代上类似,都是方法区的实现,他们最大区别是:元空间并不在JVM中,而是使用本地内存。元空间有注意有两个参数: MetaspaceSize :初始化元空间大小,控制发生GC阈值 M...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS8安装Docker,最新的服务器搭配容器使用
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS8编译安装MySQL8.0.19
- CentOS8,CentOS7,CentOS6编译安装Redis5.0.7
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- SpringBoot2整合Redis,开启缓存,提高访问速度
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- Hadoop3单机部署,实现最简伪集群
- CentOS7,CentOS8安装Elasticsearch6.8.6
- CentOS6,7,8上安装Nginx,支持https2.0的开启