国庆节前有段时间，新浪的“图床”一直不大稳定，因为新浪开启了防盗链，果然免费的永远是最贵的啊。为了不影响使用，我非常粗暴地禁止了浏览器发送Referer，然后我就发现了一件尴尬的事情，“不蒜子”统计服务无法使用了。这是一件用脚后跟想都能想明白的事情，我禁止了浏览器发送Referer，而“不蒜子”正好使用Referer来识别每个页面，所以，这是一个再明显不过的因为需求变更而引入的Bug。这个世界最离谱的事情，就是大家都认为程序员是一本“十万个为什么”，每次一出问题就找到程序员这里。其实，程序员是再普通不过的芸芸众生里的一员，人们喜欢听/看到自己愿意去听/看到的事物，而程序员同样喜欢解决自己想去解决的问题。所以，今天的话题是关于如何设计一个PV/UV统计系统。OK，Let’s Hacking Begin。
 
PV/UV的概念
 
首先，我们从两个最基本的概念PV和UV开始说起。我们都知道，互联网产品的核心就是流量，前期通过免费的产品吸引目标客户的目的，在积累了一定用户流量以后，再通过广告等增值服务实现盈利，这可以说是互联网产品的典型商业模式啦。而在这个过程中，为了对一个产品的流量进行科学地分析，就产生了譬如访客数(UV)、浏览量(PV)、访问次数(VV)等等的概念，这些概念通常作为衡量流量多少的指标。除此以外，我们还有类似日活跃用户(DAU)、月活跃用户(MAU)等等这种衡量服务用户粘性的指标，以及平均访问深度、平均访问时间、跳出率等等这种衡量流量质量优劣的指标。如果各位和我一样都写博客的话，对这些概念应该都不会感到陌生࿰

 
 
转载请注明出处
 
 
背景
 
 
 
在互联网公司中，每个项目都需要数据统计、分析，便于项目组利用详细数据研究项目的整体情况，进行下一步的调整。在数据统计中，UV统计是最常见的，也是最普遍的。有的场景要求实时性很高，有点场景要求准确性很高，有的场景比较在意计算过程中的内存。不同的场景使用不同的算法，下面我们从0到1简单介绍下UV统计领域。
 
 
什么是UV统计
 
假设我们的场景是商家这边上架一系列水果，然后需要统计出一共上架几种水果。具体如下所示：
 
 
针对这个问题，我们想到的最简单的方式就是利用STL中的set处理。
 
SET
 
上架一个水果的时候，也同时在set中插入。最后需要统计的时候，直接计算set中一共有几个水果即可。具体如下所示：
 
 这种方式准确率是绝对准确的，但是这种方式耗费的内存是很大的。
 假设每个水果需要 K 字节，那么如果有 M 个水果，一共需要 K * M 字节。那么我们能不能缩小这里的内存呢？
 稍微损失一点准确率换取内存？具体见下面HashMap的方式
 
HASHMAP
 
这种算法在上架一个水果的时候，只需要在特定的位置置1即可，而不需要存储这个位置上究竟是何种水果。然后在统计的时候，只需要统计hashmap里面有多少个1即可。
 具体如下所示：
 
 那么如果有M个水果，这里其实只需要 M / 8 字节，相比set的方式内存直接缩小到1/8。当然Hash肯定会有冲突的，所以这里肯定有一定准确率的损失。
 但是如果涉及到海量数据的UV统计，这里的内存还是很大的。
 能否用上统计学进一步缩小内存呢？具体见下面的Linear Count的方式。
 
Linear Count
 
这种算法在上架一个水果的时候，完全跟hashmap一致，在相应位置置1。
 然后在统计的时候，利用统计学的方式，根据hashmap中零的个数给出一个估算值。具体如下所示：
 
 
假设M为哈希桶长度，那么每次上架水果，每个桶被选中的概率为：
 $$\frac{1}{M}$$
 然后在上架N个元素后，某个桶为0的概率为：
 $$(1-\frac{1}{M}{)}^N$$
 所以在上架n个元素后，哈希桶中零的个数期望为：
 $$ZeroNum=\sum _{i=1}^M(1-\frac{1}{M}{)}^N=M(1-\frac{1}{M}{)}^N=M((1+\frac{1}{-M}{)}^{-M}{)}^{-\frac{N}{M}})\approx M{e}^{-\frac{N}{M}}$$
 所以最终：
 $$N=UV=-Mln(\frac{ZeroNum}{M})$$
 
