2019年8月18日

上文主要讲解kafka的文件架构，这节我们参考了《zookeeper部署安装》，完成之后，创建我们的kafka demo，探探消息队列的真相....

kafka测试demo之控制台自发自收数据

kafka配置文件

配置文件都放在/config下面了。

producer.properties


############################# Producer Basics ############################## list of brokers used for bootstrapping knowledge about the rest of the cluster
# format: host1:port1,host2:port2 ...
bootstrap.servers=localhost:9092# specify the compression codec for all data generated: none, gzip, snappy, lz4, zstd
compression.type=none# name of the partitioner class for partitioning events; default partition spreads data randomly
#partitioner.class=# the maximum amount of time the client will wait for the response of a request
#request.timeout.ms=# how long `KafkaProducer.send` and `KafkaProducer.partitionsFor` will block for
#max.block.ms=# the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together
#linger.ms=# the maximum size of a request in bytes
#max.request.size=# the default batch size in bytes when batching multiple records sent to a partition
#batch.size=# the total bytes of memory the producer can use to buffer records waiting to be sent to the server
#buffer.memory=

consumer.properties

# list of brokers used for bootstrapping knowledge about the rest of the cluster
# format: host1:port1,host2:port2 ...
bootstrap.servers=localhost:9092# consumer group id
group.id=test-consumer-group# What to do when there is no initial offset in Kafka or if the current
# offset does not exist any more on the server: latest, earliest, none
#auto.offset.reset=

zookeeper.properties

# the directory where the snapshot is stored.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=0

server.properties


# 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.
##### 每一个broker在集群中的唯一表示，要求是正数。当该服务器的IP地址发生改变时，broker.id没有变化，则不会影响consumers的消息情况/默认值=0 #########
broker.id=1# Switch to enable topic deletion or not,default value is false;是否能删除topic
delete.topic.enable=true############################# 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://localhost: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
########## broker处理消息的最大线程数，一般情况下数量为cpu核数 #####
num.network.threads=4# The number of threads that the server uses for processing requests, which may include disk I/O
######### broker处理磁盘IO的线程数，数值为cpu核数2倍 #####
num.io.threads=8# The send buffer (SO_SNDBUF) used by the socket server
######## socket的发送缓冲区，socket的调优参数SO_SNDBUFF #####
socket.send.buffer.bytes=102400# The receive buffer (SO_RCVBUF) used by the socket server
####### socket的接受缓冲区，socket的调优参数SO_RCVBUFF #####
socket.receive.buffer.bytes=102400# The maximum size of a request that the socket server will accept (protection against OOM)
####### socket请求的最大数值，防止serverOOM，message.max.bytes必然要小于socket.request.max.bytes，会被topic创建时的指定参数覆盖 ######
socket.request.max.bytes=104857600############################# Log Basics ############################## A comma separated list of directories under which to store log files
############ kafka数据的存放地址，多个地址的话用逗号分割,多个目录分布在不同磁盘上可以提高读写性能  /data/kafka-logs-1，/data/kafka-logs-2 ###########
log.dirs=/data/kafka-logs# 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=2# 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 for 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.cleanup.policy设置数据清除策略,log.retention.bytes和log.retention.minutes或log.retention.hours任意一个达到要求，都会执行删除 ######
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.
######## topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，会被topic创建时的指定参数覆盖 #####
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.cleanup.policy中设置的策略 #####
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集群的地址，可以是多个，多个之间用逗号分割 hostname1:port1,hostname2:port2,hostname3:port3  ######
zookeeper.connect=192.168.0.100:2181# Timeout in ms for connecting to zookeeper
######## 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

启动zookeeper后，接着启动kafka-server。

创建topic

sudo ./kafka-topics.sh --create --zookeeper 192.168.0.101:2181 --partitions 1 --replication-factor 1 --topic messageLogger

--create：说明是创建topic，--zookeeper：连接哪个zk，--partitions：创建分区，【分区机制】，定义分区数--replication-factor：创建副本，【副本机制】，定义副本数--topic：topic名字

查看所有topic

sudo ./kafka-topics.sh --list --zookeeper 192.168.0.101:2181

--list：罗列--zookeeper：哪个zk的相关信息

去kafka的数据log文件夹看看 (因为配置文件中：log.dirs=/data/kafka-logs)

