介绍
这是一个基于Netty框架二次封装的高性能Http接口服务，增加了对http请求路由的功能，并简化了操作，目的是将接收到的日志经过简单处理后快速推送到kafka ，服务于易企秀数据埋点业务，春节期间日处理10亿+
特点：简单、高效 （在最普通的机器环境压测QPS最高可以达到3w/s）
项目地址：https://github.com/yanchaoguo/log-server
设计
image.png
依赖
- Netty4
- logback kafka appender
- ipip
快速启动
1、编译
git clone https://github.com/yanchaoguo/log-server.git
mvn assembly:assembly
2、配置
将编译好的工程放到 /data/work/log_server，目录结构如下：
----log_server
|
----bin #依赖的jar包
|
----classes #编译后生成的classes目录
|
----bin #启动脚本
|
----logs #日志文件
cd /data/work/log_server/bin
编辑 ls.sh 脚本，配置端口号和kafka地址
port=9001
kafka=hadoop006:9092,hadoop007:9092,hadoop008:9092
3、启动
./ls.sh start
快速开发
1、目录说明
image.png
2、在action目录下新增业务处理类，比如创建FastPushAction并实现Action接口的doAction方法；该类不对日志进行处理，没有太多业务逻辑，负责将接收到的日志发送到kafka ，可以作为demo参考，具体实现如下
/**
* /fast_push为该业务逻辑的请求路径 ，如http://localhost:port/log-server/fast_push
*/
@Route(value = "/fast_push")
public class FastPushAction implements Action {
private static final Logger logger = LoggerFactory.getLogger(PushLogAction.class);
@Override
public void doAction(Request request, Response response) {
String logs = request.getContent(); //从body中获取日志内容
//从参数中获取 logger 名称，为空则取默认值
String loger = Utils.isNulDefault(request.getParam("loger"),LogConfigManager.trashTopic);
int responseStatus = CodeManager.RESPONSE_CODE_NORMAL;
//如果body中没有内容 尝试从url参数中获取
if (StrUtil.isEmpty(logs) && request.getParams().size()>0) {
logs = Utils.toJson(request.getParams());
}
if (StrUtil.isEmpty(logs)) {
response.setContent("<h2>not find log content</h2>");
responseStatus = CodeManager.RESPONSE_CODE_NOT_CONTENT;
logger.warn("not find log content");
}else {
//将日志推送到kafka，其中 logger 的名称要和logback.xml中的配置一致
KafkaLoggerFactory.getLogger(loger).info(logs);
}
// 设置返回信息：
response.setStatus(responseStatus);
response.setHeader("Access-Control-Allow-Origin", "*");
response.setHeader(CookieHeaderNames.EXPIRES , -1);
// 返回：
response.send();
}
}
3、配置logback.xml
<?xml version="1.0" encoding="UTF-8" ?>
<configuration>
<!--trash-->
<logger name="trash" level="INFO" additivity="false">
<appender-ref ref="trash_kafka"/>
</logger>
<appender name="trash_kafka" class="com.github.danielwegener.logback.kafka.KafkaAppender">
<encoder class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder">
<layout class="ch.qos.logback.classic.PatternLayout">
<pattern>%msg</pattern>
</layout>
<charset>UTF-8</charset>
</encoder>
<topic>trash</topic>
<keyingStrategy class="com.github.danielwegener.logback.kafka.keying.RoundRobinKeyingStrategy" />
<deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
<producerConfig>bootstrap.servers=${kafka.servers}
</producerConfig>
<producerConfig>linger.ms=1000</producerConfig>
<producerConfig>compression.type=none</producerConfig>
<producerConfig>acks=0</producerConfig>
</appender>
</configuration>
- producer相关配置
# 生产者会尝试将业务发送到相同的 Partition 的消息合包发送到 Broker，batch.size 设置合包的大小上限。默认为 16KB。batch.size 设太小会导致吞吐下降，设太大会导致内存使用过多。
batch.size=1638400
# Kafka producer 的 ack 有 3 种机制，分别说明如下：
# -1 或 all：Broker 在 leader 收到数据并同步给所有 ISR 中的 follower 后，才应答给 Producer 继续发送下一条（批）消息。 这种配置提供了最高的数据可靠性，只要有一个已同步的副本存活就不会有消息丢失。注意：这种配置不能确保所有的副本读写入该数据才返回，可以配合 Topic 级别参数 min.insync.replicas 使用。
# 0：生产者不等待来自 broker 同步完成的确认，继续发送下一条（批）消息。这种配置生产性能最高，但数据可靠性最低（当服务器故障时可能会有数据丢失，如果 leader 已死但是 producer 不知情，则 broker 收不到消息）
# 1：生产者在 leader 已成功收到的数据并得到确认后再发送下一条（批）消息。这种配置是在生产吞吐和数据可靠性之间的权衡（如果leader已死但是尚未复制，则消息可能丢失）
# 用户不显示配置时，默认值为1。用户根据自己的业务情况进行设置
acks=0
# 控制生产请求在 Broker 等待副本同步满足 acks 设置的条件所等待的最大时间
timeout.ms=10
# 请求发生错误时重试次数，失败重试最大程度保证消息不丢失
retries=0
# 配置生产者用来缓存消息等待发送到 Broker 的内存。用户要根据生产者所在进程的内存总大小调节
buffer.memory=335544320
# 当生产消息的速度比 Sender 线程发送到 Broker 速度快，导致 buffer.memory 配置的内存用完时会阻塞生产者 send 操作，该参数设置最大的阻塞时间
max.block.ms=0
# 设置消息延迟发送的时间，这样可以等待更多的消息组成 batch 发送。默认为0表示立即发送。当待发送的消息达到 batch.size 设置的大小时，不管是否达到 linger.ms 设置的时间，请求也会立即发送
linger.ms=1000
# 生产者能够发送的请求包大小上限，默认为1MB。在修改该值时注意不能超过 Broker 配置的包大小上限16MB
max.request.size=1048576
# 压缩格式配置，压缩比：LZ4 > Snappy ;吞吐量：LZ4 > Snappy
compression.type=[none, snappy, lz4]
# 客户端发送给 Broker 的请求的超时时间，不能小于 Broker 配置的 replica.lag.time.max.ms，目前该值为10000ms
request.timeout.ms=30000
# 客户端在每个连接上最多可发送的最大的未确认请求数，该参数大于1且 retries 大于0时可能导致数据乱序。 希望消息严格有序时，建议客户将该值设置1
max.in.flight.requests.per.connection=5
4、在 LogServer.java 中注册FastPushAction
public static void start() {
//注册action
ServerSetting.setAction(PushLogAction.class);
ServerSetting.setAction(FastPushAction.class);
try {
new LogServer().start(ServerSetting.getPort());
} catch (InterruptedException e) {
log.error("LoServer start error!", e);
}
}
性能
4核2G单实例
数据大小1k，压测命令如下：
ab -n2000000 -c1000 "http://****:9001/log-server/fast_push?debugMode=0&sdk=tracker-view.js&ver=1.1.1&d_i=20200226a78e43e2&url=https%3A%2F%2Fp.scene.eqh5.cn%2Fs%2F8vEWzYf7%3Fshare_level%3D69%26from_user%3D20200126cee71912%26from_id%3D063eeec3-b%26share_time%3D1583710918358%26userKey%3D158371091722647525%26from%3Dgroupmessage%26isappinstalled%3D0%26from%3Dgroupmessage%26isappinstalled%3D0&tit=2020%E3%80%8A%E6%88%91%E7%9A%84%E6%96%B0%E7%9B%B8%E5%86%8C%E3%80%8B&ref=&u_a=&bro=%E5%BE%AE%E4%BF%A1&os=Android&o_v=9&eng=Webkit&man=&mod=V1934A&sns=weixin-groupmessage&n_t=4g&s_i=v3x20200309fc93f6bb&c_i=3791794543062d9524811946c7d050c0"
压测结果 3w/s：
Concurrency Level: 1000
Time taken for tests: 66.556 seconds
Complete requests: 2000000
Failed requests: 0
Write errors: 0
Total transferred: 272000000 bytes
HTML transferred: 0 bytes
Requests per second: 30049.66 [#/sec] (mean)
Time per request: 33.278 [ms] (mean)
Time per request: 0.033 [ms] (mean, across all concurrent requests)
Transfer rate: 3990.97 [Kbytes/sec] received
image.png
