Kafka初入门简单配置与使用

Stella981
• 阅读 641

一 Kafka概述

1.1 Kafka是什么

在流式计算中,Kafka一般用来缓存数据,Storm通过消费Kafka的数据进行计算。

1)Apache Kafka是一个开源消息系统,由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。

2)Kafka最初是由LinkedIn开发,并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。

3)Kafka****是一个分布式消息队列。Kafka对消息保存时根据Topic进行归类,发送消息者称为Producer,消息接受者称为Consumer,此外kafka集群有多个kafka实例组成,每个实例(server)称为broker。

4)无论是kafka集群,还是producer和consumer都依赖于zookeeper集群保存一些meta信息,来保证系统可用性。

1.2 Kafka内部实现原理

(1)点对点模式(一对一,消费者主动拉取数据,消息收到后消息清除)

点对点模型通常是一个基于拉取或者轮询的消息传送模型,这种模型从队列中请求信息,而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理,即使有多个消息监听者也是如此。

(2)发布/订阅模式(一对多,数据生产后,推送给所有订阅者)

发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者,临时订阅者只在主动监听主题时才接收消息,而持久订阅者则监听主题的所有消息,即使当前订阅者不可用,处于离线状态。

1.2 为什么需要消息队列

1)解耦:

  允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。

2)冗余:

消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕。

3)扩展性:

因为消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要

另外增加处理过程即可。

4)灵活性 & 峰值处理能力:

在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷的请求而完全崩溃。

5)可恢复性:

系统的一部分组件失效时,不会影响到整个系统。消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。

6)顺序保证:

在大多使用场景下,数据处理的顺序都很重要。大部分消息队列本来就是排序的,并且能保证数据会按照特定的顺序来处理。(Kafka保证一个Partition内的消息的有序性)

7)缓冲:

有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。

8)异步通信:

很多时候,用户不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。

一 Kafka集群部署

2.1 环境准备

2.1.1 集群规划

    hadoop201                                  hadoop202                           hadoop203

    zk                                              zk                                       zk

    kafka                                         kafka                                  kafka

2.1.2 jar包下载

    官网:  http://kafka.apache.org/downloads.html

2.1.3 虚拟机准备

    1)准备3台虚拟机

    2)配置ip地址

    3)配置主机名称

    4)3台主机分别关闭防火墙

2.1.4 安装jdk

2.1.5 安装Zookeeper

2.2 Kafka集群部署

  1)解压安装包

tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt/module/

  2)修改解压后的文件名称

cd /opt/module/

  3)在/opt/module/kafka目录下创建logs文件夹

cd kafka_2.11-0.11.0.0/mkdir logs

Kafka初入门简单配置与使用

  4)修改配置文件

cd config/

    4.1修改server.properties(vi server.properties)

Kafka初入门简单配置与使用

Kafka初入门简单配置与使用

Kafka初入门简单配置与使用

(文件解释

#broker的全局唯一编号,不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径
log.dirs=/opt/module/kafka_2.11-0.11.0.0/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间,超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop201:2181,hadoop202:2181,hadoop203:2181

  5)配置环境变量

su rootvi /etc/profile

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka_2.11-0.11.0.0/
export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

Kafka初入门简单配置与使用

  6)分发安装包

cd /etc/
scp profile hadoop202:/etc/
scp profile hadoop203:/etc/
su jokerq
scp -r kafka_2.11-0.11.0.0/ hadoop202:/opt/module/
scp -r kafka_2.11-0.11.0.0/ hadoop203:/opt/module/

  7)分别在hadoop202和hadoop203上修改配置文件/opt/module/kafka/config/server.properties中

    的broker.id=2、broker.id=3

        注:broker.id不得重复

Kafka初入门简单配置与使用

  7.1)在1 2 3中(source /etc/profile)

  8)启动集群

    依次在hadoop201、hadoop202、hadoop203节点上启动kafka

cd /opt/module/kafka_2.11-0.11.0.0/bin/kafka-server-start.sh config/server.properties &(完事按下回车即可回到命令行)

2.3 Kafka命令行操作

1)查看当前服务器中的所有topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --list --zookeeper hadoop102:2181

2)创建topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --replication-factor 3 --partitions 1 --topic first

选项说明:

--topic 定义topic名

--replication-factor  定义副本数

--partitions  定义分区数

3)  删除topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --delete --zookeeper hadoop102:2181 --topic first

需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。

4)发送消息

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first

>hello world

>atguigu  atguigu

5)消费消息

[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --from-beginning --topic first

6)查看某个Topic的详情

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --topic first --describe --zookeeper hadoop102:2181

ctrl z   bg--后台
     fg--前台  (干掉进程:然后按ctrl+c 干掉进程)
点赞
收藏
评论区
推荐文章
blmius blmius
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Jacquelyn38 Jacquelyn38
2年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Stella981 Stella981
2年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Easter79 Easter79
2年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Wesley13 Wesley13
2年前
Kafka是什么,JMS是什么,常见的类JMS消息服务器,为什么需要消息队列(一)
1、Kafka是什么ApacheKafka是一个开源消息系统,由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。 Kafka最初是由LinkedIn开发,并于2011年初开源。2012年10月从ApacheIncubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
Wesley13 Wesley13
2年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
2年前
Kafka笔记
第1章Kafka简介1.1kafka起源Kafka是由LinkedIn开发并开源的分布式消息系统,2012年捐赠给Apache基金会,采用Scala语言,运行在JVM中,最新版本1.0.0。1.2kafka设计目标Kafka是一种分布式的,基于发布/订阅的消息系统。主要设计目标如下:①以时间复杂度O(1)的方式提供消息持久化能力,即
Wesley13 Wesley13
2年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
4个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这