所以Linear Count算法中，只需统计下hashmap中零的个数，然后代入上式即可。
 这种算法在N很小的时候，准确率是很高的，但是N很大的时候，它的准确率急剧下降。
 针对海量数据的情况，LogLog Count的算法更加鲁棒
 
LogLog Count
 
这种算法跟上面几种都不同，上架水果的时候，在相应桶里面记录的是二进制数后面最长的连续零个数。然后统计的时候，利用统计学的方式，根据存储中最长连续后缀零个数，得出一个估计值。具体如下所示：
 
 
它的原理如下：
 
 
 
这里如果只使用一个桶来估计的话，它的误差是很大，需要用分桶平均的方式来减少它的误差。
 
分桶平均
 
既然这里利用了分桶来减少误差，那么这里统计的时候就必须合起来，这里有4种方式：
 
算术平均：$$UV=\frac{{\sum }_{j=1}^{m}U{V}_j}{m}$$
几何平均：$$UV=\sqrt[m]{U{V}_1...UVm}$$
调和平均：$$UV=\frac{m}{{\sum }_{j=1}^{m}U{V}_j^{-1}}$$
中位数：
      
       
        
         
          U
         
         
          V
         
         
          =
         
         
          m
         
         
          e
         
         
          d
         
         
          i
         
         
          a
         
         
          m
         
         
          {
         
         
          U
         
         
          
           V
          
          
           1
          
         
         
          ,
         
         
          .
         
         
          .
         
         
          .
         
         
          ,
         
         
          U
         
         
          
           V
          
          
           m
          
         
         
          }
         
        
        
         UV=mediam \{UV_1,...,UV_m\}
        
       
      UV=mediam{UV1​,...,UVm​}
 
LogLog Count利用的是算术平均的方式，所以最终估计值为：
 $$UV=2^{\frac{{\sum }_{j=1}^{m}U{V}_j}{m}}$$
 
这种算法对于基数大的情况下准确率挺高的，但是基数小的情况下准确率很低。
 
HyperLogLog Count
 
这种算法跟LogLog Count 类似，有个区别点就是它在求均值的时候利用了调和平均数，而不是算术平均数。这里最终估计值为：
 $$UV=mm(\sum _{j=1}^m{2}^{-M_j}{)}^{-1}$$
 然后它还引入了分段误差修正。
 
误差修正
 
 
具体可以看我github上的代码：HyperLogLog
 
总结
 
准确率
内存
耗时
Set
绝对准确
K * M
O(Mlog(M))
HashMap
很高
M/8
O(M)
Linear Count
基数小高，基数大低
M/8
O(M/8)
LogLog Count
基数小低，基数大高
HyperLogLog Count
高

 
文章目录
 
1 概述
2 基本使用
 


 本文学习在 Redis 中通过 
HyperLogLog 实现 UV 统计。
 
 
1 概述
 
一般我们评估一个网站的访问量，有几个主要的参数：PV（Page View）网页的浏览量、UV（User View）访问的用户量。有很多第三方工具可以统计，如 cnzz，友盟等。如果自己实现的话，PV 比较简单，可以直接通过 Redis 计数器实现。但是 UV 就不一样，UV 涉及到去重的问题。
 
常规思路：我们首先需要在前端给每一个用户生成一个唯一 id，无论是登录用户还是未登录用户都需要，这个 id 伴随着请求一起到达后端，在后端我们可以通过 Set 数据结构中的 sadd 命令来存储这个 id，最后通过 scard 统计集合大小，进而得出 UV 数据。
 
按上述思路，如果是千万级别的 UV，需要的存储空间就非常惊人，用 Set 就不是很合适了。一般来说像 UV 统计这种，不需要特别精确，比如 800W 和 803W 的 UV，其实差别不大。因此，我们可以使用 HyperLogLog 来高效的实现。
 
