大数据中间件——Kafka

这篇具有很好参考价值的文章主要介绍了大数据中间件——Kafka。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Kafka安装配置

首先我们把kafka的安装包上传到虚拟机中:

大数据中间件——Kafka,大数据,中间件,kafka

解压到对应的目录并修改对应的文件名:

大数据中间件——Kafka,大数据,中间件,kafka

首先我们来到kafka的config目录,我们第一个要修改的文件就是server.properties文件,修改内容如下:

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
# kafka在整个集群中的身份标识,集群中的id是唯一的
broker.id=0

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
# 存储kafka数据的位置,默认存储在临时文件夹,要修改成自己的文件夹
log.dirs=/opt/model/kafka/datas

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
# 连接的zookeeper集群,需要将集群中部署zookeeper的所有节点写入
# 首先,在zookeeper中,数据的存储是以目录树的方式去存储的,如果后期我们的kafka的数据要修改,在不做任何的修改的情况下,默认是存储在zookeeper根目录下的,这样我们想要单独提取出zookeeper的数据就非常的麻烦
# 所以我们将kafka的数据的单独存储在一个文件分支中,这就是我们为什么要在最后写一个[/kafka]的原因。
# 前面写多个节点是为了防止单个zookeeper节点无法连接可以使用其他的zookeeper节点
zookeeper.connect=node1:2181,node2:2181,node3:2181/kafka

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000


############################# Group Coordinator Settings #############################

# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0

主要修改三个部分,一个是唯一标识id,kafka的文件存储路径,一个是zookeeper的节点地址。

然后我们将kafka的安装包分发到其他的节点中。

注意在分发完成之后,不要忘记修改不同节点中的唯一标识id的值。

然后我们就可以启动kafka的服务了,注意在启动kafka的服务之前,我们必须要启动zookeeper的服务。

kafka和zookeeper一样,也是要在每个节点中都分别执行启动脚本,并且kafka的启动脚本需要手动指定配置文件:

./kafka-server-start.sh -daemon ../config/server.properties

注意,我的kafka的地址和你们的可能不一样,但是只需要知道启动命令在bin目录下,配置文件在conf目录下即可,我们在三台虚拟机上分别执行脚本:

大数据中间件——Kafka,大数据,中间件,kafka

当我们看到在集群中出现kafka的进程之后,就表示我们的kafka集群启动成功了。文章来源地址https://www.toymoban.com/news/detail-723036.html

到了这里,关于大数据中间件——Kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 中间件: Kafka安装部署

    下载二进制包 修改配置 启动 按照单机部署方式启动多个Zookeeper与broker节点。 修改config/server.properties配置: broker.id 每个节点唯一 zookeeper.connect: 改成zookeeper节点 查看集群状态:

    2024年02月12日
    浏览(44)
  • 消息中间件(二)——kafka

    在大数据中,会使用到大量的数据。面对这些海量的数据,我们一是需要做到能够 收集 这些数据,其次是要能够 分析和处理 这些海量数据。在此过程中,需要一套消息系统。 Kafka专门为分 布式高吞吐量 系统设计。作为一个消息代理的替代品,Kafka往往做的比其他消息中间

    2024年02月07日
    浏览(58)
  • 中间件(三)- Kafka(二)

    6.1 Kafka的高效读写 顺序写磁盘 Kafka的producer生产数据,需要写入到log文件中,写的过程是追加到文件末端,顺序写的方式,官网有数据表明,同样的磁盘,顺序写能够到600M/s,而随机写只有200K/s,这与磁盘的机械结构有关,顺序写之所以快,是因为其省去了大量磁头寻址的时

    2024年02月07日
    浏览(39)
  • 消息中间件之Kafka(一)

    高性能的消息中间件,在大数据的业务场景下性能比较好,kafka本身不维护消息位点,而是交由Consumer来维护,消息可以重复消费,并且内部使用了零拷贝技术,性能比较好 Broker持久化消息时采用了MMAP的技术,Consumer拉取消息时使用的sendfile技术 Kafka是最初由Linkedin公司开发,

    2024年01月20日
    浏览(52)
  • 消息中间件之Kafka(二)

    1.1 为什么要对topic下数据进行分区存储? 1.commit log文件会受到所在机器的文件系统大小的限制,分区之后可以将不同的分区放在不同的机器上, 相当于对数据做了分布式存储,理论上一个topic可以处理任意数量的数据 2.提高并行度 1.2 如何在多个partition中保证顺序消费? 方案一

    2024年01月21日
    浏览(49)
  • 中间件上云部署 kafka

    Apache Kafka是一种流行的分布式流式消息平台。Kafka生产者将数据写入分区主题,这些主题通过可配置的副本存储到broker群集上。 消费者来消费存储在broker的分区生成的数据。 storageclass ingress 创建topic 创建数据生产者,添加数据 在当前终端或另一个终端中创建数据消费者,消

    2024年02月16日
    浏览(39)
  • 【Java面试丨消息中间件】Kafka

    1. 介绍 使用kafka在消息的收发过程都有可能会出现消息丢失 (1)生产者发送消息到broker丢失 (2)消息在broker中存储丢失 (3)消费者从broker接收消息丢失 2. 生产者发送消息到broker丢失 设置异步发送:同步发送会发生阻塞,一般使用异步发送方式发送消息 消息重试:由于网

    2024年02月11日
    浏览(47)
  • 【Kafka】Ubuntu 部署kafka中间件,实现Django生产和消费

    原文作者 :我辈李想 版权声明 :文章原创,转载时请务必加上原文超链接、作者信息和本声明。 队列是一种特殊的线性表,特殊之处在于它只允许在表的前端(front)进行删除操作,而在表的后端(rear)进行插入操作,和栈一样,队列是一种操作受限制的线性表。进行插入

    2024年02月16日
    浏览(50)
  • 常用中间件redis,kafka及其测试方法

    一、中间件的使用场景 引入中间件的目的一般有两个: 1、提升性能 产品架构中的性能设计: 常用的中间件: 1) 高速缓存:redis 基于内存,所以比mysql块(存在磁盘io) 为什么查询速度快? 单进程+IO多路复用去提高性能 基于内存 做缓存,极大缓解了数据库压力 非常适合

    2024年04月11日
    浏览(54)
  • golang分布式中间件之kafka

    Kafka是一个分布式发布-订阅消息系统,由LinkedIn公司开发。它被设计为快速、可靠且具有高吞吐量的数据流平台,旨在处理大量的实时数据。Kafka的架构是基于发布-订阅模型构建的,可以支持多个生产者和消费者。 在本文中,我们将讨论如何使用Go语言来实现Kafka分布式中间件

    2024年02月07日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包