azkaban 任务流设置

azkaban 简单demo

一. azkaban job简单配置

1. 目录结构

  • lib 目录下，为java的jar包
  • script 目录下，是shell脚本
  • sql 目录下，是hive sql脚本

2. 定义flow1信息

2.1. f1/az-hive-01.job

1
2
3
4
5
type: hive 
azk.hive.action: execute.query
hive.script: sql/az-hive-01.sql
hiveconf.tb: test.action_log
hiveconf.rows: 5
  • hiveconf.xxx 指定需要传入sql的参数设置，sql文件内用${hiveconf:xx} 表示

2.2. f1/az-java-02.job

1
2
3
type=javaprocess
classpath=./lib/*
java.class=nick.weixx.az.run.MyJavaJob
  • java任务，指定运行的main class

2.3. f1/az-shell.job

1
2
3
type: command
command: sh ./script/az-comm-01.sh
dependencies: az-hive-01,az-java-02
  • 执行shell 脚本，dependencies,表示需要等待，hive以及java任务执行成功后，执行。

2.4 f1/az-f1-end.job 定义flow

1
2
3
4
type: flow
flow.name: az-shell
failure.emails: weixinxing0226@gmail.com
success.emails: weixinxing0226@163.com
  • 需要和 f1无依赖的最终任务名称(az-shell.job)一致

3. 定义flow2信息

3.1. f2/az-f2-1.job

1
2
type: command
command: echo "i' im flow 2."

3.2. f2/az-f2-end.job 定义flow2信息

1
2
3
4
type: flow
flow.name: az-f2-1
failure.emails: weixinxing0226@gmail.com
success.emails: weixinxing0226@163.com
  • flow.name 需要和flow2 的任务名称一致，既az-f2-1.job

3.3. az-end.job 定义最终任务信息

1
2
3
4
type: noop
dependencies: az-f1-end,az-f2-end
failure.emails: weixinxing0226@gmail.com
success.emails: weixinxing0226@163.comw
  • 需要等待 flow1,flow2执行完毕

4. 打包运行

4.1. 打zip包，并上传

4.2. 设定任务

4.3. 设置定时执行,SLA报警

4.4. 运行结果

二. azkaban flow简单配置

1. 目录结构

  • lib 目录下，为java的jar包
  • script 目录下，是shell脚本
  • sql 目录下，是hive sql脚本
  • azkaban-demo-flow.project 定义执行flow的版本
  • azkaban-demo-flow.flow 定义工作流详细

2. 工作流设定

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
---
##成功或者失败的处理方式
config:
  failure.emails: weixinxing0226@gmail.com
  success.emails: weixinxing0226@163.com

##多任务依赖配置
nodes:
  - name: azkaban-demo-flow
    type: noop
    dependsOn:
      - hive-01
      - shell-02
      - java-03

  - name: hive-01
    type: hive
    config:
      azk.hive.action: execute.query
      hive.script: azkaban-demo-flow/sql/az-hive-01.sql
      hiveconf.tb: test.action_log
      hiveconf.rows: 5

  - name: shell-02
    type: command
    config:
      command: sh ./azkaban-demo-flow/script/az-comm-01.sh

  - name: java-03
    type: javaprocess
    config:
      classpath: ./azkaban-demo-flow/lib/*
      java.class: nick.weixx.az.run.MyJavaJob
    dependsOn:
       - hive-01
       - shell-02

[demo地址] (git@github.com:nick-weixx/azkaban-demo.git)

展开全文 >>

azkaban 环境搭建

azkaban 环境搭建，插件安装，简单demo，功能演示。

一. azkaban 环境搭建

1. azkaban 代码编译

  • 打开 azkaban 源码地址,将代码下载下来。

    1
    2
    3
    4
    5
    6
    7
     ## 选择自己适合的版本，我选的 3.49.0版本，azkaban高版本与 azkaban插件(3.0)不兼容

    git clone https://github.com/azkaban/azkaban.git

    git tag  

    git checkout 3.49.0

  • 编译代码(azkaban 用的是gradle进行的项目构建管理)

    1
    2
    ##没装gradle的，使用下面命令跳过测试构建环境
    ./gradlew clean build  -x test

  • 编译包获取

这个包只能安装到linux环境，因为exection调用了linux /proc/loadavg 与 /proc/meminfo 等命令。

  • mac 环境下需要修改azkaban.execapp.ServerStatisticsServlet 类，将判断条件去掉

2. 环境准备

  • 创建mysql账号及数据库
1
2
3
4
5
6
7
8
9
10
11
12
13
###创建azkaban用户
CREATE USER 'azkaban'@'%' IDENTIFIED BY '******';

###赋权
GRANT ALL PRIVILEGES ON `azkaban_new`.* TO 'azkaban'@'%';

###创建数据库
CREATE DATABASE IF NOT EXISTS azkaban_new DEFAULT CHARSET utf8 COLLATE utf8_general_ci;
flush privileges;

## 解压包
tar -zxvf azkaban-db-3.49.0.tar.gz
source azkaban-db-3.49.0/create-all-sql-3.49.0.sql
  • 配置环境变量(/etc/profile)
1
2
export HADOOP_HOME=/opt/cloudera/parcels/CDH-5.15.2-1.cdh5.15.2.p0.3
export HADOOP_CONF_DIR=/etc/hadoop/conf
注意: 一定要配置，否则下面的hdfs-viewer插件，将无法使用。

  • 解压文件

    1
    2
    3
    cd /opt
    tar -zxvf azkaban-exec-server-3.49.0.tar.gz
    tar -zxvf azkaban-web-server-3.49.0.tar.gz

3. 配置azkaban-web

  • 在服务根目录下配置ssl
1
2
3
4
cd azkaban-web-server-3.49.0

##生成keystore文件，需要填写一些信息
keytool -keystore keystore -alias jetty -genkey -keyalg RSA

  • 修改conf/azkaban.properties 文件
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

azkaban.name=nick-wei
azkaban.label=My Local Azkaban
azkaban.color=#FF3601
azkaban.default.servlet.path=/index
web.resource.dir=web/

#时区设置
default.timezone.id=Asia/Shanghai
# Azkaban UserManager class
user.manager.class=azkaban.user.XmlUserManager
user.manager.xml.file=conf/azkaban-users.xml
# Loader for projects
executor.global.properties=conf/global.properties
azkaban.project.dir=projects
# Velocity dev mode
velocity.dev.mode=false

# Azkaban Jetty server properties.
#jetty.use.ssl=false
jetty.maxThreads=25
jetty.ssl.port=8443
jetty.port=8081
jetty.keystore=keystore
jetty.password=weixingxing
jetty.keypassword=weixingxing
jetty.truststore=keystore
jetty.trustpassword=weixingxing


# Azkaban Executor settings
executor.port=12321


# email 设置
mail.sender=xxxx@163.com
mail.host=smtp.163.com
mail.user=xxxx@163.com
mail.password=xxxx
#mail.port=465
mail.port=25


job.failure.email=
job.success.email=
lockdown.create.projects=false
cache.directory=cache
# JMX stats
jetty.connector.stats=true
executor.connector.stats=true


# mysql 配置
database.type=mysql
mysql.port=3306
mysql.host=xxx
mysql.database=azkaban_new
  • 修改 conf/azkaban-users.xml 文件
1
2
3
4
5
6
7
8
9
##此文件是用户权限控制文件，可以修改也可以使用默认的
<azkaban-users>
  <user groups="azkaban" password="azkaban" roles="admin" username="azkaban"/>
  <user password="metrics" roles="metrics" username="metrics"/>

  <user username="admin" password="admin" roles="admin,metrics" />
  <role name="admin" permissions="ADMIN"/>
  <role name="metrics" permissions="METRICS"/>
</azkaban-users>

3. 配置azkaban-exec-server

  • 修改 conf/azkaban.properties
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
## 时区设置
default.timezone.id=Asia/Shanghai

# 权限设置
user.manager.class=azkaban.user.XmlUserManager
user.manager.xml.file=conf/azkaban-users.xml

# 全局配置
executor.global.properties=conf/global.properties
## project 本地缓存存放地址
azkaban.project.dir=projects
# exec server 服务端口配置
executor.port=12321
executor.connector.stats=true

# Velocity dev mode
velocity.dev.mode=false

# Azkaban Jetty server properties.
jetty.maxThreads=25
jetty.ssl.port=8443
jetty.port=8081
jetty.keystore=keystore
jetty.password=weixingxing
jetty.keypassword=weixingxing
jetty.truststore=keystore
jetty.trustpassword=weixingxing
#jetty.use.ssl=false
# JMX stats
jetty.connector.stats=true

# 邮箱设置
mail.sender=
mail.host=
job.failure.email=
job.success.email=
lockdown.create.projects=false
cache.directory=cache

# mysql 配置
database.type=mysql
mysql.port=3306
mysql.host=h1
mysql.database=azkaban_new
mysql.user=azkaban
mysql.password=azkaban
mysql.numconnections=100


#Multiple Executor,exec需要的资源
azkaban.use.multiple.executors=true
azkaban.executorselector.filters=StaticRemainingFlowSize,MinimumFreeMemory,CpuStatus
azkaban.executorselector.comparator.NumberOfAssignedFlowComparator=1
azkaban.executorselector.comparator.Memory=1
azkaban.executorselector.comparator.LastDispatched=1
azkaban.executorselector.comparator.CpuUsage=1
azkaban.executorselector.comparator.LastDispatched=1
azkaban.executorselector.comparator.CpuUsage=1

##jobtype插件地址
azkaban.jobtype.plugin.dir=plugins/jobtype

二. azkaban-plugins 环境搭建

1. azkaban-plugins 代码编译

1
2
## 选择azkaban插件(3.0)最高版本
git clone https://github.com/azkaban/azkaban-plugins.git
  • 编译代码（azkaban-plugins 用的是ant进行的项目构建管理）
1
2
cd azkaban-plugins
ant
注意：在更目录下，修改build.xml文件，注释掉 pigvisualizer 引用

  • 编译包获取
1
find . -type f \( -iname "azkaban-jobtype*.tar.gz" -o -iname "azkaban-javaviewer*.tar.gz" -o -iname "azkaban-jobsummary*.tar.gz" -o -iname "azkaban-hdfs-viewer*.tar.gz" -o -iname "azkaban-reportal*.tar.gz"  \)

2. 配置azkaban-web 插件

  • (1) 配置hdfs-viewer插件
1
2
3
4
5
cd plugins/viewer/
##解压文件
tar -zxvf azkaban-hdfs-viewer-3.0.0.tar.gz
##重命名插件
mv azkaban-hdfs-viewer-3.0.0 hdfs

修改配置文件: hdfs/conf/plugin.properties

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
viewer.name=HDFS
viewer.path=hdfs
viewer.order=1
viewer.hidden=false
viewer.external.classpaths=extlib/*
viewer.servlet.class=azkaban.viewer.hdfs.HdfsBrowserServlet
#hadoop.security.manager.class=azkaban.security.HadoopSecurityManager_H_1_0

##hadoop2.x 需要用这个管理器
hadoop.security.manager.class=azkaban.security.HadoopSecurityManager_H_2_0
azkaban.should.proxy=false
proxy.user=azkaban
proxy.keytab.location=
allow.group.proxy=false
file.max.lines=1000
注意：一定要在/etc/profile 里面配置HADOOP_HOME 和 HADOOP_CONF_DIR
  • (2) 配置javaviewer插件
1
2
3
4
5
cd plugins/viewer/
##解压文件
tar -zxvf azkaban-javaviewer-3.0.0.tar.gz
##重命名插件
mv azkaban-javaviewer-3.0.0 javaviewer

修改配置文件: javaviewer/conf/plugin.properties

1
2
3
4
5
6
7
8
viewer.name=JavaViewer
viewer.path=javaviewer
viewer.order=1
viewer.hidden=false
#viewer.external.classpaths=extlib/*
viewer.external.classpaths=extlib
viewer.servlet.class=azkaban.viewer.javaviewer.JavaViewerServlet
viewer.jobtypes=hadoopJava
  • (3) 配置jobsummary插件
1
2
3
4
5
cd plugins/viewer/
##解压文件
tar -zxvf azkaban-jobsummary-3.0.0.tar.gz
##重命名插件
mv azkaban-jobsummary-3.0.0 jobsummary

修改配置文件: jobsummary/conf/plugin.properties

1
2
3
4
5
6
7
viewer.name=Summary
viewer.path=jobsummary
viewer.order=1
viewer.hidden=true
viewer.external.classpaths=extlib
viewer.servlet.class=azkaban.viewer.jobsummary.JobSummaryServlet
viewer.jobtypes=java,hadoopJava,pig,hive
  • (4) 配置reportal插件
1
2
3
4
5
cd plugins/viewer/
##解压文件
tar -zxvf azkaban-reportal-3.0.0.tar.gz
##重命名插件
mv azkaban-reportal-3.0.0 reportal

修改配置文件: reportal/conf/plugin.properties

1
2
3
4
5
6
7
8
9
10
11
12
viewer.name=Reportal
viewer.path=reportal
viewer.order=1
viewer.hidden=false
viewer.external.classpaths=extlib
viewer.servlet.class=azkaban.viewer.reportal.ReportalServlet
azkaban.should.proxy=true
proxy.user=azkaban
proxy.keytab.location=
allow.group.proxy=true
reportal.output.filesystem=hdfs
hadoop.security.manager.class=azkaban.security.HadoopSecurityManager_H_2_0

3. 配置azkaban-exec 插件

  • (1) 配置azkaban-jobtype和reportal 插件
1
2
3
4
5
cd plugins/jobtypes/

##解压文件
tar -zxvf azkaban-jobtype-3.0.0.tar.gz
tar -zxvf azkaban-reportal-3.0.0.tar.gz

修改配置文件: plugins/jobtypes/common.properties

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
azkaban.should.proxy=false
## reportal 插件配置
reportal.execution.user=azkaban
#reportal.storage.user=reportal
#reportal.default.timezone=America/Los_Angeles
reportal.default.timezone=Asia/Shanghai

##必须设置hdfs，否则azkaban-web 接收不到文件
reportal.output.filesystem=hdfs
#reportal.output.dir=/tmp/reportal

## pig,hive,spark,hadoop 的home路径配置
#pig.home=
hadoop.home=/opt/cloudera/parcels/CDH/lib/hadoop
hive.home=/opt/cloudera/parcels/CDH/lib/hive
spark.home=/opt/cloudera/parcels/CDH/lib/spark

##通用hadoop包，引用配置
hadoop.classpath=${hadoop.home}/etc/hadoop,${hadoop.home}/share/hadoop/common/*,${hadoop.home}/share/hadoop/common/lib/*,${hadoop.home}/share/hadoop/hdfs/*,${hadoop.home}/share/hadoop/hdfs/lib/*,${hadoop.home}/share/hadoop/yarn/*,${hadoop.home}/share/hadoop/yarn/lib/*,${hadoop.home}/share/hadoop/mapreduce/*,${hadoop.home}/share/hadoop/mapreduce/lib/*,${hadoop.home}/client-0.20/*

## 通用jar包引用配置
jobtype.global.classpath=${hadoop.home}/etc/hadoop,${hadoop.home}/share/hadoop/common/*,${hadoop.home}/share/hadoop/common/lib/*,${hadoop.home}/share/hadoop/hdfs/*,${hadoop.home}/share/hadoop/hdfs/lib/*,${hadoop.home}/share/hadoop/yarn/*,${hadoop.home}/share/hadoop/yarn/lib/*,${hadoop.home}/share/hadoop/mapreduce/*,${hadoop.home}/share/hadoop/mapreduce/lib/*,${hadoop.home}/client-0.20/*,/opt/cloudera/parcels/CDH-5.15.2-1.cdh5.15.2.p0.3/lib/avro/*,${hive.home}/conf/

## configs for jobtype security settings
execute.as.user=false

修改配置文件: plugins/jobtypes/commonprivate.properties

1
2
3
4
5
6
7
8
9
10
11
12

hadoop.security.manager.class=azkaban.security.HadoopSecurityManager_H_2_0

 azkaban.should.proxy=false
 obtain.binary.token=false
# obtain.namenode.token=true
# obtain.jobtracker.token=true

# global classpath items for all jobs. e.g. hadoop-core jar, hadoop conf
#jobtype.global.classpath=${hadoop.home}/*,${hadoop.home}/conf
# global jvm args for all jobs. e.g. java.io.temp.dir, java.library.path
#jobtype.global.jvm.args=
  • (2) jobtypes hive 配置修改

修改 hive/plugin.properties

1
2
3
4
hive.aux.jars.path=${hive.home}/aux/lib

hive.jvm.args=-Dhive.querylog.location=. -Dhive.exec.scratchdir=/tmp/hive-${user.to.proxy} -Dhive.aux.jars.path=${hive.aux.jars.path}
jobtype.jvm.args=${hive.jvm.args}

修改 hive/private.properties

1
2
3
4
5
6
7
8
9
jobtype.classpath=${hadoop.home}/conf,${hadoop.home}/lib/*,${hive.home}/lib/*,${hive.home}/conf,${hive.aux.jar.path}
jobtype.class=azkaban.jobtype.HadoopHiveJob

hive.aux.jar.path=${hive.home}/aux/lib

## definitely set hive home otherwise we dont have conf

#jobtype.global.classpath=${hive.classpath.items}
#hive.classpath.items=${hive.home}/lib/*,${hive.home}/conf/,${hive.aux.jars.path}/*,${hadoop.home}/*,${hadoop.home}/lib/*,${hadoop.home}/conf/
  • (3) jobtypes reportalhive 配置修改

修改 reportalhive/plugin.properties

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
job.class=azkaban.jobtype.ReportalHiveRunner

##不知道为什么,commonprivate.properties中的通用设置不好使
hive.home=/opt/cloudera/parcels/CDH/lib/hive
hadoop.home=/opt/cloudera/parcels/CDH/lib/hadoop
CDH-5.15.2-1.cdh5.15.2.p0.3

env.HIVE_HOME=${hive.home}
env.HIVE_AUX_JARS_PATH=${hive.aux.jars.path}

hive.jvm.args=-Dhive.querylog.location=. -Dhive.exec.scratchdir=/tmp/hive-${user.to.proxy} -Dhive.aux.jars.path=${hive.aux.jars.path}
jobtype.global.jvm.args=${hive.jvm.args}
hive.aux.jars.path=${hive.home}/aux/lib

hadoop.dir.bin=${hadoop.home}/bin
hadoop.dir.conf=${hadoop.home}/etc/hadoop
不想编译的话可以从我的百度云盘获取：azkaban-3.49.0 ,提取码：* (联系我)

源码地址

三. azkaban 运行

1. azkaban-exec 服务启动

1
2
cd azkaban-exec-server-3.49.0
sh ./bin/start-exec.sh

executors 表中的active 字段，修改成1。否则web服务启动会报错

2. azkaban-web 服务启动

1
2
cd azkaban-web-server-3.49.0
sh ./bin/start-web.sh

3. 运行效果

展开全文 >>

技术框架介绍(大数据篇)

这是我第一篇博客文章，千里之行始于脚下，希望自己养成良好的习惯，记录下工作生活中的点点滴滴。每天进步一点，积少成多，沉淀升华。 整理了一下从业需要的了解的相关技术，想慢慢的一点点细化这些架构，组合运用，可以做出一些好玩的东西出来。

hadoop

hadoop 是一个开发和运行大规模数据的软件平台，属于appach组织，是一个开源项目，主要由java编写实现。核心架构包含hdfs(分布式存储)与mapreduce(分布式计算)组成。目前最新版本为3.0.0。它最早设计用于搜索引擎，为搜索存储海量数据，并进行索引建立。

More info: 官网

hdfs

hdfs是一个分布式存储系统，源自于Google的GFS论文，简单理解为google实现的克隆版。它是一个高度容错的系统，能检测和应对硬件故障，用于在低成本的通用硬件上运行。

Alt text
Client：切分文件；访问HDFS；与NameNode交互，获取文件位置信息；与DataNode交互，读取和写入数据。

NameNode：Master节点，在hadoop1.X中只有一个，管理HDFS的名称空间和数据块映射信息，配置副本策略，处理客户端请求。

DataNode：Slave节点，存储实际的数据，汇报存储信息给NameNode。

Secondary NameNode：辅助NameNode，分担其工作量；定期合并fsimage和fsedits，推
送给NameNode；紧急情况下，可辅助恢复NameNode，但Secondary NameNode并非NameNode的热备

mapreduce

源自于google的MapReduce论文，Hadoop MapReduce是google MapReduce 克隆版。是一个分布式计算框架，mapreduce 实现了算法迁移，写好的程序会被分配到各个datanode节点执行。mapreduce会频繁的读写本地文件，进行排序，归类。

Alt text

input: 要读取的文件列表，获取文件大小位置信息

splitting: 对文件进行切割,分片到各个节点

mapping: 详细的读取逻辑，组合成 k-v 形式，存储文件

shuffing: 根据key合并结果集，将数据分发到不同的reduce

reducing: 合并后结果，具体的处理逻辑实现

final result: 输出计算结果

hive

由facebook开源，解决了海量数据结构化查询的问题。运行在mapredece之上，记性了sql化的封装。适用于离线数据的计算，隔离了mapreduce编写，大大提升了效率。

Alt text

More info: 官网

zookeeper

源自Google的Chubby论文,是Hadoop和Hbase的重要组件。提供的功能包括：配置维护、域名服务、分布式同步、组服务等。可以理解为分布式的监听模式实现。

More info: 官网

spark

spark 为迭代式数据处理提供更好的支持。每次迭代的数据可以保存在内存中，而不是写入文件.相对于mapreduce框架，频繁的将文件读写到磁盘，性能上有极大的提升。而且spark提供了一套完整的解决方案，包括spark sql (交互封装),spark streaming (流式计算),MLlib(机器学习), GraphX(图计算)等一套完整方案 。它将动作封装成了一个个RDD，分为转化和动作两类。

More info: 官网

oozie

是一个工作流框架，可以将mapreduce任务，hive任务，shell脚本，数据库访问。。。。,根据业务封装成工作流。工作流必须标记start,end节点，可以通过kill节点杀死任务，也可以通过fork&join节点并行的执行任务。同时提供web端的任务流查询，是一个非常优秀的一个框架。

More info: 官网

hbase

源自Google的Bigtable论文,由facebook开源,是一个键值对数据库，具有可伸缩，过性能，分布式的特点。可以对大规模数据进行随机访问，实时读写，是一个横向无限大的表。

Alt text

More info: 官网

mahout

Mahout 是一个很强大的数据挖掘工具，是一个分布式机器学习算法的集合，包括：被称为Taste的分布式协同过滤的实现、分类、聚类等。Mahout最大的优点就是基于hadoop实现，把很多以前运行于单机上的算法，转化为了MapReduce模式，这样大大提升了算法可处理的数据量和处理性能。

More info: 官网

展开全文 >>

springboot mybatis集成及dubbo集成

这里创建了一个小demo，沿用上期blog需求。但是底层查询数据的架构使用了mybatis，同时使用dubbo 实现了 provider与cusumer。 provider 用于blog文章的数据提供，cusumer 消费provider并且通过restful 形式的接口提供给前端调用。

一. mybatis 集成

1. maven依赖

添加 mybatis-spring-boot-starter 与 com.alibaba连接池

1
2
3
4
5
6
7
8
9
10
11

<dependency>
			<groupId>org.mybatis.spring.boot</groupId>
			<artifactId>mybatis-spring-boot-starter</artifactId>
			<version>1.3.1</version>
</dependency>
	<dependency>
			<groupId>com.alibaba</groupId>
			<artifactId>druid</artifactId>
			<version>1.0.18</version>
	</dependency>

2. 编写 POJO对象及实现数据访问层访问

(1) POJO对象

和之前的hibernate 对象没有区别，只是将注解去掉了。

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
public class Article implements Serializable {
	private static final long serialVersionUID = -1491292595468157592L;
	// 标明自增id，及列名
	private int articleId;

	// 标明文章标题
	private String title;

	// 标明文章类别
	private String category;

	public int getArticleId() {
		return articleId;
	}

	public void setArticleId(int articleId) {
		this.articleId = articleId;
	}

	public String getTitle() {
		return title;
	}

	public void setTitle(String title) {
		this.title = title;
	}

	public String getCategory() {
		return category;
	}

	public void setCategory(String category) {
		this.category = category;
	}

	@Override
	public String toString() {
		return "Article [articleId=" + articleId + ", title=" + title + ", category=" + category + "]";
	}

}

(2) 数据访问层实现

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
@Mapper
public interface IArticleDao {

	@Select("SELECT article_id,title,category FROM articles")
	@Results({ @Result(property = "articleId", column = "article_id", javaType = Integer.class),
			@Result(property = "title", column = "title"), @Result(property = "category", column = "category")

	})
	List<Article> getAllArticles();
	
	@Select("SELECT article_id,title,category FROM articles where article_id=#{articleId}")
	@Results({ @Result(property = "articleId", column = "article_id", javaType = Integer.class),
			@Result(property = "title", column = "title"), @Result(property = "category", column = "category")
	})
	Article getArticleById(int articleId);
	
	@Insert("INSERT INTO articles(title,category) VALUES(#{title},#{category})")
	int addArticle(Article article);
	@Update("UPDATE articles SET category=#{category} WHERE article_id=#{articleId}")
	void updateArticle(Article article);
	
	@Delete("DELETE FROM articles WHERE article_id=#{articleId}")
	void deleteArticle(@Param("articleId") int articleId);
	
	@Select("SELECT article_id FROM articles WHERE title=#{title} AND category=#{category}")
	Integer articleExists(@Param("title") String title, @Param("category") String category);
}
  • 全部使用注解完成，接口上使用@Mapper 标明为访问层mapper代码实现
  • 通过@Result 注解，标明映射关系
  • @SELECT 注解标明为查询语句，@DELETE 标明为删除语句，@INSERT 标明为插入语句

(3) hibernate vs mybatis

因为我们做了分层，下面的数据库交互框架的变动并不影响业务层和mvc的使用。所以上层并不需要修改。到此访问层架构的提供已完成。非常简单。

这里多说几句，hibernate与mybatis都用过，这两种框架，各有各的优劣，没有绝对的好与坏。选型需要很多因素的评估，团队的整体水平，程序开发周期时长，底层使用的数据库，使用的场景，对性能的要求都是制约的条件。

Mybatis优势
  • MyBatis可以进行更为细致的SQL优化，可以减少查询字段。
  • MyBatis容易掌握，而Hibernate门槛较高。
  • 轻量级，可控性高。
Hibernate优势
  • Hibernate的DAO层开发比MyBatis简单，Mybatis需要维护SQL和结果映射。
  • Hibernate对对象的维护和缓存要比MyBatis好，对增删改查的对象的维护要方便。
  • Hibernate数据库移植性很好，MyBatis的数据库移植性不好，不同的数据库需要写不同SQL。
  • Hibernate有更好的二级缓存机制，可以使用第三方缓存。MyBatis本身提供的缓存机制不佳。

感兴趣的可以看一下：https://www.zhihu.com/question/21104468

二. provider 开发

1. maven引入

1
2
3
4
5
6

<dependency>
	<groupId>io.dubbo.springboot</groupId>
	<artifactId>spring-boot-starter-dubbo</artifactId>
	<version>1.0.0</version>
</dependency>

2. provider 实现

1
2
3
4
5
6
7
8
9
10
11
12

@Service(version = "1.0.0")
public class ArticleDubboServiceImpl implements IArticleDubboService {
	@Autowired
	private IArticleService articleService;

	@Override
	public Article getArticleById(Integer id) {
		return articleService.getArticleById(id);
	}

}
  • 使用注解的方式标明服务版本，@Service(version = 1.0.0)
  • 调用之前实现的服务层代码

3. 文件配置

application.properties 文件汇中配置

1
2
3
4
5
6
7
8
9
10
11
12
13

server.port=9999

spring.datasource.url=jdbc:mysql://xxxxxxx:3306/spring_boot
spring.datasource.username=xxxx
spring.datasource.password=xxxx
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

spring.dubbo.application.name=blog_provider
spring.dubbo.registry.address=zookeeper://127.0.0.1:2181
spring.dubbo.protocol.name=dubbo
spring.dubbo.protocol.port=20880
spring.dubbo.scan=com.nick.demo2.dubbo

4. 项目结构图

Alt text

  • com.nick.demo2.entity POJO实现
  • com.nick.demo2.dao 数据访问层实现
  • com.nick.demo2.service 业务层数显
  • com.nick.demo2.controller 控制器实现
  • com.nick.demo2.dubbo dubbo provider 实现

4. 注册服务

运行程序的 Application，将服务注册到zookeeper

效果图：

Alt text

三，编写消费者

1. 结构图

Alt text

  • com.nick.dubbo.consumer.controller restful 风格控制器实现
  • com.nick.demo2.entity 接口传输的对象
  • com.nick.demo2.dubbo 消费者实现

2. 消费者实现

1
2
3
4
5
6

@Component
public class DubboService {
	@Reference(version = "1.0.0")
	public IArticleDubboService articleDubbo;
}

  • 非常简单指定消费接口就行

  • 注意：
    消费者中用到的接口包路径及model必须与，服务提供者一致。其实很好理解，因为要序列化传输，所以两边的对象必须一致。所以 才会出现那种不太正常的包结构(com.nick.demo2.dubbo)。 这个可以单独抽取成一个jar包，供两边实现。很简单。

3。 配置

application.properties 文件汇中配置

1
2
3
4
5
server.port=9998

spring.dubbo.application.name=consumer
spring.dubbo.registry.address=zookeeper://127.0.0.1:2181
spring.dubbo.scan=com.nick.demo2.dubbo

4. 启动消费者及演示

启动 CosumerApplication类

效果图

Alt text

Alt text

dubbo配置详细说明：

官网： http://dubbo.io/books/dubbo-user-book/preface/background.html

他每做一件小事的时候，都好像抓住一根救命稻草，到最后你才发现，他抱住的已经是一棵参天大树了。–《士兵突击》

展开全文 >>

springboot hibernate集成及RESULFUL风格接口

我们创建一个简单的例子，这个例子中使用springboot+hibernate 实现了与mysql数据库的交互。同时使用RESULFUL风格http接口，实现数据访问展示，使用json格式数据进行交互。

假定的需求场景是博客的文章管理，对图书进行CRUD操作，这是一个非常简单的场景。

一，设计数据库

1
2
3
4
5
6
7
8
9
10
11
CREATE TABLE `articles` (
  `article_id` int(5) NOT NULL AUTO_INCREMENT COMMENT '自增的文章编号',
  `title` varchar(200) NOT NULL DEFAULT '' COMMENT '标题',
  `category` varchar(100) NOT NULL DEFAULT '' COMMENT '类别',
  PRIMARY KEY (`article_id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8;

INSERT INTO `articles` (`article_id`, `title`, `category`) VALUES
	(1, 'Java Concurrency', 'Java'),
	(2, 'Hibernate HQL ', 'Hibernate'),
	(3, 'Spring MVC with Hibernate', 'Spring');

这张表非常的简单，我们只添加了标题，与文章类别。同事插入了三行数据。

二，maven 及项目结构划分

1. maven 文件

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
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.nick</groupId>
	<artifactId>spring-boot-demo1</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>demo-1</name>
	<description>Demo project for Spring Boot</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>1.5.9.RELEASE</version>
		<relativePath /> <!-- lookup parent from repository -->
	</parent>
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-jpa</artifactId>
		</dependency>
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>


</project>

配置描述：

  • spring-boot-starter-parent 用于依赖管理的父POM
  • spring-boot-starter-web：构建Web，REST应用程序的入口。它使用tomcat服务器作为默认的嵌入式服务器。
  • spring-boot-starter-data-jpa：启动Spring数据JPA管理
  • spring-boot-devtools：它提供了开发工具,代码热部署等
  • spring-boot-maven-plugin：用于创建应用程序的可执行JAR。

2. 项目结构

Alt text

结构描述：

  • entity 数据库映射model
  • dao 数据访问层实现
  • service 业务层实现
  • controller 控制器实现

二， 数据访问层实现

1. 编写entity

我们使用的是注解的方式，配置entity。

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
import java.io.Serializable;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.Table;
@Entity //标明为Entity
@Table(name="articles") //标明映射的表名
public class Article implements Serializable { 
	private static final long serialVersionUID = 1L;
	@Id
	@GeneratedValue(strategy=GenerationType.AUTO)
	@Column(name="article_id")//1.标明自增id，及列名
        private int articleId;  
	@Column(name="title") //2. 标明文章标题
        private String title;
	@Column(name="category")//3. 标明文章类别
	private String category;
	public int getArticleId() {
		return articleId;
	}
	public void setArticleId(int articleId) {
		this.articleId = articleId;
	}
	public String getTitle() {
		return title;
	}
	public void setTitle(String title) {
		this.title = title;
	}
	public String getCategory() {
		return category;
	}
	public void setCategory(String category) {
		this.category = category;
	}
}

非常简单,很容易理解。

2. 编写访问层实现

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
@Transactional
@Repository
public class ArticleDAOImpl implements IArticleDAO {

	@PersistenceContext	
	private EntityManager entityManager;	
	@Override
	public Article getArticleById(int articleId) {
		return entityManager.find(Article.class, articleId);
	}
	
	/**
	 * 获取所有
	 */
	@SuppressWarnings("unchecked")
	@Override
	public List<Article> getAllArticles() {
		String hql = "FROM Article as atcl ORDER BY atcl.articleId";
		return (List<Article>) entityManager.createQuery(hql).getResultList();
	}	
	
	/**
	 * 添加
	 */
	@Override
	public void addArticle(Article article) {
		entityManager.persist(article);
	}
	.....
  • 类上面标记了@Transactional 受事务控制，同时标了@Repository 标注数据访问组件。
  • EntityManager 上面标@PersistenceContext,实体管理器，用于执行持久化操作。

