前言

使用golang实现一个简单的http/https代理服务器
支持http代理
支持https代理（https代理为直通模式，直接原样转发）

全部代码

package main

import (
    "flag"
    "io"
    "log"
    "net"
    "net/http"
    "strings"
)

type Pxy struct {
    Cfg Cfg
}

// 设置type
type Cfg struct {
    Addr        string   // 监听地址
    Port        string   // 监听端口
    IsAnonymous bool     // 高匿名模式
    Debug       bool     // 调试模式
}

func main() {

    // 参数
    faddr := flag.String("addr","0.0.0.0","监听地址，默认0.0.0.0")
    fprot := flag.String("port","8080","监听端口，默认8080")
    fanonymous :=  flag.Bool("anonymous",true,"高匿名，默认高匿名")
    fdebug :=  flag.Bool("debug",false,"调试模式显示更多信息，默认关闭")
    flag.Parse()

    cfg := &Cfg{}
    cfg.Addr = *faddr
    cfg.Port = *fprot
    cfg.IsAnonymous = *fanonymous
    cfg.Debug = *fdebug
    // fmt.Println(cfg)
    Run(cfg)
}

func Run(cfg *Cfg) {
    pxy := NewPxy()
    pxy.SetPxyCfg(cfg)
    log.Printf("HttpPxoy is runing on %s:%s \n", cfg.Addr, cfg.Port)
    // http.Handle("/", pxy)
    bindAddr := cfg.Addr + ":" + cfg.Port
    log.Fatalln(http.ListenAndServe(bindAddr, pxy))
}


// 实例化
func NewPxy() *Pxy {
    return &Pxy{
        Cfg: Cfg{
            Addr:        "",
            Port:        "8081",
            IsAnonymous: true,
            Debug:       false,
        },
    }
}

// 配置参数
func (p *Pxy) SetPxyCfg(cfg *Cfg) {
    if cfg.Addr != "" {
        p.Cfg.Addr = cfg.Addr
    }
    if cfg.Port != "" {
        p.Cfg.Port = cfg.Port
    }
    if cfg.IsAnonymous != p.Cfg.IsAnonymous {
        p.Cfg.IsAnonymous = cfg.IsAnonymous
    }
    if cfg.Debug != p.Cfg.Debug {
        p.Cfg.Debug = cfg.Debug
    }

}

// 运行代理服务
func (p *Pxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
    // debug
    if p.Cfg.Debug {
        log.Printf("Received request %s %s %s\n", req.Method, req.Host, req.RemoteAddr)
        // fmt.Println(req)
    }

    // http && https
    if req.Method != "CONNECT" {
        // 处理http
        p.HTTP(rw, req)
    } else {
        // 处理https
        // 直通模式不做任何中间处理
        p.HTTPS(rw, req)
    }

}

// http
func (p *Pxy) HTTP(rw http.ResponseWriter, req *http.Request) {

    transport := http.DefaultTransport

    // 新建一个请求outReq
    outReq := new(http.Request)
    // 复制客户端请求到outReq上
    *outReq = *req // 复制请求

    //  处理匿名代理
    if p.Cfg.IsAnonymous == false {
        if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil {
            if prior, ok := outReq.Header["X-Forwarded-For"]; ok {
                clientIP = strings.Join(prior, ", ") + ", " + clientIP
            }
            outReq.Header.Set("X-Forwarded-For", clientIP)
        }
    }

    // outReq请求放到传送上
    res, err := transport.RoundTrip(outReq)
    if err != nil {
        rw.WriteHeader(http.StatusBadGateway)
        rw.Write([]byte(err.Error()))
        return
    }

    // 回写http头
    for key, value := range res.Header {
        for _, v := range value {
            rw.Header().Add(key, v)
        }
    }
    // 回写状态码
    rw.WriteHeader(res.StatusCode)
    // 回写body
    io.Copy(rw, res.Body)
    res.Body.Close()
}