蓝色文件夹是分区，副本在其他服务器(演示只有一个分区和一个副本)。当然还有其他的一些文件。

往topic写数据

sudo ./kafka-console-producer.sh --broker-list 192.168.0.101:9092 --topic messageLogger

从topic读数据

sudo ./kafka-console-consumer.sh --bootstrap-server 192.168.0.101:2181 --from-beginning --topic messageLogger

查看topic详情

sudo ./kafka-topics.sh --zookeeper 192.168.0.101:2181 --describe --topic messageLogger

Kafka 实战（3）：kafka安装部署·2相关推荐

Kafka实战(3)-Kafka的自我定位
Kafka,具有高水平扩展和高吞吐量的分布式消息系统. Kafka对消息保存时根据Topic进行归类,发送消息者成为Producer,消息接受者成为Consumer,此外kafka集群有多个kafka ...
Kafka实战 - 02 Kafka生产者发送消息至topic实现数据上报
文章目录 1. 项目背景 2. 依赖和配置 3. 生产者配置 KafkaConfiguration 4. 同步数据Topic枚举 SyncDataTopicEnum 5. 请求体 DataSyncQo ...
Docker实战：Docker安装部署RabbitMQ
使用Docker安装部署RabbitMQ 1.docker search rabbitmq:management 2.docker pull rabbitmq:management 注意:如果dock ...
多易教育KAFKA实战(1)-KAFKA集群安装和shell客户端
注意kafka的安装需要依赖Zookeeper集群 ,所以安装kafka之前先安装zookeeper! zookeeper安装上传安装包解压 tar -zxvf zookeeper-3.4.6.t ...
Kafka 实战指南——Kafka 消费者配置
文章目录 1. 消费位点提交 2. 消费位点重置 3. session 超时和心跳监测 4. 拉取大消息 5. 拉取公网 6. 消息重复和消费幂等 7. 消费失败 8. 消费延迟 9. 消费阻塞以及堆 ...
Kafka实战 - 06 Kafka消费者：从指定Topic拉取工单处置记录信息并存入MongoDB数据库
文章目录 1. 处置记录表 t_disposal_record 2. kafka 主题和消费者配置 3. 定义一个线程任务 KafkaTask 1. kafka Topic中的数据:KafkaDisp ...
mysql事件探查器_【干货】Mysql的事件探查器-之Mysql-Proxy代理实战一（安装部署与实战sql拦截与性能监控）...
1:资料参考 https://blog.csdn.net/coldljy/article/details/3168906 https://www.cnblogs.com/jwentest/p/8552 ...
aerospike mysql_[置顶] Aerospike数据库实战(四) -- Aerospike 安装部署
参考下面配置,修改Aerospike Server端conf文件. service { user root group root paxos-single-replica-limit 1 # Numb ...
mysql数据实时同步：Canal安装部署、kafka安装、zk安装、mysql安装、Canal Server+Canal Client HA，Canal+mysql+Kafka，相关验证（学习笔记）
目录 Canal安装部署 1.1. 服务器准备 1.2. 设置主机名并配置hosts 1.3. 免密设置 1.4. 设置ntp时间 1.5. 关闭防火墙 1.6. 关闭selinux 1.7. 安装J ...
kafka不使用自带zk_kafka 安装部署教程
kafka 安装部署教程 1. 下载官网链接 http://kafka.apache.org/downloads 浙大源 http://mirrors.tuna.tsinghua.edu.cn/ap ...

Kafka 实战（3）：kafka安装部署·2

kafka测试demo之控制台自发自收数据

kafka配置文件

producer.properties

consumer.properties

zookeeper.properties

server.properties

创建topic

查看所有topic

往topic写数据

从topic读数据

查看topic详情

Kafka 实战（3）：kafka安装部署·2相关推荐

最新文章

热门文章

Kafka 实战 （3）：kafka安装部署·2

kafka测试demo之控制台自发自收数据

kafka配置文件

producer.properties

consumer.properties

zookeeper.properties

server.properties

创建topic

查看所有topic

往topic写数据

从topic读数据

查看topic详情

Kafka 实战 （3）：kafka安装部署·2相关推荐

最新文章

热门文章

Kafka 实战（3）：kafka安装部署·2

Kafka 实战（3）：kafka安装部署·2相关推荐