基于Openresty+Lua+Kafka对日志进行实时的实时采集

　　基于Openresty+Lua+Kafka对日志进行实时的实时采集

　　简介

　　在大量的数据采集场景中，Flume是一个高性能的采集日志工具，相信大家都知道。很多人认为Flume是一个组件，大多数人能想到的就是Flume和Kafka的结合，用于日志采集。该解决方案具有许多优点，例如高性能、高吞吐量和数据可靠性。但是如果我们在日志上要求实时采集，这显然不是一个好的解决方案。原因如下：

　　目前，Flume 可以支持对目录中的数据文件进行实时监控。一旦某个目录下的文件采集完成，就会标上完成标记。如果以后有数据进入这个文件，Flume 将不会被检测到。

　　所以，我们大多使用这个方案来定时采集，只要生成一个新的数据目录，我们就会采集这个目录下的数据文件。

　　那么这篇文章就给大家介绍采集基于Openresty+Lua+Kafka的实时日志。

　　要求

　　很多时候，我们需要对用户的埋点数据进行实时采集，然后利用这些数据对用户的行为做一些实时分析。所以，第一步当然是解决如何进行实时数据采集。

　　我们这里使用的方案是Openresty+Lua+Kafka。

　　原理介绍

　　那么什么是 Openresty？这是官方引用：

　　OpenResty 是一个基于 Nginx 和 Lua 的高性能网络平台。它集成了大量复杂的 Lua 库、第三方模块及其大部分依赖项。用于方便地构建可处理超高并发和高扩展性的动态Web应用、Web服务和动态网关。

　　OpenResty 集成了各种精心设计的 Nginx 模块，有效地将 Nginx 变成了一个强大的通用 Web 应用程序平台。这样，Web 开发人员和系统工程师可以使用 Lu 脚本语言来调动 Nginx 支持的各种 C 和 Lua 模块，快速构建一个能够支持 10K 甚至超过 1000 个单机并发连接的高性能 Web 应用系统。

　　OpenResty 的目标是让你的 web 服务直接运行在 Nginx 服务内部，充分利用 Nginx 的非阻塞 I/O 模型，不仅可以用于 HTTP 客户端请求，甚至可以用于 MySQL、PostgreSQL 等远程后端、Memcached 和 Redis 等具有一致的高性能响应。

　　简单的说就是通过客户端的请求，通过Nginx将用户的数据传递到我们指定的地方（Kafka）（本文指的是用户的行为日志），而为了实现这个需求，我们使用了一个Lua脚本，因为Openresty封装了各种Lua模块，其中之一就是分装Kafka模块，我们只需要写一个简单的脚本就可以通过Nginx将用户的数据转发到Kafka，供后续消费数据使用。

　　这里有一个架构图供大家理解：

　　

　　这里简单总结一下使用Openresty+Lua+Kafka的优势：

　　1.支持多种业务数据。针对不同的业务数据，只需要配置不同的Lua脚本，就可以将不同的业务数据发送到Kafka中的不同主题。

　　2.实时采集用户触发的埋点数据

　　3.高度可靠的集群。由于Openresty基于Nginx，其集群具有非常高的性能和稳定性。

　　4.高并发，相比tomcat、apache等web服务器，Nginx的并发量远高于其他两个。一般情况下，处理几万并发并不难。

　　接下来我们来做一些实际的练习。

　　安装 Openresty

　　本示例使用独立部署形式。单机部署成功后，集群作为单机搭建，只是在不同的机器上执行相同的步骤。

　　注：本实验基于centos7.0操作系统

　　1.下载 Openresty 依赖：

　　yum install readline-devel pcre-devel openssl-devel gcc

　　2.编译安装Openresty：

　　#1.安装openresty:

mkdir /opt/software

mkdir /opt/module

cd /opt/software/ # 安装文件所在目录

wget https://openresty.org/download/openresty-1.9.7.4.tar.gz

