KAFKA在统计系统中的应用

Kafka

kafka是LinkedIn开源的一款分布式的发布-订阅消息系统，它具有：

1. 通过O(1)的磁盘结构持久化存储消息，即使TB级的数据也能保持长期稳定；

2. 高吞吐率：即使非常普通的硬件，kafka也能支持每秒数十万的消息；

3. 支持通过kafka服务器和消费集群来分区消息；

4. 支持Hadoop并行加载；

设计流程

通过集成kafka和log4j在各个需要采集日志的系统进行日志采集，日志统一发送到kafka的消息队列。再通过定时运行kafka-hadoop的作业，将数据从KAFKA同步到hadoop的HDFS中。hadoop的各类map-reduce作业可以对数据进行统计，存储到db或缓存中。

安装

首先安装kafka，下载kafka到本地，然后执行：

> tar xzf kafka-<VERSION>.tgz
> cd kafka-<VERSION>
> ./sbt update
> ./sbt package

启动zookeeper

> bin/zookeeper-server-start.sh config/zookeeper.properties
[2010-11-21 23:45:02,335] INFO Reading configuration from: config/zookeeper.properties...

启动kafka

> bin/kafka-server-start.sh config/server.properties
jkreps-mn-2:kafka-trunk jkreps$ bin/kafka-server-start.sh config/server.properties 
[2010-11-21 23:51:39,608] INFO starting log cleaner every 60000 ms (kafka.log.LogManager)
[2010-11-21 23:51:39,628] INFO connecting to ZK: localhost:2181 (kafka.server.KafkaZooKeeper)...

发送一些消息

> bin/kafka-console-producer.sh --zookeeper localhost:2181 --topic test 
This is a message
This is another message

查看消息

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
This is a message
This is another message

至此，kafka已经启动成功。

安装hadoop，下载hadoop0.20.2，按照hadoop的操作手册安装，此处省略安装过程。

配置log4j和kafka-log

#配置KAFKA
#Hostname表示应用的名称
#Topic表示打印日志生成的目录名
log4j.appender.KAFKA=com.comp.kafka.AsyncKafkaAppender
log4j.appender.KAFKA.topic=TOPIC
log4j.appender.KAFKA.bufferSize = 10
log4j.appender.KAFKA.brokerList=0:kafka-server-ip:9092
log4j.appender.KAFKA.serializerClass=kafka.serializer.StringEncoder
log4j.appender.KAFKA.hostname=APP-NAME
log4j.appender.KAFKA.layout=org.apache.log4j.PatternLayout
log4j.appender.KAFKA.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n
log4j.logger.com.comp= DEBUG, KAFKA

配置中，配置了kafka的ip和端口，并设置了topic和序列化类和pattern。
AsyncKafkaAppender是继承了log4j的AsyncAppender，是log4j异步发送日志的模式，当log达到bufferSize的大小时，会统一由log4j异步执行打印操作。代码如下：

public class AsyncKafkaAppender extends AsyncAppender {
    private java.lang.String topic;
    private java.lang.String serializerClass;
    private java.lang.String zkConnect;
    private java.lang.String brokerList;
    private java.lang.String hostname;

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getSerializerClass() {
        return serializerClass;
    }

    public void setSerializerClass(String serializerClass) {
        this.serializerClass = serializerClass;
    }

    public String getZkConnect() {
        return zkConnect;
    }

    public void setZkConnect(String zkConnect) {
        this.zkConnect = zkConnect;
    }

    public String getBrokerList() {
        return brokerList;
    }

    public void setBrokerList(String brokerList) {
        this.brokerList = brokerList;
    }

    public String getHostname() {
        return hostname;
    }

    public void setHostname(String hostname) {
        this.hostname = hostname;
    }
    @Override
    public void activateOptions() {
        super.activateOptions();
        synchronized (this) {
            KafkaLog4jAppender kafka = new KafkaLog4jAppender();
            kafka.setLayout(getLayout());
            kafka.setHostname(getHostname());
            kafka.setBrokerList(getBrokerList());
            kafka.setSerializerClass(getSerializerClass());
            kafka.setZkConnect(getZkConnect());
            kafka.setTopic(getTopic());
            kafka.activateOptions();
            addAppender(kafka);
        }
    }

    @Override
    public boolean requiresLayout() {
        return true;
    }
}

这时我们就可以撰写一个测试代码，测试log是否已经发送到了kafka。

KAFKA消息发送到Hadoop