三，业务层与控制器实现

1. 业务层实现

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
@Service
public class ArticleService implements IArticleService {

	@Autowired
	private IArticleDAO articleDAO;

	@Override
	public Article getArticleById(int articleId) {
		Article obj = articleDAO.getArticleById(articleId);
		return obj;
	}

	@Override
	public List<Article> getAllArticles() {
		return articleDAO.getAllArticles();
	}

	@Override
	public synchronized boolean addArticle(Article article) {
		if (articleDAO.articleExists(article.getTitle(), article.getCategory())) {
			return false;
		} else {
			articleDAO.addArticle(article);
			return true;
		}
	}
  • 使用@Service标明，为服务层。
  • IArticleDAO 上面标 @Autowired，引入数据层实现

2. 控制器实现

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
@Controller
@RequestMapping("user")
public class ArticleController {
	@Autowired
	private IArticleService articleService;

	/**
	 * 获取文章信息,设备get方式
	 * @param id
	 * @return
	 */
	@GetMapping("article/{id}")
	public ResponseEntity<Article> getArticleById(@PathVariable("id") Integer id) {
		Article article = articleService.getArticleById(id);
		return new ResponseEntity<Article>(article, HttpStatus.OK);
	}
	
	/**
	 * 获取所有
	 * @return
	 */
	@GetMapping("articles")
	public ResponseEntity<List<Article>> getAllArticles() {
		List<Article> list = articleService.getAllArticles();
		return new ResponseEntity<List<Article>>(list, HttpStatus.OK);
	}
  • 类上面@Controller，@RequestMapping(“user”) 标明为控制器，且mapping为/user
  • getArticleById 方法上标明@GetMapping(“article/{id}”),既后面的参数id 为查询的文章编号

四，application.properties 配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
##server
server.port=8888
##db 
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.url=jdbc:mysql://103.37.163.105:51000/spring_boot
spring.datasource.username=spring
spring.datasource.password=123
spring.datasource.tomcat.max-wait=20000
spring.datasource.tomcat.max-active=50
spring.datasource.tomcat.max-idle=20
spring.datasource.tomcat.min-idle=15
##jpa
spring.jpa.show-sql=true
spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.MySQLDialect
spring.jpa.properties.hibernate.id.new_generator_mappings = false
spring.jpa.properties.hibernate.format_sql = true
##log
logging.level.org.hibernate.SQL=DEBUG
logging.level.org.hibernate.type.descriptor.sql.BasicBinder=TRACE

五，演示

Alt text

Alt text

spring boot 帮助文档：https://docs.spring.io/spring-boot/docs/current/reference/html/

demo git地址： https://github.com/nick-weixx/springboot-demo1

展开全文 >>

java 错误及gc问题排查

最近参加了一次公司的分享，讲的非常的好。都是实战加方法论总结出来的东西。怕自己忘记，希望通过写博客的方式记录下来，顺便整理消化成自己的东西。

一，问题分类

1. jvm问题

(1) 分类