2 基本使用
 
Redis 中提供的 HyperLogLog 就是专门用来解决上述问题的，HyperLogLog 提供了一套不怎么精确但是够用的去重方案，会有误差，官方给出的误差数据是 0.81%，这对于 UV 的统计够用了。
 
HyperLogLog 主要提供了以下命令：
 
pfadd：用来添加记录，类似于 sadd ，添加过程中，重复的记录会自动去重。
pfcount：则用来统计数据。
pfmerge：合并多个统计结果，在合并的过程中，会自动去重多个集合中重复的元素。
 
数据量少的时候看不出来误差，我们在 Java 中多添加几个元素：
 
public class HyperLogLogTest {
    @Test
    public void testHyperLogLog() {
        Redis redis = new Redis();
        redis.execute(jedis -> {
            for (int i = 0; i < 1000; i++) {
                // 重复加入数据，理论值上总数为 1001
                jedis.pfadd("uv", "u" + i, "u" + (i + 1));
            }
            long uv = jedis.pfcount("uv");
            System.out.println(uv);
        });
    }
}
 
理论值上总数为 1001，实际打印出来 994，有误差，但是在可以接受的范围内。
 
 
Redis 教程合集（微信左下方阅读全文可直达）。
本文示例代码：https://github.com/cxy35/samples/tree/master/redis/redis-jedis
 
 
扫码关注微信公众号 程序员35 ，获取最新技术干货，畅聊 #程序员的35，35的程序员# 。独立站点：https://cxy35.com
 

go pv uv 统计
 
更多干货
 
 
分布式实战（干货）
 
 
spring cloud 实战（干货）
 
 
mybatis 实战（干货）
 
 
spring boot 实战（干货）
 
 
React 入门实战（干货）
 
 
构建中小型互联网企业架构（干货）
 
 
python 学习持续更新
 
 
ElasticSearch 笔记
 
 
kafka storm 实战 (干货)
 
 
scala 学习持续更新
 
 
RPC
 
 
深度学习
 
 
GO 语言 持续更新
 
 
 
概述
 
当你网站访问量过大时 第3方统计系统就不愿意了，这时就需要自己来设计个流量统计了。
 
 
流量分析系统是用来分析NGINX的日志数据的，NGINX的日志数据是用户网站访问后生成的，所以我们需要搭建一个网站，来产生一些用户行为数据。然后通过javascript快速上报到打点服务器，然后打点服务器再上报到nginx，nginx会生成用户行为日志，再通过我们用GO开发的流量统计系统分析处理后中，再由前台数据展示出来。
 
在一个比较大的WEB网站，几百万几千万流量的上报打点请求，如果我们用一个低性能的WEBSERVER做打点服务器，会非常卡，影响效率，所以最好用好一些的服务器。 我们的打点服务器需要用到nginx的一个模块，ngx_http_empty_gif_module,这个模块，这个module可以生成1x1像素大小的透明的gif图片。非常小的
 