在kafka的contrib目录的hadoop-consumer中有一系列的文件，包括脚本，jar，配置文件等。我们可以直接使用这个目录下的脚本进行定时数据同步。

具体的执行方式：https://github.com/kafka-dev/kafka/tree/master/contrib/hadoop-consumer

需要先修改test目录下的test.properties：

# kafka的topic名称
kafka.etl.topic=TOPIC

# hdfs location of jars
hdfs.default.classpath.dir=/tmp/kafka/lib

# number of test events to be generated
event.count=1000

# hadoop id and group
hadoop.job.ugi=kafka,hadoop

# kafka server uri
kafka.server.uri=tcp://localhost:9092

# hdfs location of input directory
input=hdfs://localhost:9000/tmp/kafka/data

# hdfs location of output directory
output=hdfs://localhost:9000/tmp/kafka/output

# limit the number of events to be fetched;
# value -1 means no limitation
kafka.request.limit=-1

# kafka parameters
client.buffer.size=1048576
client.so.timeout=60000

修改topic名称以及对应hadoop的input，output目录。
首先生成offset，执行：

./run-class.sh kafka.etl.impl.DataGenerator test/test.properties

然后拷贝依赖的jar文件：

./copy-jars.sh ${hdfs.default.classpath.dir}

最后执行hadoop任务：

./run-class.sh kafka.etl.impl.SimpleKafkaETLJob test/test.properties

执行完成后，运行hadoop脚本查看是否已经在output目录生成数据：

bin/hadoop fs -cat /tmp/kafka/output/part-00000 | wc -l

至此，从日志从各个系统已经成功收集到hadoop的HDFS文件系统中。
现在只需要撰写一些map-reduce作业，就可以利用hadoop进行数据统计了。

KAFKA消息清除

# The minimum age of a log file to be eligible for deletion
log.retention.hours=168
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.file.size=536870912

在配置中，如果一个log file距离上一次写入时间达到168小时，也就是一周，会自动清除这个日志文件；日志文件的上限大小是536870912，超过这个大小会创建新的日志文件。

MAC下清除DNS缓存的方法

想清除被污染的DNS，在MAC下使用：
sudo killall -HUP mDNSResponder [Lion&MountainLion]
sudo dscacheutil -flushcache 10.6及以前的用户

在Ubuntu12.04上安装scribe

        Scribe是Facebook开源的一个日志收集系统，它能够从各种数据源收集日志，并存储到一个中央存储系统。它的原理图如下：

安装它不是一个很容易的事情，首先它依赖于Thrift，而Thrift又有很多的依赖。

前置依赖

Thrift版本<0.4.0; boost版本1.4.5；gcc，g++版本4.4；

安装Thrift

下载地址

首先安装前置依赖：

sudo apt-get install libevent-dev automake libtool flex bison pkg-config g++ libssl-dev 

需要java依赖的还需要安装java：

sudo apt-get install openjdk-6-jdk ant ivy

首先进入到thrift目录，安装：

sudo ./configure --prefix=/opt/thrift
sudo make & make install

然后进入thrift目录下的contrib/fb303，

sudo ./bootstrap.sh
sudo ./configure --prefix=/opt/fb303 --with-thriftpath=/opt/thrift
sudo make & make install

如果一切顺利的话，thrift就安装完了。

安装scribe

下载地址

首先，需要保证你的系统的boost库是1.4.5的，去boost网站下载1.4.5版本的源码，下载地址

解压缩，进入到boost目录，然后执行

sudo ./bootstrap.sh
sudo ./bjam
sudo ./bjam install

好，此时可以切到目录/usr/local/lib grep一下boost，看看版本是否是1.4.5
然后修改g++编译器的版本，

g++ --version

先查看本地g++版本，如果>4.4，则执行：

sudo apt-get remove g++
sudo apt-get install g++-4.4
cd /usr/bin
sudo ln -s g++-4.4 g++

此时再执行：

g++ --version

发现已经是4.4.3版本。然后解压缩scribe源码包，进入目录scribe-master
执行：

sudo ./bootstrap.sh
sudo ./configure --with-thriftpath=/opt/thrift --with-fb303path=/opt/fb303
sudo make & make install

安装完成执行scribed，发现已经安装完成。

intellijIDEA常用快捷键备忘录(MAC)