tar -xzf openresty-1.9.7.4.tar.gz -C /opt/module/

cd /opt/module/openresty-1.9.7.4

#2.配置:

# 指定目录为/opt/openresty,默认在/usr/local。

./configure --prefix=/opt/openresty \

--with-luajit \

--without-http_redis2_module \

--with-http_iconv_module

make

make install

　　3.安装 lua-resty-kafka

　　因为我们需要通过nginx+lua脚本将数据转发到Kafka，所以在编写lua脚本时需要用到lua模块中的一些Kafka依赖。

　　#下载lua-resty-kafka:

cd /opt/software/

wget https://github.com/doujiang24/lua-resty-kafka/archive/master.zip

unzip master.zip -d /opt/module/

#拷贝kafka相关依赖脚本到openresty

cp -rf /opt/module/lua-resty-kafka-master/lib/resty/kafka/ /opt/openresty/lualib/resty/

　　注意：由于大家对Kafka比较熟悉，这里就不介绍它的安装了。

　　安装Openresty后，目录结构如下：

　　drwxr-xr-x 2 root root 4096 Mar 24 14:26 bin

drwxr-xr-x 6 root root 4096 Mar 24 14:26 luajit

drwxr-xr-x 7 root root 4096 Mar 24 14:29 lualib

drwxr-xr-x 12 root root 4096 Mar 24 14:40 nginx

　　4.配置文件

　　编辑/opt/openresty/nginx/conf/nginx.conf

　　user nginx; #Linux的用户

worker_processes auto;

worker_rlimit_nofile 100000;

#error_log logs/error.log;

#error_log logs/error.log notice;

#error_log logs/error.log info;

#pid logs/nginx.pid;

events {

worker_connections 102400;

multi_accept on;

use epoll;

}

http {

include mime.types;

default_type application/octet-stream;

log_format main '$remote_addr - $remote_user [$time_local] "$request" '

'$status $body_bytes_sent "$http_referer" '

'"$http_user_agent" "$http_x_forwarded_for"';

access_log /var/log/nginx/access.log main;

resolver 8.8.8.8;

#resolver 127.0.0.1 valid=3600s;

sendfile on;

keepalive_timeout 65;

underscores_in_headers on;

gzip on;

include /opt/openresty/nginx/conf/conf.d/common.conf; #common.conf这个文件名字可自定义

}

　　编辑/opt/openresty/nginx/conf/conf.d/common.conf

　　##api

lua_package_path "/opt/openresty/lualib/resty/kafka/?.lua;;";

lua_package_cpath "/opt/openresty/lualib/?.so;;";

lua_shared_dict ngx_cache 128m; # cache

lua_shared_dict cache_lock 100k; # lock for cache

server {

listen 8887; #*敏*感*词*端口

server_name 192.168.3.215; #埋点日志的ip地址或域名，多个域名之间用空格分开

root html; #root指令用于指定虚拟主机的网页根目录，这个目录可以是相对路径，也可以是绝对路径。

lua_need_request_body on; #打开获取消息体的开关，以便能获取到消息体

access_log /var/log/nginx/message.access.log main;

error_log /var/log/nginx/message.error.log notice;

location = /lzp/message {

lua_code_cache on;

charset utf-8;

default_type 'application/json';

content_by_lua_file "/opt/openresty/nginx/lua/testMessage_kafka.lua";#引用的lua脚本

}

}

　　编辑/opt/openresty/nginx/lua/testMessage_kafka.lua

　　#创建目录mkdir /opt/openresty/nginx/lua/

vim /opt/openresty/nginx/lua/testMessage_kafka.lua

#编辑内存如下：

　　-- require需要resty.kafka.producer的lua脚本，没有会报错

local producer = require("resty.kafka.producer")

-- kafka的集群信息，单机也是可以的

local broker_list = {

{host = "192.168.3.215", port = 9092},

}

-- 定义最终kafka接受到的数据是怎样的json格式

