文档章节

高吞吐量的分布式发布订阅消息系统Kafka--安装及测试

sunney888
 sunney888
发布于 2016/07/05 16:27
字数 1007
阅读 24
收藏 0

一、Kafka概述 

    Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop的一样的日志数据和离线分析系统,但又要求实时处理的限制,这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群机来提供实时的消费。

二、Kafka相关术语

  • Broker
    Kafka集群包含一个或多个服务器,这种服务器被称为broker
  • Topic
    每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic。(物理上不同Topic的消息分开存储,逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处)
  • Partition
    Partition是物理上的概念,每个Topic包含一个或多个Partition.
  • Producer
    负责发布消息到Kafka broker
  • Consumer
    消息消费者,向Kafka broker读取消息的客户端。
  • Consumer Group
    每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)。

二、Kafka下载及安装

     1、下载

  2、安装

tar zxvf kafka_2.11-0.9.0.1.tgz
cd kafka_2.11-0.9.0.1

  3、集群配置

         设定有两台服务器192.168.1.237、192.168.1.238,两台服务器各安装有两zookeeper,端口都为2181(zookeeper不再说明),每个服务器都为Kafka配置3个broker。

     3.1、server.properties配置

broker.id = 10
port = 9090
host.name=192.168.1.237
advertised.host.name=192.168.1.237
log.dirs=/tmp/kafka-logs/server0
zookeeper.connect=192.168.1.237:2181,192.168.1.238:2181

  说明:host.name\advertised.host.name两个参数还是要配置为IP,否则会有各种各样的问题。

    3.2、server1.properties配置

cp config/servier.properties config/server1.properties
vim config/server1.properties
broker.id = 11
port = 9091
host.name=192.168.1.237
advertised.host.name=192.168.1.237
log.dirs=/tmp/kafka-logs/server1
zookeeper.connect=192.168.1.237:2181,192.168.1.238:2181

   3.3、server2.properties配置

cp config/servier.properties config/server2.properties
vim config/server2.properties
broker.id = 12
port = 9092
host.name=192.168.1.237
advertised.host.name=192.168.1.237
log.dirs=/tmp/kafka-logs/server2
zookeeper.connect=192.168.1.237:2181,192.168.1.238:2181

  说明:同一台服务器port、log.dirs不能相同,不同的服务器broker.id只要在一个集群中都不能相同。

      3.4、同理 另一台服务器的server.properties,server1.properties,server2.properties的broker.id分别为:20、21、22,port分别为:9090、9091、9092  其它:host.name=192.168.1.238、advertised.host.name=192.168.1.238

      3.5、启动

bin/kafka-server-start.sh config/server.properties &
bin/kafka-server-start.sh config/server1.properties &
bin/kafka-server-start.sh config/server2.properties &

  3.6、监控端口

netstat -tunpl |grep 2181
netstat -tunpl |grep 9090
netstat -tunpl |grep 9091
netstat -tunpl |grep 9092

  看一下这4个端口起来没有,并看一下iptables有没有加入这4个IP的启动,或要把iptables相关,否则JAVA连接不进来。

    四、测试

        4.1、创建Topic

bin/kafka-topics.sh --create --zookeeper 192.168.1.237:2181 --replication-factor 3 --partitions 1 --topic testTopic

    4.2、查看创建情况

bin/kafka-topics.sh --describe --zookeeper 192.168.1.237:2181 --topic testTopic

    4.3、生产者发送消息

bin/kafka-console-producer.sh --broker-list 192.168.1.237:9090 --topic testTopic

        4.4、消费都接收消息

bin/kafka-console-consumer.sh --zookeeper 192.168.1.237:2181 --from-beginning --topic testTopic

    4.5、检查consumer offset位置

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect 192.168.1.237:2181 --group testTopic

  五、遇到的问题

          1、运行一段时间报错