  • 内存占用过大，内存溢出等
  • 线程数量，线程死锁等
  • GC问题
  • ClassLoader/Class 问题(Class 未找到，转化失败)

(2) 方法论

  • 首先需要确认自己程序，是内存密集型还是cpu密集型服务。
  • 在自己重要业务上一定要将日志加全，好在后期排查问题。
  • 依赖外部开源监控程序，实时监控自己的业务。

（3）内存问题排查

排查其他系统占用

  • 首先 top (shift+m) ,按内存排序，查看进程所占用的内存比率
  • vmstat 3 (3秒一次)查看 Swap下的si(每秒交换回内存)，so(每秒交换出内存).这两个值越大证明系统越繁忙。

排查外部内存泄露

  • ?

确定为自己程序问题

  • 如果还是有问题，查看GC，jstat -gcutil pid 5s 60 (每隔5秒，执行60次)，查看 FGC(full GC)情况。如果老年代内存一直不释放，且一直在做GC。内存没有释放，打印堆栈查看

  • 首先 jmap -histo pid 打印出程序jvm class的实例数目,内存占用,类全名信息。对象分布情况。如果是web程序，需要查看nio的direct buffer，它通过jni申请了堆外内存，非jvm内存。 查看自己的连接是否未释放等。

(4) cpu 过高