打点js 参照：
 
 
[js获取客户端time,cookie,url,ip,refer,user_agent信息](https://blog.csdn.net/qq_27384769/article/details/81009884)
 
 
[nginx空白图片 访问打点](https://blog.csdn.net/qq_27384769/article/details/81012625)
 
第三方库
 
go get github.com/sirupsen/logrus
 
go get github.com/mgutz/str
 
go get github.com/mediocregopher/radix.v2
 
代码
 
package main

import (
	"flag"
	"github.com/sirupsen/logrus"
	"time"
	"os"
	"bufio"
	"io"
	"strings"
	"github.com/mgutz/str"
	"net/url"
	"crypto/md5"
	"encoding/hex"
	"github.com/mediocregopher/radix.v2/pool"
	"strconv"
)

const HANDLE_DIG = " /dig?"
const HANDLE_MOVIE = "/movie/"
const HANDLE_LIST = "/list/"
const HANDLE_HTML = ".html"


type cmdParams struct {
	logFilePath string
	routineNum int
}
type digData struct{
	time   string
	url    string
	refer  string
	ua        string
}
type urlData struct {
	data   digData
	uid    string
	unode  urlNode
}
type urlNode struct {
	unType     string // 详情页 或者 列表页 或者 首页
	unRid  int       // Resource ID 资源ID
	unUrl  string // 当前这个页面的url
	unTime  string // 当前访问这个页面的时间
}
type storageBlock struct {
	counterType       string
	storageModel   string
	unode        urlNode
}

var log = logrus.New()

func init() {
	log.Out = os.Stdout
	log.SetLevel( logrus.DebugLevel )
}

func main() {
	// 获取参数
	logFilePath := flag.String( "logFilePath", "/Users/pangee/Public/nginx/logs/dig.log", "log file path" )
	routineNum := flag.Int( "routineNum", 5, "consumer numble by goroutine" )
	l := flag.String( "l", "/tmp/log", "this programe runtime log target file path" )
	flag.Parse()

	params := cmdParams{ *logFilePath, *routineNum }

	// 打日志
	logFd, err := os.OpenFile( *l, os.O_CREATE|os.O_WRONLY, 0644 )
	if err == nil {
		log.Out = logFd
		defer logFd.Close()
	}
	log.Infof( "Exec start." )
	log.Infof( "Params: logFilePath=%s, routineNum=%d", params.logFilePath, params.routineNum )

	// 初始化一些channel，用于数据传递
	var logChannel = make(chan string, 3*params.routineNum)
	var pvChannel = make(chan urlData, params.routineNum)
	var uvChannel = make(chan urlData, params.routineNum)
	var storageChannel = make(chan storageBlock, params.routineNum)

	// Redis Pool
	redisPool, err := pool.New( "tcp", "localhost:6379", 2*params.routineNum );
	if err != nil{
		log.Fatalln( "Redis pool created failed." )
		panic(err)
	} else {
		go func(){
			for{
				redisPool.Cmd( "PING" )
				time.Sleep( 3*time.Second )
			}
		}()
	}


	// 日志消费者
	go readFileLinebyLine( params, logChannel )

	// 创建一组日志处理
	for i:=0; i<params.routineNum; i++ {
		go logConsumer( logChannel, pvChannel, uvChannel )
	}

	// 创建PV UV 统计器
	go pvCounter( pvChannel, storageChannel )
	go uvCounter( uvChannel, storageChannel, redisPool )
	// 可扩展的 xxxCounter

	// 创建 存储器
	go dataStorage( storageChannel, redisPool )

	time.Sleep( 1000*time.Second )
}

// HBase 劣势：列簇需要声明清楚
func dataStorage( storageChannel chan storageBlock, redisPool *pool.Pool) {
	for block := range storageChannel {
		prefix := block.counterType + "_"

		// 逐层添加，加洋葱皮的过程
		// 维度： 天-小时-分钟
		// 层级： 定级-大分类-小分类-终极页面
		// 存储模型： Redis  SortedSet
		setKeys := []string{
			prefix+"day_"+getTime(block.unode.unTime, "day"),
			prefix+"hour_"+getTime(block.unode.unTime, "hour"),
			prefix+"min_"+getTime(block.unode.unTime, "min"),
			prefix+block.unode.unType+"_day_"+getTime(block.unode.unTime, "day"),
			prefix+block.unode.unType+"_hour_"+getTime(block.unode.unTime, "hour"),
			prefix+block.unode.unType+"_min_"+getTime(block.unode.unTime, "min"),
		}

		rowId := block.unode.unRid

		for _,key := range setKeys {
			ret, err := redisPool.Cmd( block.storageModel, key, 1, rowId ).Int()
			if ret<=0 || err!=nil {
				log.Errorln( "DataStorage redis storage error.", block.storageModel, key, rowId )
			}
		}
	}
}

func pvCounter( pvChannel chan urlData, storageChannel chan storageBlock ) {
	for data := range pvChannel {
		sItem := storageBlock{ "pv", "ZINCRBY", data.unode }
		storageChannel <- sItem
	}
}
func uvCounter( uvChannel chan urlData, storageChannel chan storageBlock, redisPool *pool.Pool ) {
	for data := range uvChannel {
		//HyperLoglog redis
		hyperLogLogKey := "uv_hpll_"+getTime(data.data.time, "day")
		ret, err := redisPool.Cmd( "PFADD", hyperLogLogKey, data.uid, "EX", 86400 ).Int()
		if err!=nil {
			log.Warningln( "UvCounter check redis hyperloglog failed, ", err )
		}
		if ret!=1 {
			continue
		}

		sItem := storageBlock{ "uv", "ZINCRBY", data.unode }
		storageChannel <- sItem
	}
}

func logConsumer( logChannel chan string, pvChannel, uvChannel chan urlData ) error {
	for logStr := range logChannel {
		// 切割日志字符串，扣出打点上报的数据
		data := cutLogFetchData( logStr )

		// uid
		// 说明： 课程中模拟生成uid， md5(refer+ua)
		hasher := md5.New()
		hasher.Write( []byte( data.refer+data.ua ) )
		uid := hex.EncodeToString( hasher.Sum(nil) )

		// 很多解析的工作都可以放到这里完成
		// ...
		// ...

		uData := urlData{ data, uid, formatUrl( data.url, data.time ) }

		pvChannel <- uData
		uvChannel <- uData
	}
	return nil
}
func cutLogFetchData( logStr string ) digData {
	logStr = strings.TrimSpace( logStr )
	pos1 := str.IndexOf( logStr,  HANDLE_DIG, 0)
	if pos1==-1 {
		return digData{}
	}
	pos1 += len( HANDLE_DIG )
	pos2 := str.IndexOf( logStr, " HTTP/", pos1 )
	d := str.Substr( logStr, pos1, pos2-pos1 )

	urlInfo, err := url.Parse( "http://localhost/?"+d )
	if err != nil {
		return digData{}
	}
	data := urlInfo.Query()
	return digData{
		data.Get("time"),
		data.Get("refer"),
		data.Get("url"),
		data.Get("ua"),
	}
}
func readFileLinebyLine( params cmdParams, logChannel chan string ) error {
	fd, err := os.Open( params.logFilePath )
	if err != nil {
		log.Warningf( "ReadFileLinebyLine can't open file:%s", params.logFilePath )
		return err
	}
	defer fd.Close()

	count := 0
	bufferRead := bufio.NewReader( fd )
	for {
		line, err := bufferRead.ReadString( '\n' )
		logChannel <- line
		count++

		if count%(1000*params.routineNum) == 0 {
			log.Infof( "ReadFileLinebyLine line: %d", count )
		}
		if err != nil {
			if err == io.EOF {
				time.Sleep( 3*time.Second )
				log.Infof( "ReadFileLinebyLine wait, raedline:%d", count )
			} else {
				log.Warningf( "ReadFileLinebyLine read log error" )
			}
		}
	}
	return nil
}


func formatUrl( url, t string ) urlNode{
	// 一定从量大的着手,  详情页>列表页≥首页
	pos1 := str.IndexOf( url, HANDLE_MOVIE, 0)
	if pos1!=-1 {
		pos1 += len( HANDLE_MOVIE )
		pos2 := str.IndexOf( url, HANDLE_HTML, 0 )
		idStr := str.Substr( url , pos1, pos2-pos1 )
		id, _ := strconv.Atoi( idStr )
		return urlNode{ "movie", id, url, t }
	} else {
		pos1 = str.IndexOf( url, HANDLE_LIST, 0 )
		if pos1!=-1 {
			pos1 += len( HANDLE_LIST )
			pos2 := str.IndexOf( url, HANDLE_HTML, 0 )
			idStr := str.Substr( url , pos1, pos2-pos1 )
			id, _ := strconv.Atoi( idStr )
			return urlNode{ "list", id, url, t }
		} else {
			return urlNode{ "home", 1, url, t}
		} // 如果页面url有很多种，就不断在这里扩展
	}
}

func getTime( logTime, timeType string ) string {
	var item string
	switch timeType {
	case "day":
		item = "2018-01-02"
		break
	case "hour":
		item = "2018-01-02 15"
		break
	case "min":
		item = "2018-01-02 15:04"
		break
	}
	t, _ := time.Parse( item, time.Now().Format(item) )
	return strconv.FormatInt( t.Unix(), 10 )
}