#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (malloc) failed to allocate 986513408 bytes for committing reserved memory.
# An error report file with more information is saved as:
# //hs_err_pid6500.log
OpenJDK 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000bad30000, 986513408, 0) failed; error='Cannot allocate memory' (errno=12)

      解决:

    you can adjust the JVM heap size by editing kafka-server-start.sh, zookeeper-server-start.shand so on:

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

    The -Xms parameter specifies the minimum heap size. To get your server to at least start up, try changing it to use less memory. Given that you only             have 512M, you should change the maximum heap size (-Xmx) too:

export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M"

        I'm not sure what the minimal memory requirements of kafka in default config are - maybe you need to adjust the message size in kafka to get it to run.

 

  

 

本文转载自:http://www.cnblogs.com/yuanermen/p/5452478.html

sunney888

sunney888

粉丝 11
博文 15
码字总数 1499
作品 0
深圳
部门经理
私信 提问
Spring Boot 中使用 kafka

Kafka 是一种高吞吐的分布式发布订阅消息系统,能够替代传统的消息队列用于解耦合数据处理,缓存未处理消息等,同时具有更高的吞吐率,支持分区、多副本、冗余,因此被广泛用于大规模消息数据...

PengLei
2017/10/17
0
0
SpringBoot开发案例之整合Kafka实现消息队列

前言 最近在做一款秒杀的案例,涉及到了同步锁、数据库锁、分布式锁、进程内队列以及分布式消息队列,这里对SpringBoot集成Kafka实现消息队列做一个简单的记录。Kafka简介 Kafka是由Apache软...

小柒2012
2018/05/18
0
0
Kafka、RabbitMQ、RocketMQ等消息中间件的介绍和对比

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/t4i2b10X4c22nF6A/article/details/89507748

JAVA高级架构v
04/24
0
0
Kafka、RabbitMQ、RocketMQ消息中间件的对比—— 消息发送性能

中间件小哥 浏览 228 2016-04-08 11:21:09 摘要引言 分布式系统中,我们广泛运用消息中间件进行系统间的数据交换,便于异步解耦。现在开源的消息中间件有很多,前段时间我们自家的产品 RocketMQ...

tantexian
2016/05/04
512
0
Kafka~Linux环境下的部署

概念 Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键...

mcy247
2017/12/06
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Dubbo-自适应拓展机制

背景 在 Dubbo 中,很多拓展都是通过 SPI 机制进行加载的,比如 Protocol、Cluster、LoadBalance 等,这些都是Dubbo的基础组件。这些基础组件的拓展不是在系统框架启动阶段被加载,而是拓展方...

rock-man
14分钟前
4
0
Kali安装fcitx输入法(五笔)

安装fcitx > sudo apt-get install fcitx-rime fcitx-config-gtk3 重启 > sudo reboot fcitx配置 效果就是这样 配置输入法切换 系统设置...

yeahlife
16分钟前
3
0
IE之css3效果兼容

本文转载于:专业的前端网站▷IE之css3效果兼容 一、兼容css阴影效果(ie滤镜) 1.Shadow,阴影 .shadow { -moz-box-shadow: 3px 3px 4px #000; -webkit-box-shadow: 3px 3px 4px #000; box-sha...

前端老手
19分钟前
3
0
NiushopB2C开源商城功能列表说明:

B2C单商户免费版:PC商城+微商城 B2C单商户标准版:PC商城+微商城组合套餐+阶梯优惠核销功能 B2C单商户企业版:PC商城+微商城拼团+组合套餐阶梯优惠+核销功能 B2C单商户分销版:PC商城+微商城...

niushop-芳
21分钟前
3
0
图片如何转GIF图片呢

如何将生活中拍摄的好玩有趣的图片制作成GIF动图呢?相信很多小伙伴都不知道要如何制作,其实制作方法非常的简单,下面分享一个图片转GIF动图的方法,希望这个方法能够帮助大家在与好友斗图时...

白米稀饭2019
28分钟前
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部