local log_json = {}

--增加read_body之后即可获取到消息体，默认情况下可能会是nil

log_json["body"] = ngx.req.read_body()

log_json["body_data"] = ngx.req.get_body_data()

-- 定义kafka同步生产者，也可设置为异步 async

-- -- 注意！！！当设置为异步时，在测试环境需要修改batch_num,默认是200条，若大不到200条kafka端接受不到消息

-- -- encode()将log_json日志转换为字符串

-- -- 发送日志消息,send配套之第一个参数topic:

-- -- 发送日志消息,send配套之第二个参数key,用于kafka路由控制:

-- -- key为nill(空)时，一段时间向同一partition写入数据

-- -- 指定key，按照key的hash写入到对应的partition

-- -- batch_num修改为1方便测试

local bp = producer:new(broker_list, { producer_type = "async",batch_num = 1 })

-- local bp = producer:new(broker_list)

local cjson = require("cjson.safe")

local sendMsg = cjson.encode(log_json)

local ok, err = bp:send("testMessage",nil, sendMsg)

if not ok then

ngx.log(ngx.ERR, 'kafka send err:', err)

elseif ok then

ngx.say("the message send successful")

else

ngx.say("未知错误")

end

　　5.开始服务运行：

　　useradd nginx #创建用户

passwd nginx #设置密码

#设置openresty的所有者nginx

chown -R nginx:nginx /opt/openresty/

#启动服务

cd /opt/openresty/nginx/sbin

./nginx -c /opt/openresty/nginx/conf/nginx.conf

查看服务：

ps -aux | grep nginx

nginx 2351 0.0 0.1 231052 46444 ? S Mar30 0:33 nginx: worker process

nginx 2352 0.0 0.1 233396 48540 ? S Mar30 0:35 nginx: worker process

nginx 2353 0.0 0.1 233396 48536 ? S Mar30 0:33 nginx: worker process

nginx 2354 0.0 0.1 232224 47464 ? S Mar30 0:34 nginx: worker process

nginx 2355 0.0 0.1 231052 46404 ? S Mar30 0:33 nginx: worker process

nginx 2356 0.0 0.1 232224 47460 ? S Mar30 0:34 nginx: worker process

nginx 2357 0.0 0.1 231052 46404 ? S Mar30 0:34 nginx: worker process

nginx 2358 0.0 0.1 232224 47484 ? S Mar30 0:34 nginx: worker process

root 7009 0.0 0.0 185492 2516 ? Ss Mar24 0:00 nginx: master process ./nginx -c /opt/openresty/nginx/conf/nginx.conf

查看端口：

netstat -anput | grep 8887

tcp 0 0 0.0.0.0:8887 0.0.0.0:* LISTEN 2351/nginx: worke

　　看到上面的过程，证明服务运行正常

　　6.使用postman发送post请求做一个简单的测试看Kafka能否接受数据

　　

　　7.kafka 消费数据：

　　kafka-console-consumer --bootstrap-server 192.168.3.215:9092 --topic testMessage --from-beginning

　　如果数据被消费，则配置成功。如果没有调整，可以查看/var/log/nginx/message.access.log和/var/log/nginx/message.error.log相关的错误日志进行调整

　　总结

　　使用Openresty+Lua+Kafka可以将用户的埋点数据采集实时传输到kafka集群，而且Openresty是基于Nginx的，Nginx可以处理几万并发，所以即使用户的数据随着时间的激增，这种架构可以轻松应对，而不会导致集群崩溃。另一方面，如果数据太多导致集群过载，我们也可以随时多加一台机器，非常方便。

　　另外一个小扩展：如果有很多业务数据需要发送到不同的topic，我们不需要写多个脚本，但是可以联系后端添加一个json格式的字段，取值该字段的 是主题的名称。我们只需要写一个通用脚本，解析Json数据，取出主题名即可。

　　

　　关于寻找教程网络