kafka 修改topic的partition数量
1 | //修改前检查 |
kafka use
kafka 常用命令
1 | //查看topic |
1 | //查看当前消费者是否已经读到最新的数据 |
kafka info
kafka 关键词
1 | Kafka部分名词解释如下: |
kafka副本分配算法
1 | 副本分配算法如下:（mod表示余数,例如: 3 mod 2 = 1） |
kafka存储过程
1 | Kafka运行时很少有大量读磁盘的操作,主要是定期批量写磁盘操作,因此操作磁盘很高效.这跟Kafka文件存储中读写message的设计是息息相关的.Kafka中读写message有如下特点: |
kafka Partition Recovery机制
1 | 每个Partition会在磁盘记录一个RecoveryPoint, 记录已经flush到磁盘的最大offset。当broker fail 重启时,会进行loadLogs。 首先会读取该Partition的RecoveryPoint,找到包含RecoveryPoint的segment及以后的segment, 这些segment就是可能没有 完全flush到磁盘segments。然后调用segment的recover,重新读取各个segment的msg,并重建索引 |
kafka Partition Replica同步机制
1 | 1.Partition的多个replica中一个为Leader,其余为follower |
数据可靠性保证
1 | 当Producer向Leader发送数据时,可以通过acks参数设置数据可靠性的级别 |
数据一致性保证
1 | 一致性定义:若某条消息对Consumer可见,那么即使Leader宕机了,在新Leader上数据依然可以被读到 |
kafka optimize
主要优化原理和思路
磁盘优化
1
2
3
4
5
6
7
8
9
10
11
12
13kafka是一个高吞吐量分布式消息系统,并且提供了持久化.
其高性能的有两个重要特点:
* 利用了磁盘连续读写性能远远高于随机读写的特点
* 并发,将一个topic拆分多个partition
要充分发挥kafka的性能,就需要满足这两个条件
kafka读写的单位是partition.因此,将一个topic拆分为多个partition可以提高吞吐量.但是,这里有个前提,就是不同partition需要位于不同的磁盘（可以在同一个机器).如果多个partition位于同一个磁盘,那么意味着有多个进程同时对一个磁盘的多个文件进行读写,使得操作系统会对磁盘读写进行频繁调度,也就是破坏了磁盘读写的连续性.
具体配置上,是将不同磁盘的多个目录配置到broker的log.dirs,例如:
log.dirs=/disk1/kafka-logs,/disk2/kafka-logs,/disk3/kafka-logs
kafka会在新建partition的时候,将新partition分布在partition最少的目录上,因此,一般不能将同一个磁盘的多个目录设置到log.dirs.
同一个ConsumerGroup内的Consumer和Partition在同一时间内必须保证是一对一的消费关系.任意Partition在某一个时刻只能被一个Consumer Group内的一个Consumer消费(反过来一个Consumer则可以同时消费多个Partition)JVM参数配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15推荐使用最新的G1来代替CMS作为垃圾回收器,推荐JDK最低版本为1.7u51
-Xms30g -Xmx30g -XX:PermSize=48m -XX:MaxPermSize=48m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35
G1相比较于CMS的优势:
* G1是一种适用于服务器端的垃圾回收器,很好的平衡了吞吐量和响应能力
* 对于内存的划分方法不同,Eden,Survivor,Old区域不再固定,使用内存会更高效.G1通过对内存进行Region的划分,有效避免了内存碎片问题
* G1可以指定GC时可用于暂停线程的时间(不保证严格遵守)而CMS并不提供可控选项
* CMS只有在FullGC之后会重新合并压缩内存,而G1把回收和合并集合在一起
* CMS只能使用在Old区,在清理Young时一般是配合使用ParNew,而G1可以统一两类分区的回收算法
G1的适用场景：
* JVM占用内存较大(At least 4G)
* 应用本身频繁申请,释放内存,进而产生大量内存碎片时。
* 对于GC时间较为敏感的应用Broker参数配置
1
2
3
4
5
6
7
8
9
10
11//网络和io操作线程配置优化
# broker处理消息的最大线程数
num.network.threads=xxx
# broker处理磁盘IO的线程数
num.io.threads=xxx
说明:
num.network.threads用于接收并处理网络请求的线程数,默认为3.其内部实现是采用Selector模型.启动一个线程作为Acceptor来负责建立连接,再配合启动num.network.threads个线程来轮流负责从Sockets里读取请求,一般无需改动,除非上下游并发请求量过大.一般num.network.threads主要处理网络io,读写缓冲区数据,基本没有io等待,配置线程数量为cpu核数加1.
num.io.threads主要进行磁盘io操作,高峰期可能有些io等待,因此配置需要大些.配置线程数量为cpu核数2倍,最大不超过3倍.1
2
3
4
5//log数据文件刷盘策略
为了大幅度提高producer写入吞吐量,需要定期批量写文件:
# 每当producer写入10000条消息时,刷数据到磁盘 log.flush.interval.messages=10000
# 每间隔1秒钟时间,刷数据到磁盘
log.flush.interval.ms=10001
2
3
4
5
6
7
8//日志保留策略配置
当kafka server被写入海量消息后,会生成很多数据文件,且占用大量磁盘空间,如果不及时清理,可能磁盘空间不够用,kafka默认是保留7天.
建议配置：
# 保留三天，也可以更短
log.retention.hours=72
# 段文件配置1GB,有利于快速回收磁盘空间,重启kafka加载也会加快(如果文件过小,则文件数量比较多,kafka启动时是单线程扫描目录(log.dir)下所有数据文件)
log.segment.bytes=10737418241
2
3
4
5
6
7Kafka官方并不建议通过Broker端的log.flush.interval.messages和log.flush.interval.ms来强制写盘,认为数据的可靠性应该通过Replica来保证,而强制Flush数据到磁盘会对整体性能产生影响.
可以通过调整/proc/sys/vm/dirty_background_ratio和/proc/sys/vm/dirty_ratio来调优性能:
* 脏页率超过第一个指标会启动pdflush开始Flush Dirty PageCache
* 脏页率超过第二个指标会阻塞所有的写操作来进行Flush
* 根据不同的业务需求可以适当的降低dirty_background_ratio和提高dirty_ratio
* 如果topic的数据量较小可以考虑减少log.flush.interval.ms和log.flush.interval.messages来强制刷写数据,减少可能由于缓存数据未写盘带来的不一致配置jmx
1
2
3[penn@root kafka_2.10-0.8.1]$ vim bin/kafka-run-class.sh
#最前面添加一行
JMX_PORT=8060Replica相关配置
1
2
3
4
5
6
7
8
9
10replica.lag.time.max.ms:10000
replica.lag.max.messages:4000
num.replica.fetchers:1
#在Replica上会启动若干Fetch线程把对应的数据同步到本地,而num.replica.fetchers这个参数是用来控制Fetch线程的数量.
#每个Partition启动的多个Fetcher,通过共享offset既保证了同一时间内Consumer和Partition之间的一对一关系,又允许我们通过增多Fetch线程来提高效率
default.replication.factor:1
#这个参数指新创建一个topic时,默认的Replica数量
#Replica过少会影响数据的可用性,太多则会白白浪费存储资源,一般建议在2~3为宜purgatory
1
2fetch.purgatory.purge.interval.requests:1000
producer.purgatory.purge.interval.requests:1000
http://blog.csdn.net/vegetable_bird_001/article/details/51858915
kafka install
java install
参考java 环境安装
zookeeper install
参考zookeeper install
kafka standard install
kafka安装
1
2
3
4
5
6
7[root@10 ~]# cd /mnt/ops/app/
[root@10 app]# tar xzf kafka_2.11-0.9.0.0.tgz
[root@10 app]# mv kafka_2.11-0.9.0.0 /mnt/app/kafka
[root@10 app]# chown -R wisdom.wisdom /mnt/app/kafka
[root@10 app]# mkdir -p /mnt/{data,log}/kafka
[root@10 app]# chown -R wisdom.wisdom /mnt/{data,log}/kafkakafka配置文件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27[root@10 app]# cp /mnt/app/kafka/config/{server.properties,server.properties.bak}
[root@10 app]# cat > /mnt/app/kafka/config/server.properties <<EOF
> broker.id=113
> advertised.host.name=10.0.2.113
> advertised.port=9092
> delete.topic.enable=true
> listeners=PLAINTEXT://:9092
> num.network.threads=9
> num.io.threads=16
> socket.send.buffer.bytes=102400
> socket.receive.buffer.bytes=102400
> socket.request.max.bytes=104857600
> log.dirs=/mnt/data/kafka
> num.partitions=3
> num.recovery.threads.per.data.dir=2
> default.replication.factor = 1
> replica.fetch.max.bytes=20000000
> num.replica.fetchers=2
> message.max.bytes=10000000
> log.flush.interval.messages=10000
> log.flush.interval.ms=1000
> log.retention.hours=48
> log.segment.bytes=1073741824
> log.retention.check.interval.ms=300000
> zookeeper.connect=10.0.2.113:2181
> zookeeper.connection.timeout.ms=6000
> EOFkafka log存放位置
1
2
3
4
5
6
7
8
9[root@10 app]# vim /mnt/app/kafka/bin/kafka-run-class.sh
# Log directory to use
if [ "x$LOG_DIR" = "x" ]; then
LOG_DIR="/mnt/log/kafka"
JMX_PORT=8092
else
LOG_DIR="/mnt/log/kafka"
JMX_PORT=8092
fikafka jvm修改
1
2
3
4[root@10 app]# vim /mnt/app/kafka/bin/kafka-server-start.sh
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-Xmx4G -Xms4G -Xmn2G -XX:PermSize=64m -XX:MaxPermSize=128m -XX:SurvivorRatio=6 -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly"
fikafka 启动
1
2[root@10 app]# su - wisdom
[wisdom@10 ~]$ /mnt/app/kafka/bin/kafka-server-start.sh -daemon /mnt/app/kafka/config/server.properties
kafka cluster install
kafka安装
1
2
3
4
5
6
7[root@10 ~]# cd /mnt/ops/app/
[root@10 app]# tar xzf kafka_2.11-0.9.0.0.tgz
[root@10 app]# mv kafka_2.11-0.9.0.0 /mnt/app/kafka
[root@10 app]# chown -R wisdom.wisdom /mnt/app/kafka
[root@10 app]# mkdir -p /mnt/{data,log}/kafka
[root@10 app]# chown -R wisdom.wisdom /mnt/{data,log}/kafkakafka配置文件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85[root@10 app]# cp /mnt/app/kafka/config/{server.properties,server.properties.bak}
kafka cluster-1:
[root@10 app]# cat > /mnt/app/kafka/config/server.properties <<EOF
> broker.id=113
> advertised.host.name=10.0.2.113
> advertised.port=9092
> delete.topic.enable=true
> listeners=PLAINTEXT://:9092
> num.network.threads=4
> num.io.threads=8
> socket.send.buffer.bytes=102400
> socket.receive.buffer.bytes=102400
> socket.request.max.bytes=104857600
> log.dirs=/mnt/data/kafka
> num.partitions=3
> num.recovery.threads.per.data.dir=2
> default.replication.factor = 1
> replica.fetch.max.bytes=20000000
> num.replica.fetchers=2
> message.max.bytes=10000000
> log.flush.interval.messages=10000
> log.flush.interval.ms=1000
> log.retention.hours=48
> log.segment.bytes=1073741824
> log.retention.check.interval.ms=300000
> zookeeper.connect=10.0.2.113:2181,10.0.2.114:2181,10.0.2.115:2181
> zookeeper.connection.timeout.ms=6000
> EOF
kafka cluster-2:
[root@10 app]# cat > /mnt/app/kafka/config/server.properties <<EOF
> broker.id=114
> advertised.host.name=10.0.2.114
> advertised.port=9092
> delete.topic.enable=true
> listeners=PLAINTEXT://:9092
> num.network.threads=4
> num.io.threads=8
> socket.send.buffer.bytes=102400
> socket.receive.buffer.bytes=102400
> socket.request.max.bytes=104857600
> log.dirs=/mnt/data/kafka
> num.partitions=5
> num.recovery.threads.per.data.dir=2
> default.replication.factor = 1
> replica.fetch.max.bytes=20000000
> num.replica.fetchers=2
> message.max.bytes=10000000
> log.flush.interval.messages=10000
> log.flush.interval.ms=1000
> log.retention.hours=48
> log.segment.bytes=1073741824
> log.retention.check.interval.ms=300000
> zookeeper.connect=10.0.2.113:2181,10.0.2.114:2181,10.0.2.115:2181
> zookeeper.connection.timeout.ms=6000
> EOF
kafka cluster-3:
[root@10 app]# cat > /mnt/app/kafka/config/server.properties <<EOF
> broker.id=115
> advertised.host.name=10.0.2.115
> advertised.port=9092
> delete.topic.enable=true
> listeners=PLAINTEXT://:9092
> num.network.threads=4
> num.io.threads=8
> socket.send.buffer.bytes=102400
> socket.receive.buffer.bytes=102400
> socket.request.max.bytes=104857600
> log.dirs=/mnt/data/kafka
> num.partitions=5
> num.recovery.threads.per.data.dir=2
> default.replication.factor = 1
> replica.fetch.max.bytes=20000000
> num.replica.fetchers=2
> message.max.bytes=10000000
> log.flush.interval.messages=10000
> log.flush.interval.ms=1000
> log.retention.hours=48
> log.segment.bytes=1073741824
> log.retention.check.interval.ms=300000
> zookeeper.connect=10.0.2.113:2181,10.0.2.114:2181,10.0.2.115:2181
> zookeeper.connection.timeout.ms=6000
> EOFkafka log存放位置
1
2
3
4
5
6
7
8
9[root@10 app]# vim /mnt/app/kafka/bin/kafka-run-class.sh
# Log directory to use
if [ "x$LOG_DIR" = "x" ]; then
LOG_DIR="/mnt/log/kafka"
JMX_PORT=8092
else
LOG_DIR="/mnt/log/kafka"
JMX_PORT=8092
fikafka jvm修改
1
2
3
4[root@10 app]# vim /mnt/app/kafka/bin/kafka-server-start.sh
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-Xmx4G -Xms4G -Xmn2G -XX:PermSize=64m -XX:MaxPermSize=128m -XX:SurvivorRatio=6 -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly"
fikafka 启动
1
2[root@10 app]# su - wisdom
[wisdom@10 ~]$ /mnt/app/kafka/bin/kafka-server-start.sh -daemon /mnt/app/kafka/config/server.properties
java performance analysis
jps 分析
1 | * jps位于jdk的bin目录下,其作用是显示当前系统的java进程情况,及其id号 |
zookeeper optimize
zookeeper参数说明
1 | 1.tickTime：Client-Server通信心跳时间 |
zookeeper 参数调优
1 | 1.zookeeper默认jvm没有配置Xmx,Xms等信息,可以在conf目录下创建java.env文件 |
command split
- split 分割文件
1
2
3
4
5
6
7
8[root@localhost ~]# split -l 10000 ccc -d -a 2 num_
说明:
-l 表示按照多少行进行分割
-d 表示分割后的文件名以数字结尾
-a 2 表示结尾是2个字符,文件后缀
-d -a 2 表示分割后的文件以2个数字为后缀
num_ 表示分割后的文件名
postgresql plugin install
1. pg 安装本地插件库(all)
1 | [root@localhost app]# tar xzf postgresql-9.6.1.tar.gz |
2. pg_pathman download(https://pgxn.org/dist/pg_pathman/)
pg_pathman download
1
2
3[root@localhost ~]# git clone https://github.com/postgrespro/pg_pathman /mnt/ops/app/postgresql-9.6.1/contrib/pg_pathman
说明: 我将pg_pathman下载到postgresql源码的插件目录contrib目录下pg_pathman install
1
2
3[root@localhost ~]# cd /mnt/ops/app/postgresql-9.6.1/contrib/pg_pathman
[root@localhost pg_pathman]# make
[root@localhost pg_pathman]# make install在配置文件中添加pg_pathman参数
1
2
3
4[root@localhost pg_pathman]# su - wisdom
[wisdom@localhost ~]$ vim /mnt/data/pgsql/postgresql.conf
shared_preload_libraries = 'pg_pathman,pg_stat_statements'
[wisdom@localhost ~]$ /mnt/app/pgsql/bin/pg_ctl -D /mnt/data/pgsql/ -l /mnt/log/pgsql/pgsql.log restartpg_pathman use
1
2
3
4[root@localhost pg_pathman]# su - wisdom
[wisdom@localhost ~]$ /mnt/app/pgsql/bin/psql -d postgres
postgres=# create extension pg_pathman;
CREATE EXTENSION
command sed
删除行首空格
1
2
3
4
5
6
7sed 's/^[ \t]*//g'
说明：
第一个/的左边是s表示替换,即将空格替换为空
第一个/的右边是表示后面的以xx开头中括号表示“或”，空格或tab中的任意一种。
这是正则表达式的规范。 中括号右边是*，表示一个或多个。
第二个和第三个\中间没有东西，表示空 g表示替换原来buffer中的，sed在处理字符串的时候并不对源文件进行直接处理，先创建一个buffer，但是加g表示对原buffer进行替换 整体的意思是：用空字符去替换一个或多个用空格或tab开头的本体字符串删除行末空格
1
2
3
4sed 's/[ \t]*$//g'
和上面稍微有些不同是前面删除了^符,在后面加上了美元符,这表示以xx结尾的字符串为对象。
但是要注意在KSH中，Tab并不是\t而是直接打入一个Tab就可以了。