  • top (shift+p) 查看cpu使用率
  • 超过预期值，使用jstack 查看当前线程再做什么操作
  • 线程死锁，查看是否程序一直在GC，而导致的这个问题。
  • 另一种可能就是当前程序某些线程有问题，使用 jstack 查看当前所有线程在干什么。
1
2
3
4
5
6
1. top (shift+p)       查看cpu使用率
2. jstat -gcutil  pid  查看gc情况
3. jmap –heap pid/jmap –histo pid 查看堆对象存储快照
4. top -Hp pid         查看当前进程的线程使用资源情况
5. printf 0x%x tid     线程id转化为16进制id
6. jstack pid | grep nid –A 10  使用jstack 查看 当前线程在做什么操作，后10行信息

2. 问题发现点

(1). cpu 异常

  • 复杂的CPU运算
  • 死循环
  • 频繁的fullgc

(2). 内存异常

  • 堆内存
  • metaspace/thread stack
  • 非堆内存

(3). IO异常

  • 流量异常，如频繁的远程数据库I/O
  • time_wait/close_wait

3. 工具

（1）linux

  • top、free、vmstat、sysstat(iostat/sar)、netstat、iftop

（2）jvm

  • jps、jinfo、jmap、jstat、jstack

（3） jmx

  • jvisualvm及其相关插件、jmc、jfr，MXBean

（4） 其他