⌘+/ 注销当前行
⌘+X 剪切当前行或当前选中代码
⌘+D 复制当前行
⌘+N 查找类
ctrl+N 在pom中查找依赖
alt+enter 查找类依赖，类似eclipse中的alt+/
⌘+alt+L 格式化代码
⌘+alt+O 优化导入的类包代码
alt+shift+up/down 代码上移或下移一行
⌘+shift+up/down 光标处的代码逻辑块整体上移或下移
fn+F3 选中单词的情况下查找下一个相同的单词
ctrl+shift+/ 注释/取消注释 选中的代码块
⌘+alt+B 查看接口或方法的实现，如果只有一个实现，直接打开实现类或方法。
shift+enter 开始新的一行
ctrl+J 查看方法说明
⌘+alt+V 从粘贴历史里选择一个进行粘贴
⌘+Z 取消当前一个操作
⌘+shift+Z 恢复取消当前一个操作，⌘+Z的逆向
⌘+R 查找替换
ctrl+shfit+F 全项目搜索
⌘+delete 删除当前行

os10.5以上的系统需要在preference中选择keymap，选择Mac OSX 10.5以上
MAC OS X 快捷键大全

Twitter Finagle-Scala语言的RPC框架

简介

Finagle是Twitter的一个开源scala框架，支持Java语言和Scala语言，支持的协议包括Apache Thrift， Kestrel， Memcached， http， redis， mysql等。由于国内的scala应用程度不高，而Twitter已经逐渐发展成为了Scala的重度使用用户。Finagle是基于NIO和netty的，主体的结构如下图：

本文的目的是实践Finagle框架的thrift协议，并在底层通过JPA，Hibernate与mysql进行交互，同时利用Spring framework管理Bean。

scrooge

scrooge也是twitter开源的一个项目，项目是用scala编写的一个thrift代码生成器。我们直接下载zip包使用scrooge脚本生成代码即可。

假设我们的thrift IDL代码如下：

struct User{
        1:i32 id,
        2:string name,
        3:i32 createdAt,
        4:i32 updated
}

service UserService{
        User getUser(1:i32 id),
        User saveUser(1:User user),
        User updateUser(1:User user)
}

使用下面的命令生成scala代码：

> scrooge -finagle demo.thrift

会生成一个thrift目录，scala代码就在目录下。

maven

通过eclipse新建一个scala项目，选择scala-archetype-simple，创建一个带maven支持的scala项目。

记得enable maven支持的同时，选中scala菜单的添加scala特性，否则不会编译。

需要调整pom中的scala版本和本地scala版本一致。

Finagle

拷贝scrooge生成的代码到工程中，并新建ThriftServer和ThriftClient两个scalaObject。

代码ThriftServer.scala

object ThriftServer {
  def main(args: Array[String]) {
   
    val protocol = new TBinaryProtocol.Factory()
    val ctx = new ClassPathXmlApplicationContext("applicationContext.xml")
    val userService = ctx.getBean("userService").asInstanceOf[UserService.FutureIface]
    val server = new UserService.FinagledService(userService, protocol)
    val address = new InetSocketAddress("localhost", 9000)
    var builder = ServerBuilder()
      .codec(ThriftServerFramedCodec())
      .name("UserService")
      .bindTo(address)
      .build(server)
  }
}

代码ThriftClient.scala：

object ThriftClient {
  def main(args: Array[String]) {
    val service = ClientBuilder()
      .hosts(new InetSocketAddress("localhost", 9000))
      .codec(ThriftClientFramedCodec())
      .hostConnectionLimit(100)
      .build()
    val client = new UserService.FinagledClient(service)
    val user = client.getUser(10)
    println("getUser-->"+user.apply())
    val user1 = client.saveUser(user.apply());
    println("save db-->"+user1.apply())
  }
}

ThriftServer中使用了spring管理了userService。
新建一个实现类，并调用了封装的dao层代码：

class UserServiceImpl extends FutureIface {
  
  @BeanProperty
  var dao:Dao = null
  
  def getUser(`id`: Int): Future[User] = {
    val currentTime = (System.currentTimeMillis() / 1000).asInstanceOf[Int];
    var user = if (id == 0) null else User.apply(id, "wenfeng", currentTime, currentTime)
    Future.value(user)
  }
  def saveUser(`user`: User): Future[User] = {
    var u = new com.scala.core.User(user._1, user._2, user._3, user._4)
    try{
     val id = dao.save(u)
     u.setId(id.asInstanceOf[Long]);
    }catch{
      case ex:DalException => ex.printStackTrace()
    }
    println("save user-->"+user)
    val res = User.apply(u.getId().asInstanceOf[Int], 
        u.getName(), (u.getCreatedAt() / 1000).asInstanceOf[Int], 
        (u.getUpdated() / 1000).asInstanceOf[Int])
    Future.value(res)
  }
  def updateUser(`user`: User): Future[User] = Future.value(user)
}