// https
func (p *Pxy) HTTPS(rw http.ResponseWriter, req *http.Request) {

    // 拿出host
    host := req.URL.Host
    hij, ok := rw.(http.Hijacker)
    if !ok {
        log.Printf("HTTP Server does not support hijacking")
    }

    client, _, err := hij.Hijack()
    if err != nil {
        return
    }

    // 连接远程
    server, err := net.Dial("tcp", host)
    if err != nil {
        return
    }
    client.Write([]byte("HTTP/1.0 200 Connection Established\r\n\r\n"))

    // 直通双向复制
    go io.Copy(server, client)
    go io.Copy(client, server)
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

package main

import (

"flag"

"io"

"log"

"net"

"net/http"

"strings"

)

type Pxy struct {

Cfg Cfg

}

// 设置type

type Cfg struct {

Addr string // 监听地址

Port string // 监听端口

IsAnonymous bool // 高匿名模式

Debug bool // 调试模式

}

func main() {

// 参数

faddr := flag.String("addr","0.0.0.0","监听地址，默认0.0.0.0")

fprot := flag.String("port","8080","监听端口，默认8080")

fanonymous := flag.Bool("anonymous",true,"高匿名，默认高匿名")

fdebug := flag.Bool("debug",false,"调试模式显示更多信息，默认关闭")

flag.Parse()

cfg := &Cfg{}

cfg.Addr = *faddr

cfg.Port = *fprot

cfg.IsAnonymous = *fanonymous

cfg.Debug = *fdebug

// fmt.Println(cfg)

Run(cfg)

}

func Run(cfg *Cfg) {

pxy := NewPxy()

pxy.SetPxyCfg(cfg)

log.Printf("HttpPxoy is runing on %s:%s \n", cfg.Addr, cfg.Port)

// http.Handle("/", pxy)

bindAddr := cfg.Addr + ":" + cfg.Port

log.Fatalln(http.ListenAndServe(bindAddr, pxy))

}

// 实例化

func NewPxy() *Pxy {

return &Pxy{

Cfg: Cfg{

Addr: "",

Port: "8081",

IsAnonymous: true,

Debug: false,

}

// 配置参数

func (p *Pxy) SetPxyCfg(cfg *Cfg) {

if cfg.Addr != "" {

p.Cfg.Addr = cfg.Addr

}

if cfg.Port != "" {

p.Cfg.Port = cfg.Port

}

if cfg.IsAnonymous != p.Cfg.IsAnonymous {

p.Cfg.IsAnonymous = cfg.IsAnonymous

}

if cfg.Debug != p.Cfg.Debug {

p.Cfg.Debug = cfg.Debug

}

// 运行代理服务

func (p *Pxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) {

// debug

if p.Cfg.Debug {

log.Printf("Received request %s %s %s\n", req.Method, req.Host, req.RemoteAddr)

// fmt.Println(req)

}

// http && https

if req.Method != "CONNECT" {

// 处理http

p.HTTP(rw, req)

} else {

// 处理https

// 直通模式不做任何中间处理

p.HTTPS(rw, req)

}

// http

func (p *Pxy) HTTP(rw http.ResponseWriter, req *http.Request) {

transport := http.DefaultTransport

// 新建一个请求outReq

outReq := new(http.Request)

// 复制客户端请求到outReq上

*outReq = *req // 复制请求

// 处理匿名代理

if p.Cfg.IsAnonymous == false {

if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil {

if prior, ok := outReq.Header["X-Forwarded-For"]; ok {

clientIP = strings.Join(prior, ", ") + ", " + clientIP

}

outReq.Header.Set("X-Forwarded-For", clientIP)

}

// outReq请求放到传送上

res, err := transport.RoundTrip(outReq)

if err != nil {

rw.WriteHeader(http.StatusBadGateway)

rw.Write([]byte(err.Error()))

return

}

// 回写http头

for key, value := range res.Header {

for _, v := range value {

rw.Header().Add(key, v)

}

// 回写状态码

rw.WriteHeader(res.StatusCode)

// 回写body

io.Copy(rw, res.Body)

res.Body.Close()

}

// https

func (p *Pxy) HTTPS(rw http.ResponseWriter, req *http.Request) {

// 拿出host

host := req.URL.Host

hij, ok := rw.(http.Hijacker)

if !ok {

log.Printf("HTTP Server does not support hijacking")

}

client, _, err := hij.Hijack()

if err != nil {

return

}

// 连接远程

server, err := net.Dial("tcp", host)

if err != nil {

return

}

client.Write([]byte("HTTP/1.0 200 Connection Established\r\n\r\n"))

// 直通双向复制

go io.Copy(server, client)

go io.Copy(client, server)

}

curl验证

# https
$  curl -x http://127.0.0.1:8080/ -I https://www.baidu.com/
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
  0   277    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0HTTP/1.0 200 Connection Established

HTTP/1.1 200 OK
Accept-Ranges: bytes
Cache-Control: private, no-cache, no-store, proxy-revalidate, no-transform
Connection: Keep-Alive
Content-Length: 277
Content-Type: text/html
Date: Tue, 09 Apr 2019 23:32:51 GMT
Etag: "575e1f7c-115"
Last-Modified: Mon, 13 Jun 2016 02:50:36 GMT
Pragma: no-cache
Server: bfe/1.0.8.18

# http
$  curl -x http://127.0.0.1:8080/ -I http://www.baidu.com/
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
  0   277    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0HTTP/1.1 200 OK
Accept-Ranges: bytes
Cache-Control: private, no-cache, no-store, proxy-revalidate, no-transform
Connection: Keep-Alive
Content-Length: 277
Content-Type: text/html
Date: Tue, 09 Apr 2019 23:35:33 GMT
Etag: "575e1f80-115"
Last-Modified: Mon, 13 Jun 2016 02:50:40 GMT
Pragma: no-cache
Server: bfe/1.0.8.18

# https

$ curl -x http://127.0.0.1:8080/ -I https://www.baidu.com/

% Total % Received % Xferd Average Speed Time Time Time Current

Dload Upload Total Spent Left Speed

0 277 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 0HTTP/1.0 200 Connection Established

HTTP/1.1 200 OK

Accept-Ranges: bytes

Cache-Control: private, no-cache, no-store, proxy-revalidate, no-transform

Connection: Keep-Alive

Content-Length: 277

Content-Type: text/html

Date: Tue, 09 Apr 2019 23:32:51 GMT

Etag: "575e1f7c-115"

Last-Modified: Mon, 13 Jun 2016 02:50:36 GMT

Pragma: no-cache

Server: bfe/1.0.8.18

# http

$ curl -x http://127.0.0.1:8080/ -I http://www.baidu.com/

% Total % Received % Xferd Average Speed Time Time Time Current

Dload Upload Total Spent Left Speed

0 277 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 0HTTP/1.1 200 OK

Accept-Ranges: bytes

Cache-Control: private, no-cache, no-store, proxy-revalidate, no-transform

Connection: Keep-Alive

Content-Length: 277

Content-Type: text/html

Date: Tue, 09 Apr 2019 23:35:33 GMT

Etag: "575e1f80-115"

Last-Modified: Mon, 13 Jun 2016 02:50:40 GMT

Pragma: no-cache

Server: bfe/1.0.8.18

参考资料

1. 前言

云端开发环境云端开发环境安装搭建云ide 远程调试开发
一直想建立一个云端开发环境
有以下尝试，以及目前用的最终方案
仅供参考

2. 常见云端IDE环境

2.1 SSH+VIM

此乃一苇渡江，vim高手首选方案
但是对于vim只停留在基础的配置文件修改水平，习惯了vscode/Jetbrains全家桶后就不太适应这种需要较高学习和配置曲线的方案了
有人说你可以用一些预制vim配置,实际用起来你还是需要去熟悉和调整的,要花大把时间的

2.2 AWS Cloud9

Cloud9官网
可以用AWS的云端服务,也可以私有部署
如果为了环境干净可以自己打docker镜像
哪有本地IDE工具功能强大啊

2.3 CodingIDE(Cloud Studio)

Cloud Studio官网
Cloud Studio – 开启云端开发模式WebIDE
可用云端服务,目前已经卖给腾讯云了,也可以私有部署（暂不清楚开源版还在更新没）,功能上弱于AWS Cloud9
哪有本地IDE工具功能强大啊

2.4 code-server

code-server官网
这货是vscode的服务端运行版本
使用基本和本地vscode保持一致,功能上强于 Cloud9和CodingIDE
当然自己也可以打docker镜像使用,不过打docker镜像有个弊端,就是开发环境变更问题
虽然开发环境几乎不会这么变,但是终归还是会有变动的,如果用docker的方式,每次变更都去同步修改Dockfile打镜像还是烦的

2.5 Linux Desktop + VNC + noVNC

选一个自己喜欢的Linux桌面发行版,装VNC server,再装一个noVNC让VNC在web浏览器中可以访问
Linux Desktop 我选的 Kali,他基于debian定制的, 主打安全并集成了较多安全方面的工具
图形化选的Xfce
和本地开发完全一致,想用什么IDE用什么IDE

3. 结论

对vim非常熟悉那么随便一个vps/vm虚拟机就开干
有条件开虚拟机的建议开vm虚拟机用Linux桌面版本
不能自己开虚拟机但是有vps/vm虚拟机用的可以装 code-server 用
喜欢让系统干干净净的处女座也不怕麻烦可以把code-server、Cloud9等打docker镜像用

没有最好,只有最适合,满足自己需求即可
个人最终选的方案5

4. 扩展

另外一个知识点，docker中是可以用桌面版的，这个大多数人都不知道
例如实验楼这个网站中的实验环境其实就是docker中运行的Linux桌面
搜索关键字“docker linux desktop”获取跟多的信息
要用docker中的Linux桌面版，可以采用VNC + noVNC的方式

前言

Caddy 是一款用golang编写的轻量级web server

老早就看到caddy web server,但是找不到去用的理由,因为nginx足够好
如果用docker打包应用,如果要实现跑php php-fpm应用且要单容器,用nginx就不太爽了
一般用php官方docker镜像作为base,然后在上面再装nginx,nginx复杂的编译环境非常麻烦
试用了下 caddy,他就没那么麻烦,还是用php官方docker镜像,然后只需要简单的几行就可以把caddy加进去
非常特别的一点是caddy自带php-fpm的进程启动
在配置文件中增加on startup php-fpm —nodaemonize 即可
容器的入口只需要启动caddy即可,caddy可以自动启动php-fpm进程
目前发现唯一的问题是caddy对个人免费,对公司和商业收费,价格还非常贵
下面是一些参考文件

Dockerfile 参考

FROM php:7.2-fpm-alpine

LABEL version="7.2" \
    description="php-fpm7.2 alpine" \
    maintainer="wwek<licoolgo@gmail.com>"

# 国内源
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories
# 系统基础包
RUN apk update \
    && apk upgrade \
    # && apk add s6 \
    && apk add bash \
    # && apk add openssl \
    # && apk add openssl-dev \
    && apk add ca-certificates \
    #&& apk add wget \
    && apk add curl \
    #&& apk add tcpdump \
    && apk add iputils \
    && apk add iproute2 \
    && apk add libc6-compat \
    && apk add zlib \
    && apk add -U tzdata \
    && apk add libpng \
    && apk add libpng-dev \
    #&& apk add libstdc++ \
    && rm -rf /var/cache/apk/*

# 系统设置(设置时区)
RUN rm -rf /etc/localtime \
    && ln -s /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \
    && echo "Asia/Shanghai" /etc/timezone

# --- PHP --- #
## PHP Core核心扩展
RUN docker-php-ext-install opcache mysqli pdo_mysql mbstring zip pcntl gd
## 三方扩展
## redis
RUN set -xe \
    #&& apk add --no-cache --update --virtual .phpize-deps $PHPIZE_DEPS \
    && apk add --update --virtual .phpize-deps $PHPIZE_DEPS \
    && pecl install -o -f redis  \
    #&& pecl install -o -f swoole  \
    && echo "extension=redis.so" > /usr/local/etc/php/conf.d/redis.ini \
    #&& echo "extension=swoole.so" > /usr/local/etc/php/conf.d/swoole.ini \
    && rm -rf /usr/share/php \
    && rm -rf /tmp/* \
    && apk del  .phpize-deps


# 增加 PHP 配置文件
ADD docker/www.conf /usr/local/etc/php-fpm.d/
ADD docker/php.ini /usr/local/etc/php

##  PHP设置
# 设置php运行权限为母鸡中 www的用户和组 1000:1000
# RUN addgroup -g 1000 -S www && \
#     adduser -u 1000 -S www -G www www

# 设置php-fpm运行账户和组为 www www
# RUN set -ex \
#     && cd /usr/local/etc \
#     && { \
#         echo 'listen.owner = www'; \
#         echo 'listen.group = www'; \
#         echo 'listen.mode = 0660'; \
#     } | tee php-fpm.d/www.conf

# -- Caddy -- #
## 安装 Caddy Web Server
#RUN curl -sSL -o- "https://caddyserver.com/download/linux/amd64?plugins=http.jwt,http.login,http.prometheus,http.realip,http.restic,http.expires&license=personal&telemetry=off" | tar -xvz -C /usr/bin/
RUN curl --silent --show-error --fail --location \
    --header "Accept: application/tar+gzip, application/x-gzip, application/octet-stream" -o - \
    "https://caddyserver.com/download/linux/amd64?plugins=http.jwt,http.login,http.prometheus,http.realip,http.restic,http.expires&license=personal&telemetry=off" \
    | tar --no-same-owner -C /usr/bin/ -xz caddy \
    && chmod 0755 /usr/bin/caddy \
    && /usr/bin/caddy -version

## 增加 Caddy 配置文件
ADD docker/Caddyfile /etc/Caddyfile


# -- APP -- #
## APP环境变量
ENV APPROOT="/data/wwwroot/www.iamle.com/" \
    PHP_MYSQL_HOSTNAME='' \
    PHP_MYSQL_DATABASE='' \
    PHP_MYSQL_USERNAME='' \
    PHP_MYSQL_PASSWORD='' \
    PHP_MYSQL_PORT='' \
    PHP_REDIS_HOST=''

## APP增加
RUN mkdir -p $APPROOT
COPY ./ $APPROOT


## APP配置
WORKDIR $APPROOT
# 巴拉巴拉 APP 配置修改


## APP 权限
## APP设置文件权限 这里的82为php-fpm容器中的 www-data的 usreid 和groupid
RUN chown -R www-data:www-data $APPROOT
#RUN chown -R 82:82 $APPROOT

# 入口
CMD ["/usr/bin/caddy", "--conf", "/etc/Caddyfile", "--log", "stdout"]

100

101

102

103

104

105

106

107

108

109

110

111

FROM php:7.2-fpm-alpine

LABEL version="7.2" \

description="php-fpm7.2 alpine" \

maintainer="wwek<licoolgo@gmail.com>"

# 国内源

RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories

# 系统基础包

RUN apk update \

&& apk upgrade \

# && apk add s6 \

&& apk add bash \

# && apk add openssl \

# && apk add openssl-dev \

&& apk add ca-certificates \

#&& apk add wget \

&& apk add curl \

#&& apk add tcpdump \

&& apk add iputils \

&& apk add iproute2 \

&& apk add libc6-compat \

&& apk add zlib \

&& apk add -U tzdata \

&& apk add libpng \

&& apk add libpng-dev \

#&& apk add libstdc++ \

&& rm -rf /var/cache/apk/*

# 系统设置(设置时区)

RUN rm -rf /etc/localtime \

&& ln -s /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \

&& echo "Asia/Shanghai" /etc/timezone

# --- PHP --- #

## PHP Core核心扩展

RUN docker-php-ext-install opcache mysqli pdo_mysql mbstring zip pcntl gd

## 三方扩展

## redis

RUN set -xe \

#&& apk add --no-cache --update --virtual .phpize-deps $PHPIZE_DEPS \

&& apk add --update --virtual .phpize-deps $PHPIZE_DEPS \

&& pecl install -o -f redis \

#&& pecl install -o -f swoole \

&& echo "extension=redis.so" > /usr/local/etc/php/conf.d/redis.ini \

#&& echo "extension=swoole.so" > /usr/local/etc/php/conf.d/swoole.ini \

&& rm -rf /usr/share/php \

&& rm -rf /tmp/* \

&& apk del .phpize-deps

# 增加 PHP 配置文件

ADD docker/www.conf /usr/local/etc/php-fpm.d/

ADD docker/php.ini /usr/local/etc/php

## PHP设置

# 设置php运行权限为母鸡中 www的用户和组 1000:1000

# RUN addgroup -g 1000 -S www && \

# adduser -u 1000 -S www -G www www

# 设置php-fpm运行账户和组为 www www

# RUN set -ex \

# && cd /usr/local/etc \

# && { \

# echo 'listen.owner = www'; \

# echo 'listen.group = www'; \

# echo 'listen.mode = 0660'; \

# } | tee php-fpm.d/www.conf

# -- Caddy -- #

## 安装 Caddy Web Server

#RUN curl -sSL -o- "https://caddyserver.com/download/linux/amd64?plugins=http.jwt,http.login,http.prometheus,http.realip,http.restic,http.expires&license=personal&telemetry=off" | tar -xvz -C /usr/bin/

RUN curl --silent --show-error --fail --location \

--header "Accept: application/tar+gzip, application/x-gzip, application/octet-stream" -o - \

"https://caddyserver.com/download/linux/amd64?plugins=http.jwt,http.login,http.prometheus,http.realip,http.restic,http.expires&license=personal&telemetry=off" \

| tar --no-same-owner -C /usr/bin/ -xz caddy \

&& chmod 0755 /usr/bin/caddy \

&& /usr/bin/caddy -version

## 增加 Caddy 配置文件

ADD docker/Caddyfile /etc/Caddyfile

# -- APP -- #

## APP环境变量

ENV APPROOT="/data/wwwroot/www.iamle.com/" \

PHP_MYSQL_HOSTNAME='' \

PHP_MYSQL_DATABASE='' \

PHP_MYSQL_USERNAME='' \

PHP_MYSQL_PASSWORD='' \

PHP_MYSQL_PORT='' \

PHP_REDIS_HOST=''

## APP增加

RUN mkdir -p $APPROOT

COPY ./ $APPROOT

## APP配置

WORKDIR $APPROOT

# 巴拉巴拉 APP 配置修改

## APP 权限

## APP设置文件权限这里的82为php-fpm容器中的 www-data的 usreid 和groupid

RUN chown -R www-data:www-data $APPROOT

#RUN chown -R 82:82 $APPROOT

# 入口

CMD ["/usr/bin/caddy", "--conf", "/etc/Caddyfile", "--log", "stdout"]

Caddyfile 参考

0.0.0.0:80
log stdout
errors stdout

root /data/wwwroot/www.iamle.com/public/

fastcgi / 127.0.0.1:9000 php {
    ext   .php
    split .php
    index index.php
}

rewrite {
    to {path} {path}/ /index.php?s={uri}
}

on startup php-fpm --nodaemonize

0.0.0.0:80

log stdout

errors stdout

root /data/wwwroot/www.iamle.com/public/

fastcgi / 127.0.0.1:9000 php {

ext .php

split .php

index index.php

}

rewrite {

to {path} {path}/ /index.php?s={uri}

}

on startup php-fpm --nodaemonize

www.conf

[www]
prefix = /var/
user = www-data
group = www-data
listen = 127.0.0.1:9000
;listen.backlog = 511
listen.owner = www-data
listen.group = www-data
listen.mode = 0660
;listen.acl_users =
;listen.acl_groups =
;listen.allowed_clients = 127.0.0.1

pm = dynamic
; Note: Used when pm is set to 'static', 'dynamic' or 'ondemand'
pm.max_children = 100
; Default Value: min_spare_servers + (max_spare_servers - min_spare_servers) / 2
pm.start_servers = 20
pm.min_spare_servers = 20
pm.max_spare_servers = 100
;pm.process_idle_timeout = 10s;
pm.max_requests = 50000
;pm.status_path = /status

;ping.path = /ping
;ping.response = pong

;access.log = log/$pool.access.log

;access.format = "%R - %u %t \"%m %r%Q%q\" %s %f %{mili}d %{kilo}M %C%%"

slowlog = log/$pool.slow.log

request_slowlog_timeout = 30

;request_terminate_timeout = 0

;rlimit_files = 1024

;rlimit_core = 0

;chroot =

;chdir = /var/www

catch_workers_output = yes

;clear_env = no

;security.limit_extensions = .php .php3 .php4 .php5 .php7

;env[HOSTNAME] = $HOSTNAME
;env[PATH] = /usr/local/bin:/usr/bin:/bin
;env[TMP] = /tmp
;env[TMPDIR] = /tmp
;env[TEMP] = /tmp


;php_admin_value[sendmail_path] = /usr/sbin/sendmail -t -i -f www@my.domain.com
;php_flag[display_errors] = off
;php_admin_value[error_log] = /var/log/fpm-php.www.log
;php_admin_flag[log_errors] = on
;php_admin_value[memory_limit] = 32M

[www]

prefix = /var/

user = www-data

group = www-data

listen = 127.0.0.1:9000

;listen.backlog = 511

listen.owner = www-data

listen.group = www-data

listen.mode = 0660

;listen.acl_users =

;listen.acl_groups =

;listen.allowed_clients = 127.0.0.1

pm = dynamic

; Note: Used when pm is set to 'static', 'dynamic' or 'ondemand'

pm.max_children = 100

; Default Value: min_spare_servers + (max_spare_servers - min_spare_servers) / 2

pm.start_servers = 20

pm.min_spare_servers = 20

pm.max_spare_servers = 100

;pm.process_idle_timeout = 10s;

pm.max_requests = 50000

;pm.status_path = /status

;ping.path = /ping

;ping.response = pong

;access.log = log/$pool.access.log

;access.format = "%R - %u %t \"%m %r%Q%q\" %s %f %{mili}d %{kilo}M %C%%"

slowlog = log/$pool.slow.log

request_slowlog_timeout = 30

;request_terminate_timeout = 0

;rlimit_files = 1024

;rlimit_core = 0

;chroot =

;chdir = /var/www

catch_workers_output = yes

;clear_env = no

;security.limit_extensions = .php .php3 .php4 .php5 .php7

;env[HOSTNAME] = $HOSTNAME

;env[PATH] = /usr/local/bin:/usr/bin:/bin

;env[TMP] = /tmp

;env[TMPDIR] = /tmp

;env[TEMP] = /tmp

;php_admin_value[sendmail_path] = /usr/sbin/sendmail -t -i -f www@my.domain.com

;php_flag[display_errors] = off

;php_admin_value[error_log] = /var/log/fpm-php.www.log

;php_admin_flag[log_errors] = on

;php_admin_value[memory_limit] = 32M

其他

在dockerfile中，是直接下载的caddy二进制程序，这个会受到caddy的授权协议影响，对个人免费，对商业收费
如果是基于源码自己编译安装那么就只需要遵守Apache license协议即可，即可以用于商业

常见应用框架配置参考
https://github.com/caddyserver/examples

cady缺省（占位符）变量
https://caddyserver.com/docs/placeholders

Filebeat Nginx Module 自定义字段

2019-03-23 by wwek·0评论

1. 前言

filebeat 6.6.2 nginx module自定义字段
filebeat提供了多种Module预制模块，简化了各种日志的格式化
在nginx中默认的字段并不满足实际需求，例如我们需要记录额外的Nginx字段
例如请求时间、后端响应时间、主机头等信息
那么在filebeat的nginx module中需要同步定义

2. rpm模式安装filebeat

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.6.2-x86_64.rpm
sudo rpm -vi filebeat-6.6.2-x86_64.rpm

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.6.2-x86_64.rpm

sudo rpm -vi filebeat-6.6.2-x86_64.rpm

3. Nginx 自定义Log格式化

nginx原始log格式化定义：

        #log format
        log_format access '$remote_addr - $remote_user [$time_local] "$request" '
                            '$status $body_bytes_sent "$http_referer" '
                            '"$http_user_agent" "$http_x_forwarded_for" '

#log format

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

'$status $body_bytes_sent "$http_referer" '

'"$http_user_agent" "$http_x_forwarded_for" '

nginx自定义格式化log：

        #log format
        log_format access '$remote_addr - $remote_user [$time_local] "$request" '
                            '$status $body_bytes_sent "$http_referer" '
                            '"$http_user_agent" "$http_x_forwarded_for" '
                            '"$http_x_real_ip" "$server_addr" "$host" '
                            '$request_time $upstream_response_time "$upstream_addr" '
                            '"$time_iso8601"';

#log format

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

'$status $body_bytes_sent "$http_referer" '

'"$http_user_agent" "$http_x_forwarded_for" '

'"$http_x_real_ip" "$server_addr" "$host" '

'$request_time $upstream_response_time "$upstream_addr" '

'"$time_iso8601"';

4. 启用filebeat的nginx module

执行

filebeat modules enable nginx

1 2	filebeat modules enable nginx

filebeat nginx module 增加字段
文件 /usr/share/filebeat/module/nginx/access/ingest/default.json 中
原始：

    "grok": {
      "field": "message",
      "patterns":[
        "\"?%{IP_LIST:nginx.access.remote_ip_list} - %{DATA:nginx.access.user_name} \\[%{HTTPDATE:nginx.access.time}\\] \"%{GREEDYDATA:nginx.access.info}\" %{NUMBER:nginx.access.response_code} %{NUMBER:nginx.access.body_sent.bytes} \"%{DATA:nginx.access.referrer}\" \"%{DATA:nginx.access.agent}\""
        ],
      "pattern_definitions": {
        "IP_LIST": "%{IP}(\"?,?\\s*%{IP})*"
      }

"grok": {

"field": "message",

"patterns":[

"pattern_definitions": {

"IP_LIST": "%{IP}(\"?,?\\s*%{IP})*"

}

自定义修改：

    "grok": {
      "field": "message",
      "patterns":[
        "\"?%{IP_LIST:nginx.access.remote_ip_list} - %{DATA:nginx.access.user_name} \\[%{HTTPDATE:nginx.access.time}\\] \"%{GREEDYDATA:nginx.access.info}\" %{NUMBER:nginx.access.response_code} %{NUMBER:nginx.access.body_sent.bytes} \"%{DATA:nginx.access.referrer}\" \"%{DATA:nginx.access.agent}\" \"%{DATA:nginx.access.xff}\" \"%{DATA:nginx.access.x_real_ip}\" \"%{DATA:nginx.access.server_addr}\" \"%{DATA:nginx.access.host}\" %{DATA:nginx.access.request_time} %{DATA:nginx.access.upstream_response_time} \"%{DATA:nginx.access.upstream_addr}\" \"%{DATA:nginx.access.time_iso8601}\""
        ],
      "pattern_definitions": {
        "IP_LIST": "%{IP}(\"?,?\\s*%{IP})*"
      }

"grok": {

"field": "message",

"patterns":[

"\"?%{IP_LIST:nginx.access.remote_ip_list} - %{DATA:nginx.access.user_name} \\[%{HTTPDATE:nginx.access.time}\\] \"%{GREEDYDATA:nginx.access.info}\" %{NUMBER:nginx.access.response_code} %{NUMBER:nginx.access.body_sent.bytes} \"%{DATA:nginx.access.referrer}\" \"%{DATA:nginx.access.agent}\" \"%{DATA:nginx.access.xff}\" \"%{DATA:nginx.access.x_real_ip}\" \"%{DATA:nginx.access.server_addr}\" \"%{DATA:nginx.access.host}\" %{DATA:nginx.access.request_time} %{DATA:nginx.access.upstream_response_time} \"%{DATA:nginx.access.upstream_addr}\" \"%{DATA:nginx.access.time_iso8601}\""

"pattern_definitions": {

"IP_LIST": "%{IP}(\"?,?\\s*%{IP})*"

}

如果定制自己的字段
可以使用http://grokdebug.herokuapp.com/ 在线工具验证grok规则
可以使用Json工具处理转义问题

文件 /etc/filebeat/fields.yml 中
找到nginx字段配置
在

            - name: agent
              type: text
              description: >
                Contains the un-parsed user agent string. Only present if the user
                agent Elasticsearch plugin is not available or not used.

- name: agent

type: text

description: >

Contains the un-parsed user agent string. Only present if the user

agent Elasticsearch plugin is not available or not used.

在后面增加

            - name: xff
              type: group
              description: >
                http_x_forwarded_for.
            - name: x_real_ip
              type: group
              description: >
                http_x_real_ip.
            - name: server_addr
              type: group
              description: >
                server_addr 服务器地址.
            - name: host
              type: group
              description: >
                host http_host http主机头.
            - name: request_time
              type: group
              description: >
                request_time 请求时间.
            - name: upstream_response_time
              type: group
              description: >
                upstream_response_time 后端响应时间.
            - name: upstream_addr
              type: group
              description: >
                upstream_addr 后端地址.
            - name: time_iso8601
              type: group
              description: >
                time_iso8601 iso8601格式时间.

- name: xff

type: group

description: >

http_x_forwarded_for.

- name: x_real_ip

type: group

description: >

http_x_real_ip.

- name: server_addr

type: group

description: >

server_addr 服务器地址.

- name: host

type: group

description: >

host http_host http主机头.

- name: request_time

type: group

description: >

request_time 请求时间.

- name: upstream_response_time

type: group

description: >

upstream_response_time 后端响应时间.

- name: upstream_addr

type: group

description: >

upstream_addr 后端地址.

- name: time_iso8601

type: group

description: >

time_iso8601 iso8601格式时间.

5. Filebeta中 nginx 日志路径定义

在文件 /etc/filebeat/modules.d/nginx.yml 中修改日志路径

- module: nginx
  # Access logs
  access:
    enabled: true

    # Set custom paths for the log files. If left empty,
    # Filebeat will choose the paths depending on your OS.
    var.paths: ["/data/wwwlogs/*.log*"]

  # Error logs
  error:
    enabled: true

    # Set custom paths for the log files. If left empty,
    # Filebeat will choose the paths depending on your OS.
    var.paths: ["/usr/local/nginx/logs/error.log*"]

- module: nginx

# Access logs

access:

enabled: true

# Set custom paths for the log files. If left empty,

# Filebeat will choose the paths depending on your OS.

var.paths: ["/data/wwwlogs/*.log*"]

# Error logs

error:

enabled: true

# Set custom paths for the log files. If left empty,

# Filebeat will choose the paths depending on your OS.

var.paths: ["/usr/local/nginx/logs/error.log*"]

6. 扩展

6.1 解决增加字段后Kibana中不显示

在Kibana的 Dev Tools中执行

DELETE _ingest/pipeline/filebeat-6.6.2-nginx-access-default

1 2	DELETE _ingest/pipeline/filebeat-6.6.2-nginx-access-default

字段出来以后有黄色小三角,刷新一下index 的字段缓存就好了
重启下Filebeat
方法来源

开源API网关大全

2019-03-22 by wwek·0评论

1. Kong

Kong官网
 Kong开源
Openresty + Lua 技术栈
Kong是一个云原生，快速，可扩展和分布式微服务抽象层（也称为API网关，API中间件或在某些情况下为Service Mesh）。作为2015年的开源项目，其核心价值在于高性能和可扩展性。
Kong积极维护，广泛应用于从创业公司到Global 5000以及政府组织等公司的生产。

Features

Cloud-Native: Platform agnostic, Kong can run from bare metal to
Kubernetes.
Dynamic Load Balancing: Load balance traffic across multiple upstream
services.
Hash-based Load Balancing: Load balance with consistent hashing/sticky
sessions.
Circuit-Breaker: Intelligent tracking of unhealthy upstream services.
Health Checks: Active and passive monitoring of your upstream services.
Service Discovery: Resolve SRV records in third-party DNS resolvers like
Consul.
Serverless: Invoke and secure AWS Lambda or OpenWhisk functions directly
from Kong.
WebSockets: Communicate to your upstream services via WebSockets.
OAuth2.0: Easily add OAuth2.0 authentication to your APIs.
Logging: Log requests and responses to your system over HTTP, TCP, UDP,
or to disk.
Security: ACL, Bot detection, whitelist/blacklist IPs, etc…
Syslog: Logging to System log.
SSL: Setup a Specific SSL Certificate for an underlying service or API.
Monitoring: Live monitoring provides key load and performance server
metrics.
Forward Proxy: Make Kong connect to intermediary transparent HTTP proxies.
Authentications: HMAC, JWT, Basic, and more.
Rate-limiting: Block and throttle requests based on many variables.
Transformations: Add, remove, or manipulate HTTP requests and responses.
Caching: Cache and serve responses at the proxy layer.
CLI: Control your Kong cluster from the command line.
REST API: Kong can be operated with its RESTful API for maximum
flexibility.
Geo-Replicated: Configs are always up-to-date across different regions.
Failure Detection & Recovery: Kong is unaffected if one of your Cassandra
nodes goes down.
Clustering: All Kong nodes auto-join the cluster keeping their config
updated across nodes.
Scalability: Distributed by nature, Kong scales horizontally by simply
adding nodes.
Performance: Kong handles load with ease by scaling and using NGINX at
the core.
Plugins: Extendable architecture for adding functionality to Kong and
APIs.

点评: 开源版本功能基本够用

2. Gravitee

Gravitee官网
 Gravitee开源
Java技术栈
Gravitee.io – API Management – OpenSource API Gateway
网关是Gravitee.io平台的核心组件。您可以将其与“智能”代理进行比较，以了解其目标。
与传统的HTTP代理不同，网关能够根据您的需要将策略（即规则）应用于HTTP请求和响应，这意味着您可以通过添加转换，安全性和许多其他疯狂功能来增强请求和响应处理！

主要功能特性

REST API：通过 Web UI 执行的每个操作都使用内部的 Rest API
一键部署：只需单击一下，API 就会部署到每个 Gravitee.io 网关，并且可以随时使用
高可扩展：可以轻松地将新的 Gravitee.io 网关添加到集群中。并且数据已同步，无需浪费时间进行配置
自定义策略：Gravitee.io 提供了许多开箱即用的策略（速率限制、CORS、IP过滤等）。如果还不够，可以自己进行开发
健康检测：作为 API 提供商，请为你的服务添加运行状况检查，并向你的用户提供有关 API 可用性的反馈

点评: 功能非常全,自带的dashboard也非常好用

3. Zuul

Zuul官网
 Zuul开源
Java技术栈
Zuul is a gateway service that provides dynamic routing, monitoring, resiliency, security, and more.
Zuul是一种网关服务，可提供动态路由，监控，弹性，安全性等。

Authentication and Security – identifying authentication requirements for each resource and rejecting requests that do not satisfy them.
Insights and Monitoring – tracking meaningful data and statistics at the edge in order to give us an accurate view of production.
Dynamic Routing – dynamically routing requests to different backend clusters as needed.
Stress Testing – gradually increasing the traffic to a cluster in order to gauge performance.
Load Shedding – allocating capacity for each type of request and dropping requests that go over the limit.
Static Response handling – building some responses directly at the edge instead of forwarding them to an internal cluster
Multiregion Resiliency – routing requests across AWS regions in order to diversify our ELB usage and move our edge closer to our members
点评: 用Spring Cloud全家桶首选

4. Tky

Tyk官网
 Tyk开源
Golang技术栈
Tyk Open Source API Gateway written in Go
Tyk是一个轻量级的开源API网关，管理平台使您能够控制访问API的人员，访问API以及访问API的方式。 Tyk还将记录用户如何与您的API进行交互以及出现问题时的详细分析。
Key Features of Tyk
Tyk offers powerful, yet lightweight features that allow fine grained control over your API ecosystem.

RESTFul API – Full programmatic access to the internals makes it easy to manage your API users, keys and Api Configuration from within your systems
Multiple access protocols – Out of the box, Tyk supports Token-based, HMAC Signed, Basic Auth and Keyless access methods
Rate Limiting – Easily rate limit your API users, rate limiting is granular and can be applied on a per-key basis
Quotas – Enforce usage quotas on users to manage capacity or charge for tiered access
Granular Access Control – Grant api access on a version by version basis, grant keys access to multiple API’s or just a single version
Key Expiry – Control how long keys are valid for
API Versioning – API Versions can be easily set and deprecated at a specific time and date
Blacklist/Whitelist/Ignored endpoint access – Enforce strict security models on a version-by-version basis to your access points
Analytics logging – Record detailed usage data on who is using your API’s (raw data only)
Webhooks – Trigger webhooks against events such as Quota Violations and Authentication failures
IP Whitelisting – Block access to non-trusted IP addresses for more secure interactions
Zero downtime restarts – Tyk configurations can be altered dynamically and the service restarted without affecting any active request

点评: 功能较全,但是授权方面是个问题

5. GO Gateway

GO Gateway网关官网
 GO Gateway网关开源
Golang技术栈
Gateway 是一个基于HTTP协议的restful的API网关。可以作为统一的API接入层。

Features

流量控制(Server或API级别)
熔断(Server或API级别)
负载均衡
服务发现
插件机制
路由(分流，复制流量)
API 聚合
API 参数校验
API 访问控制（黑白名单）
API 默认返回值
API 定制返回值
API 结果Cache
JWT Authorization
API Metric导入Prometheus
API 失败重试
后端server的健康检查
开放管理API(GRPC、Restful)
支持websocket
支持在线迁移数据

点评: 全开源,自主可控性高

6. GOKU API Gateway

GOKU API Gateway官网
 GOKU API Gateway开源
Golang技术栈
GoKu API Gateway CE，中文名：悟空API网关（开源版），是eoLinker旗下的API网关，帮助企业进行API服务治理、API性能安全维护，为企业数字化赋能。
GoKu API Gateway CE，支持OpenAPI与微服务管理，支持私有云部署，实现API转发、请求参数转换、数据校验等功能，提供图形化界面管理，能够快速管理多个API网关，提高API业务安全性。

产品功能

管理
访问统计统计API的访问数据，及时了解API使用情况
图形化界面通过界面即可设定网关的各项参数
企业人员管理为企业内部人员分配操作权限
OPEN API 提供完善的OPEN API，方便企业接入内部系统
性能与安全
负载均衡合理分配流量，提高业务的处理能力
权限控制针对不同用户设定不同的访问权限
集群部署支持集群部署，提高系统可用性
请求转发
支持REST API支持接入REST API
API整合通过不同的URL访问同一个API
超时重试超过预设的时间没有得到返回结果，则重新发起请求
告警机制
告警邮件通过邮件接收告警信息
告警推送通过API接受告警信息
拓展
插件支持提供官方插件，支持自定义插件
导入eoLinker AMS数据支持导入eoLinker AMS的API文档
日志系统
请求日志通过日志详细记录请求过程
告警日志通过API接受告警信息
插件日志通过日志详细记录插件使用过程

插件系统

用户鉴权
Basic鉴权通过Basic鉴权访问API
Apikey鉴权通过Apikey鉴权访问API
OAuth2.0 鉴权通过OAuth2.0鉴权访问API
JWT鉴权通过JWT鉴权访问API
流量管控
流量控制限制单位时间内的最大访问次数
数据缓存缓存后，用户请求时网关直接返回结果
请求大小限制在大小限制内的请求才能通过访问
安全防御
IP黑白名单根据IP判断请求合法性
防重放攻击保证请求的唯一性
跨域插件设定特定的响应头部实现跨域
返回头部自定义网关的返回头部
参数处理
参数映射实现参数A转换成参数B
额外参数转发时可自动添加预设的额外参数
参数校验通过校验规则的参数方可正常发送
格式转换参数的json格式与xml格式相互转换
响应机制
熔断机制达到熔断条件网关停止对该API的转发，返回预设内容
服务降级达到降级条件网关停止对该API的转发，返回降级内容
默认返回当匹配不到URI时，网关返回预先设定的值

点评:开源版基本不能投产,商业版功能丰富

7. Ocelot

Ocelot官网
 Ocelot开源
.NET技术栈
Ocelot是一个.NET API网关。该项目针对的是使用.NET运行面向微服务/面向服务的体系结构的人，这些体系结构需要统一的入口点。
但是，它可以与任何说HTTP并在ASP.NET Core支持的任何平台上运行的东西一起使用。

Features

Routing
Request Aggregation
Service Discovery with Consul & Eureka
Service Fabric
WebSockets
Authentication
Authorisation
Rate Limiting
Caching
Retry policies / QoS
Load Balancing
Logging / Tracing / Correlation
Headers / Query String / Claims Transformation
Custom Middleware / Delegating Handlers
Configuration / Administration REST API
Platform / Cloud Agnostic

其他API网关

小豹API网关(收费)

小豹API网关官网
Java技术栈
功能特性介绍

认证支持4种认证方式,包括：Baisc认证、JWT认证、微信小程序Token认证、NULL认证（放行）
鉴权支持2种授权方式，包括：基于接入应用进行授权、NULL授权（放行）
流量管控支持2种流程管理，包括：1、基于客户端应用进行流量管控，防止客户端恶意获取数据。 2、基于API的流量管理，保护上游服务。
缓存缓存API后端服务返回的数据，降低后端服务压力。
安全防护保护API的安全是API网关重任。 1、客户端身份认证；2、只有授权的应用才能访问对应的API。3、公私钥算法保护API接口调用和数据传输安全。4、IP白名单、黑名单机制。5、通过组件二次开发可以扩展很多定制安全功能，如：设备id，地理位置、敏感数据二次检验等。
服务路由识别请求参数和路径转发上游API服务。
日志记录默认以分钟纬度对API调用情况进行汇总记录，也支持配置第三方服务记录完整的API调用详细信息，方便自定义审计和统计。
服务注册和暴露支持Http rest、Webservice服务注册和暴露。
负载均衡支持网关内部负载均衡，用户可以选择注册代理和服务实例地址。
服务编排将多个后端服务聚合成一个API暴露，减小客户端开发复杂度，快速生产新的API。
超时控制通过API配置访问超时时间，超时后立即返回，避免工作线程长时间等等。
灰度发布产品支持让指定部分客户端应用使用新版本API，其他应用继续使用以前的API，当新版本的API逐渐稳定后，所有用户能平滑过渡到新版本的API。使用灰度发布能够及早获得用户的意见反馈，完善产品功能，提升产品质量。让用户参与产品测试，加强与用户互动，降低产品升级所影响的用户范围。
熔断后端服务N秒内连续50%调用出现超时或者异常，系统自动熔断该服务，M秒后尝试恢复链路；该机制可以有效避免无效等等。
动态配置更新运行时变更API、分组、组件等配置，自动同步、及时生效。
API系统迁移支持API以系统维度接入API网关，自动纪录API，让API系统迁移自动化。
高可用产品支持企业级高可用部署，避免单点故障。
高性能网关无状态，支持横向扩展，运行时弱依赖持久层，最大化提升网关整体性能。
分布式缓存(redis) 当系统API缓存量大时，可使用redis集群提高系统的响应。
自定义组件开发采用组件化的架构，可根据需求通过开发组件匹配企业需求。
管理中心 API网关后台管理中心简洁明了，方便API网关管理员对API进行管理，数据面板方便了解系统运行情况。
监控中心监控中心对API运行情况一目了然，通过分析API异常告警辅助系统运维。
微信小程序支持对小程序类型的客户端应用进行了特殊处理，简化小程序对微信用户进行认证，后端服务只需要关系业务逻辑。

Flink-基于Docker的开发环境搭建

2019-03-16 by wwek·0评论

1. Flink开发环境安装

1.1 前置安装

开发工具IDEA
Docker环境(Flink平台运行在docker中)
jdk支持
Maven支持
netcat支持

jdk,Maven推荐OSX用brew安装,Windows用Chocolatey安装
netcat在Windows上用Chocolatey安装,命令为”choco install netcat”

1.1 docker-compose 安装flink

docker-compose.yml

version: "2.1"
services:
  jobmanager:
    image: flink
    expose:
      - "6123"
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

  taskmanager:
    image: flink
    expose:
      - "6121"
      - "6122"
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - "jobmanager:jobmanager"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

version: "2.1"

services:

jobmanager:

image: flink

expose:

- "6123"

ports:

- "8081:8081"

command: jobmanager

environment:

- JOB_MANAGER_RPC_ADDRESS=jobmanager

taskmanager:

image: flink

expose:

- "6121"

- "6122"

depends_on:

- jobmanager

command: taskmanager

links:

- "jobmanager:jobmanager"

environment:

- JOB_MANAGER_RPC_ADDRESS=jobmanager

运行

docker-compose up -d

1 2	docker-compose up -d

浏览器打开 http://127.0.0.1:8081 可以看到dashboard

2. 编写一个简单的流处理 job

假设这个job项目名称为helloword
这个job用nc程序监听一个tcp服务,使其可以输入文本流
计算单词出现次数

2.1 Maven 初始化项目

mvn archetype:generate \
      -DarchetypeGroupId=org.apache.flink \
      -DarchetypeArtifactId=flink-quickstart-java \
      -DarchetypeVersion=1.7.2

mvn archetype:generate \

-DarchetypeGroupId=org.apache.flink \

-DarchetypeArtifactId=flink-quickstart-java \

-DarchetypeVersion=1.7.2

设建立的项目
groupId为com.iamle.flink
artifactId为helloword

如果是 Windows,建议使用 Git Bash 终端执行mvn

2.2 IDEA打开helloword

删除自动生成的 BatchJob.java、StreamingJob.java
新建SocketWindowWordCount.java

package com.iamle.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class SocketWindowWordCount {

    public static void main(String[] args) throws Exception {

        // 输入tcp流
        final int port;
        final String host;
        port = 9008; // nc监听的tcp端口
        host = "192.168.0.8"; // docker宿主机ip

        // get the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // get input data by connecting to the socket
        DataStream<String> text = env.socketTextStream(host, port, "\n");

        // parse the data, group it, window it, and aggregate the counts
        DataStream<WordWithCount> windowCounts = text
                .flatMap(new FlatMapFunction<String, WordWithCount>() {
                    @Override
                    public void flatMap(String value, Collector<WordWithCount> out) {
                        for (String word : value.split("\\s")) {
                            out.collect(new WordWithCount(word, 1L));
                        }
                    }
                })
                .keyBy("word")
                .timeWindow(Time.seconds(5), Time.seconds(1))
                .reduce(new ReduceFunction<WordWithCount>() {
                    @Override
                    public WordWithCount reduce(WordWithCount a, WordWithCount b) {
                        return new WordWithCount(a.word, a.count + b.count);
                    }
                });

        // print the results with a single thread, rather than in parallel
        windowCounts.print().setParallelism(1);

        env.execute("Socket Window WordCount");
    }

    // Data type for words with count
    public static class WordWithCount {

        public String word;
        public long count;

        public WordWithCount() {}

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }
}

package com.iamle.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.functions.ReduceFunction;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.util.Collector;

public class SocketWindowWordCount {

public static void main(String[] args) throws Exception {

// 输入tcp流

final int port;

final String host;

port = 9008; // nc监听的tcp端口

host = "192.168.0.8"; // docker宿主机ip

// get the execution environment

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// get input data by connecting to the socket

DataStream<String> text = env.socketTextStream(host, port, "\n");

// parse the data, group it, window it, and aggregate the counts

DataStream<WordWithCount> windowCounts = text

.flatMap(new FlatMapFunction<String, WordWithCount>() {

@Override

public void flatMap(String value, Collector<WordWithCount> out) {

for (String word : value.split("\\s")) {

out.collect(new WordWithCount(word, 1L));

}

})

.keyBy("word")

.timeWindow(Time.seconds(5), Time.seconds(1))

.reduce(new ReduceFunction<WordWithCount>() {

@Override

public WordWithCount reduce(WordWithCount a, WordWithCount b) {

return new WordWithCount(a.word, a.count + b.count);

}

});

// print the results with a single thread, rather than in parallel

windowCounts.print().setParallelism(1);

env.execute("Socket Window WordCount");

}

// Data type for words with count

public static class WordWithCount {

public String word;

public long count;

public WordWithCount() {}

public WordWithCount(String word, long count) {

this.word = word;

this.count = count;

}

@Override

public String toString() {

return word + " : " + count;

}

pom.xml 入口类
com.iamle.flink.StreamingJob 改为
com.iamle.flink.SocketWindowWordCount

2.3 编译项目jar

cd  helloword
mvn clean package

cd helloword

mvn clean package

在项目中的target目录中得到 helloword-1.0-SNAPSHOT-shaded.jar

3. 提交job

先运行流的输入服务

nc -l 9008
# nc -l -p 9008 # windows

nc -l 9008

# nc -l -p 9008 # windows

方式1
打开 http://127.0.0.1:8081/#/submit dashboard后
在Submit new job中上传helloword-1.0-SNAPSHOT-shaded.jar
upload后>勾选>Submit

在nc终端,随便输入一些字符串后回车
在名为taskmanager的docker实例标准终端中可以看到计算结果(可以用kitematic打开docker容器实例查看其标准终端)

4. 官方示例

flink官方示例