  • mat
  • google-perftools
  • btrace
  • show-busy-java-threads

4. 博客推荐

注： 上面内容还没有完全整理完毕，会抽取时间不断完善。

展开全文 >>

你好，spring boot

一，spring boot 介绍

Spring Boot使您可以轻松创建独立的生产级基于Spring的应用程序，您可以“just run”。我们对Spring平台和第三方库进行了自己的封装，所以你可以非常容易的开始。大多数Spring Boot应用程序只需要很少的Spring配置。

特性：

  • 创建独立的Spring应用程序
  • 直接嵌入Tomcat，Jetty或Undertow（无需部署WAR文件）
  • 提供POM来简化您的Maven配置
  • 尽可能自动配置Spring
  • 提供生产就绪功能，如指标，运行状况检查和外部配置
  • 绝对不会生成代码，也不需要XML配置

spring boot 配合 spring cloud 非常适合于开发微服务，简单快速，灵活,实现SOA，服务注册发现。

二，spring boot 程序生成

1. 工具推荐

spring boot 程序可以通过 Spring Boot CLI 快速的生成原型，也可以直接创建maven项目，引用对应的spring boot 包。但是推荐大家使用sts(Spring Tool Sulte)开发工具，里面集成了所有spring boot的程序模板，非常的方便，而且它是在eclipse 的基础上开发而来，容易上手。

2. 写一个hello word

(1). 创建项目

Alt text

创建项目名称，配置maven的group id ,artifactId，版本等

Alt text

定义为web程序

Alt text

生成的项目结构图，是一个标准的maven项目。

(2). 结构简介

1. pom.xml 文件分析
1
2
3
4
5
6
7
8
9
10
11
12
13
14
  <parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>1.5.9.RELEASE</version>
</parent>
   <dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-web-services</artifactId>
</dependency>
<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-test</artifactId>
		<scope>test</scope>
</dependency>
  • 引用spring boot parent服务
  • 引用sprint-boot-xxxx-web-services定义为web程序
  • 引用sprint-boot-xxx-test进行测试
2. 生成程序启动入口，配置控制器信息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
@SpringBootApplication
public class HellowordDemoApplication {
	@RequestMapping("/")
	@ResponseBody
	String home() {
		return "Hello World!";
	}
	public static void main(String[] args) {
		SpringApplication.run(HellowordDemoApplication.class, args);
	}
}
3. application.properties 文件，在这里面可以设置程序参数，如启动端口,数据库连接等.
1
2
 ## 服务访问端口
server.port=9527

3. 程序效果

Alt text

spring boot 集成了嵌入式的tomcat,jetty 容器，非常方便引用即可用。降低了程序开发的难度。将所有用到的框架，梳理封装成了一个整体。不在需要程序员再像原来一样一点点在maven文件里引用，排包冲突等，是一套完整的解决方案。如果说以前实现程序开发，像是自己组装零件的话，那spring boot 就像是直接给了你n套半成品机械，只需要编写业务即可。好处很多，但是坏处还是有两点的，第一，出现时间太短，在国内也不是那么的流行。第二，封装程度过高，出现底层问题，靠自己排查会非常的困难。但相信经过时间的积累，它会越发的稳定，且被大众接受，成为一套标准。

展开全文 >>

flume源码分析1

入口类Application解析

1.1 类图

Alt text

. 1. components 对象，存储了解析flume-conf.properties 的实现方法

. 2. supervisor 对象,生命周期管理者，管理所有继承与LifecycleAware接口的所有服务的启停及状态监控

. 3. materializedConfiguration对象，存储了flume传入的配置信息。启动source,channel,sink时负责传入参数

. 4. monitorServer 对象，用与启动http监控或则Ganglia监控。监控服务端信息

. 5. main(String[] args) 主程序方法，命令参数校验，加载配置文件，加载监控等。

. 6. start() 启动配置文件加载类

. 7. stop() 停止monitorServer 监控和supervisor管理服务.

. 8. main() 观察者方法，用于启动source,channel,sink服务

1.3 mian方法

. 使用了 commons-cli.jar 中的Options，校验入参的完整性。及解析参数信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
main(){....

      Options options = new Options();
      //具有参数
      Option option = new Option("n", "name", true, "the name of this agent");
      //强制参数，必须得有
      option.setRequired(true);
      options.addOption(option);

      option = new Option("f", "conf-file", true,
          "specify a config file (required if -z missing)");
      option.setRequired(false);
      options.addOption(option);
      ...
      ...
      option = new Option("h", "help", false, "display help text");
      options.addOption(option);
      
      CommandLineParser parser = new GnuParser();
      CommandLine commandLine = parser.parse(options, args);
      
      String agentName = commandLine.getOptionValue('n');
      File configurationFile = new File(commandLine.getOptionValue('f'));

Option(String opt, String longOpt, boolean hasArg, String description)
参数名称，完整名称，是否接受参数，参数描述。 比如命令n ，具有参数且必须传入参数.如果步传入这个参数，则直接抛出异常。

命令h ，不具有参数，只提供帮助信息打印

根据配置文件解析程序

. 参数模拟：-f /Users/playcrab/dev/java_code/ws3/flume/conf/flume-conf.properties -n pagent

flume-conf.properties内容 

1
2
3
4
5
pagent.sources = file_source
pagent.channels = file_channel
pagent.sinks = kafka_sink

......

加载配置文件解析方法

1
2
3
4
5
6
7
8
9
main(){....
          EventBus eventBus = new EventBus(agentName + "-event-bus");
          PollingPropertiesFileConfigurationProvider configurationProvider =
            new PollingPropertiesFileConfigurationProvider(
              agentName, configurationFile, eventBus, 30);
          components.add(configurationProvider);
          application = new Application(components);
          //订阅事件
          eventBus.register(application);

观察者方法

1
2
3
4
5
@Subscribe
  public synchronized void handleConfigurationEvent(MaterializedConfiguration conf) {
    stopAllComponents();
    startAllComponents(conf);
  }

被观察者 PollingPropertiesFileConfigurationProvider，中的FileWatcherRunnable线程

1
2
run()...
     eventBus.post(getConfiguration());

. 使用监听模式，guava 的EventBus，被观察者通知观察者是否已解析完配置文件。如果解析完成，观察者运行handleConfigurationEvent()方法，启动所有的channel,source,sink

PollingPropertiesFileConfigurationProvider 配置文件解析线程

2.1 类图

Alt text

    1. 解析flume配置文件，启动线程定期检测配置文件更新情况，并且重新进行加载。
    1. 注册到生命周期管理，实现线程管理。

2.2 start 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Override
public void start() {
  LOGGER.info("Configuration provider starting");

  Preconditions.checkState(file != null,
      "The parameter file must not be null");
  //周期任务调度器,启动单线程定时执行任务
  executorService = Executors.newSingleThreadScheduledExecutor(
          new ThreadFactoryBuilder().setNameFormat("conf-file-poller-%d")
              .build());

  FileWatcherRunnable fileWatcherRunnable =
      new FileWatcherRunnable(file, counterGroup);
  // */30 * * * * 秒调度一次,timeunit 时间转化器
  executorService.scheduleWithFixedDelay(fileWatcherRunnable, 0, interval,
      TimeUnit.SECONDS);

  lifecycleState = LifecycleState.START;

  LOGGER.debug("Configuration provider started");
}

重写了接口LifecycleAware的start方法，方法内单独启动一个线程，30秒执行一次。调用FileWatcherRunnable，对配置文件进行解析。

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
public class FileWatcherRunnable implements Runnable {

  private final File file;
  private final CounterGroup counterGroup;

  private long lastChange;

  public FileWatcherRunnable(File file, CounterGroup counterGroup) {
    super();
    this.file = file;
    this.counterGroup = counterGroup;
    this.lastChange = 0L;
  }

  @Override
  public void run() {
    LOGGER.debug("Checking file:{} for changes", file);
    //检查次数
    counterGroup.incrementAndGet("file.checks");

    long lastModified = file.lastModified();

    if (lastModified > lastChange) {
      LOGGER.info("Reloading configuration file:{}", file);
      //加载次数
      counterGroup.incrementAndGet("file.loads");

      lastChange = lastModified;

      try {
      	//提交消息
        eventBus.post(getConfiguration());
      } catch (Exception e) {
        LOGGER.error("Failed to load configuration data. Exception follows.",
            e);
      } catch (NoClassDefFoundError e) {
        LOGGER.error("Failed to start agent because dependencies were not " +
            "found in classpath. Error follows.", e);
      } catch (Throwable t) {
        // caught because the caller does not handle or log Throwables
        LOGGER.error("Unhandled error", t);
      }
    }
  }
}

这个才是配置文件处理的真实线程，当文件被更新。则重新解析配置文件，并且告诉观察者 Application的 handleConfigurationEvent()方法。重新加载source,channel,sink。

2.3 AbstractConfigurationProvider

是PollingPropertiesFileConfigurationProvider 的抽象父类，实现了getConfiguration()方法，解析配置文件。

1
2
3
4
5
protected abstract FlumeConfiguration getFlumeConfiguration();
public MaterializedConfiguration getConfiguration() {
	MaterializedConfiguration conf = new SimpleMaterializedConfiguration();
	FlumeConfiguration fconfig = getFlumeConfiguration();
	....

展开全文 >>

guava 观察者模式(监听模式)

1. 传统观察者模式

这是设计模式中的一种，举一个不恰当的例子，一个老师，盯着3个学生做作业。每当一个孩子做完作业，老师都要去及时的批改作业。

这其中就存在二个对象：

1，观察者对象：就是老师

2，被观察对象(受查者)：被监听的三个倒霉孩子

实现：

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
import java.util.Observable;
import java.util.Observer;

public class WatcherCls {
	class TeaCherWatcher implements Observer {

		@Override
		public void update(Observable o, Object arg) {
			StudentWatched b = (StudentWatched) o;
			System.out.println(String.format("%s,写完了作业,老师要进行批改了...", b.call()));
		}
	}

	class StudentWatched extends Observable {
		private String name;

		public StudentWatched(String name) {
			this.name = name;
		}

		public String call() {
			return name;
		}

		public void work(boolean isOk) {
			if (isOk) {//如果作业写完，通知观察者老师
				setChanged();
				notifyObservers();
			}
		}
	}

	public static void main(String[] args) {
		WatcherCls ws = new WatcherCls();
		StudentWatched nick = ws.new StudentWatched("nick");// 受查者
		StudentWatched devid = ws.new StudentWatched("devid");// 受查者
		StudentWatched ken = ws.new StudentWatched("ken");// 受查者

		TeaCherWatcher teatherWatcher = ws.new TeaCherWatcher();// 观察者
		nick.addObserver(teatherWatcher);
		devid.addObserver(teatherWatcher);
		ken.addObserver(teatherWatcher);
		nick.work(true);
		devid.work(false);
		ken.work(true);

	}
}

  • 观察者(theather)重写Observer接口，被观察者(student) 继承Observable。

  • 触发notifyObservers 父类方法通知观察者。setChanged方法，表明内容有变化，不调用则不会通知。为开关方法。

2. 监听模式

设计模式中的一种，还是已上面那个例子为准。老师盯着三个孩子写作业，每当孩子完成作业都要通知老师来检查。

定义：事件源经过事件的封装传给监听器，当事件源触发事件后，监听器接收到事件对象可以回调事件的方法

我们拆分一下对象，存在3个对象

1，事件源：老师

2，监听器：老师监听作业完成情况

3， 事件：学生

###实现：

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
import java.util.Enumeration;
import java.util.Vector;

public class EventCls2 {
	public class TeacherSource {
		private Vector<TeacherListener> repository = new Vector<TeacherListener>();// 监听自己的监听器队列

		public TeacherSource() {
		}

		public void addListener(TeacherListener dl) {
			repository.addElement(dl);
		}

		public void notifyEvent() {// 通知所有的监听器
			Enumeration<TeacherListener> em = repository.elements();
			while (em.hasMoreElements()) {
				TeacherListener dl = (TeacherListener) em.nextElement();
				dl.work(new StudentEvent(this));
			}
		}
	}

	public class StudentEvent extends java.util.EventObject {
		private static final long serialVersionUID = 1L;
		boolean isOk = false;

		public StudentEvent(Object source) {
			super(source);		}

		public void say(String name, boolean isok) {
			String rs = isok ? name + "作业写完了。。" : name + "作业没写完。。";
			isOk = isok;
			System.out.println(rs);
		}
	}

	public interface TeacherListener extends java.util.EventListener {
		public void work(StudentEvent dm);
	}

	public static void main(String[] args) {
		EventCls2 ec = new EventCls2();
		TeacherSource stSource = ec.new TeacherSource();
		stSource.addListener(new TeacherListener() {
			String name = "nick";

			@Override
			public void work(StudentEvent dm) {
				dm.say(name, false);
			}
		});
		stSource.addListener(new TeacherListener() {
			String name = "devide";

			@Override
			public void work(StudentEvent dm) {
				dm.say(name, true);
			}
		});
		stSource.addListener(new TeacherListener() {
			String name = "ken";

			@Override
			public void work(StudentEvent dm) {
				dm.say(name, false);
			}
		});
		stSource.notifyEvent();// 触发事件、通知监听器

	}
}

监听器模式是观察者模式的另一种形态，同样基于事件驱动模型。监听器模式更加灵活，可以对不同事件作出相应。但也是付出了系统的复杂性作为代价的，因为我们要为每一个事件源定制一个监听器以及事件，这会增加系统的负担。 监听器可以实现多个时间的监控，及抽象扩展性非常的灵活。

##guava 中的观察者模式

1.简单方法

还是一样的定义，我们实现老师监控学生写作业。

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
import java.util.HashSet;
import java.util.Set;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;

public class EventCls3 {
	public class StudentEvent {
		String name;
		boolean isok;

		public StudentEvent(String name, boolean isok) {
			this.name = name;
			this.isok = isok;
		}

		public boolean getStatus() {
			return isok;
		}

		public String getName() {
			return name;
		}
	}

	public class TeacherListener {
		Set<String> rs;

		public TeacherListener() {
			rs = new HashSet<String>();
		}

		@Subscribe
		public void listen(StudentEvent event) {
			String say = event.getStatus() ? event.getName() + ", 完成了作业.." : event.getName() + "没完成作业";
			rs.add(say);
		}

		public Set<String> rs() {
			return rs;
		}
	}

	public static void main(String[] args) {
		EventCls3 ec = new EventCls3();

		TeacherListener listener = ec.new TeacherListener();//观察者
		EventBus eventBus = new EventBus();
		//注册监听器
		eventBus.register(listener);

		StudentEvent nickEvent = ec.new StudentEvent("nick", true);//受查者
		StudentEvent devidEvent = ec.new StudentEvent("devid", false);//受查者
		StudentEvent kenEvent = ec.new StudentEvent("ken", true);//受查者
		//提交监听事件,并触发监听方法
		eventBus.post(nickEvent);
		eventBus.post(devidEvent);
		eventBus.post(kenEvent);
		//遍历返回信息
		for (String say : listener.rs()) {
			System.out.println(say);
		}
	}
}

1，相对于jdk中提供的观察者方法，guava中不需要继承接口，没有代码侵入性。干净且利于理解。
2，guava 使用注解@Subscribe可以自定义监听器的监听方法，可以同时实现多个事件的监听。

2, Event的继承

如果Listener A监听Event A, 而Event A有一个子类Event B, 此时Listener A将同时接收Event A和B消息，实例如下：

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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import java.util.HashSet;
import java.util.Set;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;

public class EventCls3 {

	public class ParentEvent {
		protected String name;
		protected boolean isok;

		public ParentEvent(String name, boolean isok) {
			this.name = name;
			this.isok = isok;
		}

		public boolean getStatus() {
			return isok;
		}

		public String getName() {
			return name;
		}

		public String parentName() {
			return name;
		}
	}

	public class StudentEvent extends ParentEvent {

		public StudentEvent(String name, boolean isok) {
			super(name, isok);
		}

		@Override
		public String getName() {
			return super.getName() + "的儿子";
		}

	}

	public class TeacherListener {
		Set<String> rs;

		public TeacherListener() {
			rs = new HashSet<String>();
		}

		@Subscribe
		public void listenStudent(StudentEvent event) {
			String say = event.getStatus() ? event.getName() + ", 完成了作业.." : event.getName() + " ,没完成作业";
			rs.add(say);
		}

		@Subscribe
		public void listenParent(StudentEvent event) {
			String say = event.getStatus() ? event.parentName() + ", 你儿子没完成作业.." : event.parentName() + " ,你儿子没完成作业";
			rs.add(say);
		}

		public Set<String> rs() {
			return rs;
		}
	}

	public class TeacherListenerParent {

		Set<String> rs;

		public TeacherListenerParent() {
			rs = new HashSet<String>();
		}

		@Subscribe
		public void listenParent(StudentEvent event) {
			String say = event.getStatus() ? event.parentName() + ", 你儿子完成了作业" : event.parentName() + " ,你儿子没完成作业，来学校一趟";
			rs.add(say);
		}

		public Set<String> rs() {
			return rs;
		}
	}

	public static void main(String[] args) {
		EventCls3 ec = new EventCls3();

		TeacherListener listener = ec.new TeacherListener();
//		TeacherListenerParent listenerParent = ec.new TeacherListenerParent();

		EventBus eventBus = new EventBus("test");
		eventBus.register(listener);
//		eventBus.register(listenerParent);

		StudentEvent nickEvent = ec.new StudentEvent("nick", true);
		StudentEvent devidEvent = ec.new StudentEvent("devid", false);
		StudentEvent kenEvent = ec.new StudentEvent("ken", true);
		eventBus.post(nickEvent);
		eventBus.post(devidEvent);
		eventBus.post(kenEvent);
		for (String say : listener.rs()) {
			System.out.println(say);
		}
//		System.out.println("send parent.......");
//		for (String say : listenerParent.rs()) {
//			System.out.println(say);
//		}
	}
}

就写到这里，基本操作就是这样。这个设计模式在很多地方都可以用到，想到了一句话叫暗中观察。。。

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

健康快乐