Spring的配置文件:

<beans xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context" xmlns:sca="http://www.springframework.org/schema/sca" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://www.springframework.org/schema/beans" xsi:schemalocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd
            http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
            http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd
            http://www.springframework.org/schema/sca http://www.osoa.org/xmlns/sca/1.0/spring-sca.xsd">
 <context:component-scan base-package="com.scala">
 <bean class="com.scala.dao.impl.DaoImpl" id="dao">
 <bean class="com.scala.service.impl.UserServiceImpl" id="userService">
  <property name="dao" ref="dao">
 </property></bean>
</bean></context:component-scan></beans>

新建一个JPA的User类，使用scala的@BeanProperty 生成getter和setters

@Entity
@Table(name="user")
class User extends java.io.Serializable{
 @Id
 @GeneratedValue(strategy = GenerationType.AUTO)
 @BeanProperty 
 var id:Long = _
 
 @Column(name="name")
 @BeanProperty 
 var name:String = _
 
 @Column(name="created_at")
 @BeanProperty 
 var createdAt:Long = _
 
 @Column(name="updated")
 @BeanProperty 
 var updated:Long = _
 
 def this(id:Long, name:String, createdAt:Long, update:Long) = {
   this()
   this.id = id
   this.name = name
   this.createdAt = createdAt
   this.updated = updated
 }
 override def toString() = "User["+id+","+name+","+createdAt+","+updated+"]"

测试插入一条用户数据，查看log和DB后，发现数据已经成功写入DB。

Memcached一致性Hash实测

使用的客户端

spy memcached

环境

先启动4个memcached服务，然后向这4个memcached实例写入1w个数据，测试数据分布情况；再增加一个实例，在5个实例的情况下，读取1w条记录，计算实际命中率

代码

启用一致性hash很容易，代码如下：

List servers = new ArrayList();
servers.add(new InetSocketAddress("192.168.12.167", 33001));
servers.add(new InetSocketAddress("192.168.12.167", 33002));
servers.add(new InetSocketAddress("192.168.12.167", 33003));
servers.add(new InetSocketAddress("192.168.12.167", 33004));
servers.add(new InetSocketAddress("192.168.12.167", 33005));
memCachedClient = new MemcachedClient(new KetamaConnectionFactory(), servers);

写入的分布比例

server1 2304
server2 2603
server3 2513
server4 2588

加入一个实例后的命中率

81.38%

GOF设计模式-命令模式

command pattern

UML图如下：

命令模式分为四个角色，分别是Invoker，Command抽象接口，Command实现类和Receiver。

用一个Java与模式上的例子解释，玉皇大帝发圣旨，让太白金星招安孙悟空。这里玉皇大帝就是Client，圣旨有一个抽象接口，可以称作圣旨模板，圣旨则是实现类。而太白金星则是任务直接执行Invoker，孙悟空是任务接受者Receiver，真正的执行代码在孙悟空这里。

用Scala代码实现的命令模式如下：
首先是client代码：

object Client extends App {
 val receiver = new Receiver
 val command = new ConcreteCommand(receiver)
 val invoker = new Invoker(command)
 invoker.action
}

然后是Command代码：

trait Command {
 def execute
}

ConcerteCommand:

class ConcreteCommand(re:Receiver) extends Command{
 val receiver = re
 override def execute = receiver.action
}

Receiver:

class Receiver {

  def action = {
    println("Action has been taken.")
  }
}

Invoker:

class Invoker(com:Command) {
 val command = com
 
 def action = command.execute
}

执行结果：

Action has been taken.

命令模式的优点：

1. 新命令很容易加入到系统
2. 允许接受者一方决定是否要否决请求
3. 能较容易的设计一个命令队列
4. 也可以很容易实现对请求的undo和redo。
5. 在需要的情况下，可以很容易的将命令计入日志。
6. 将请求的操作对象和具体执行的操作对象分开；
7. 可以合成命令。

命令模式的使用场景

1. 作为回调，先定义好一个函数，在以后调用此函数；

2. 需要在不同的时间请求，将请求排队

3. 系统需要支持命令的撤销；

4. 如果系统需要记录命令日志，方便在系统崩溃后通过日志调用execute方法恢复系统在崩溃前的数据更新；

5. 一个系统需要支持交易；