# active mq
docker run -itd --name activemq \
-p 61616:61616 -p 8161:8161 -p 5672:5672 -p 61613:61613 -p 1883:1883 -p 61614:61614 \
-v /usr/local/activemq/conf:/opt/activemq/conf \
-v /usr/local/activemq/data:/opt/activemq/data \
rmohr/activemq:5.15.6

. 用于分割名字中的多个单词
* 表示任一名字, 不包括点号(.)
> 表示任一名字, 包括点号(.), 用于表示前缀, >符号后面不会再跟其他限制条件.

<destinationInterceptors> 
            <virtualDestinationInterceptor> 
                <virtualDestinations> 
                    <virtualTopic name=">" prefix="Consumer.*." selectorAware="false"/>    
                </virtualDestinations>
            </virtualDestinationInterceptor> 
</destinationInterceptors>

<broker>
   
  <destinationPolicy>
    <policyMap>
      <policyEntries>
        <!-- Set the following policy on all queues using the '>' wildcard -->
        <policyEntry queue=">">
          <deadLetterStrategy>
            <!--
              Use the prefix 'DLQ.' for the destination name, and make
              the DLQ a queue rather than a topic
            -->
            <individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true"/>
          </deadLetterStrategy>
        </policyEntry>
      </policyEntries>
    </policyMap>
  </destinationPolicy>
</broker>

<!-- 
Tell the dead letter strategy to also place non-persisted messages 
onto the dead-letter queue if they can't be delivered.
-->
<deadLetterStrategy>
<...  processNonPersistent="true" />
</deadLetterStrategy>

package main

import (
"context"
"github.com/go-stomp/stomp"
"github.com/hanjm/log"
"os"
"os/signal"
"strconv"
"sync"
"syscall"
"time"
)

func main() {
var wg sync.WaitGroup
ctx, cancel := context.WithCancel(context.Background())

wg.Add(1)
go func() {
defer wg.Done()
publisher(ctx, "/topic/stomp")
}()

wg.Add(1)
go func() {
defer wg.Done()
Subscriber(ctx, "channel1", "Consumer.channel1.stomp")
}()
//
wg.Add(1)
go func() {
defer wg.Done()
Subscriber(ctx, "channel2", "Consumer.channel2.stomp")
}()

wg.Add(1)
go func() {
defer wg.Done()
Subscriber(ctx, "channel3", "/topic/stomp")
}()

defer func() {
cancel()
wg.Wait()
}()
SignalsListen()
}

func publisher(ctx context.Context, destination string) {
conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
if err != nil {
log.Fatal(err)
return
}
defer conn.Disconnect()
for i := 0; ; i++ {
select {
case <-ctx.Done():
return
case <-time.After(time.Second):
err = conn.Send(
destination,  // destination
"text/plain", // content-type
[]byte("Test message #"+strconv.Itoa(i)), stomp.SendOpt.Header("persistent", "true")) // body
if err != nil {
log.Error(err)
return
}
}
}
}

func Subscriber(ctx context.Context, clientID string, destination string) {
conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
if err != nil {
log.Fatal(err)
return
}
defer conn.Disconnect()
sub, err := conn.Subscribe(destination, stomp.AckClientIndividual, stomp.SubscribeOpt.Id(clientID), stomp.SubscribeOpt.Header("persistent", "true"))
if err != nil {
log.Fatal(err)
return
}
go func() {
select {
case <-ctx.Done():
err := sub.Unsubscribe()
if err != nil {
log.Fatal(clientID, err)
return
}
return
}
}()
for m := range sub.C {
if m.Err != nil {
log.Fatal(err)
return
}
log.Infof("%s msg body:%s", clientID, m.Body)
//log.Infof("%s msg header:%s", clientID, *m.Header)
//log.Infof("%s msg content-type:%s", clientID, m.ContentType)
//log.Infof("%s msg destination:%s", clientID, m.Destination)
m.Conn.Ack(m)
}
log.Info("close sub")
}

func SignalsListen() {
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGQUIT,
syscall.SIGTERM,
syscall.SIGINT,
syscall.SIGUSR1,
syscall.SIGUSR2)

switch <-sigs {
case syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT:
log.Info("service close")
}
return
}

package all_bench

import (
"bytes"
"context"
"github.com/eclipse/paho.mqtt.golang"
"github.com/go-stomp/stomp"
"github.com/hanjm/log"
"pack.ag/amqp"
"sync/atomic"
"testing"
"time"
)

var msgData = bytes.Repeat([]byte("1"), 1024)

var (
stompDestination = "bench-stomp"
amqpDestination  = "bench-amqp"
mqttDestination  = "bench-mqtt"
pubMsgCount      = 20000
subMsgCount      = 100
)

func TestMain(m *testing.M) {
m.Run()
}

// go test -bench Publish  -benchmem
// go test -bench Sub  -benchmem
func BenchmarkStompPublish(b *testing.B) {
conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
if err != nil {
log.Fatal(err)
return
}
defer conn.Disconnect()

b.N = pubMsgCount
b.ReportAllocs()
b.ResetTimer()

for i := 0; i < b.N; i++ {
err = conn.Send(
stompDestination, // destination
"text/plain",     // content-type
msgData) // body
if err != nil {
log.Error(err)
return
}
}
}

func BenchmarkAmqpPublish(b *testing.B) {
// Create client
client, err := amqp.Dial("amqp://127.0.0.1",
amqp.ConnSASLPlain("system", "manager"),
)
if err != nil {
log.Fatal("Dialing AMQP server:", err)
}
defer client.Close()

// Open a session
session, err := client.NewSession()
if err != nil {
log.Fatal("Creating AMQP session:", err)
}
defer func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err = session.Close(ctx)
if err != nil {
log.Errorf("failed to close session:%s", err)
return
}
//log.Info("session close")
}()

// Create a sender
sender, err := session.NewSender(
amqp.LinkTargetAddress(amqpDestination),
amqp.LinkSourceDurability(amqp.DurabilityUnsettledState),
amqp.LinkSourceExpiryPolicy(amqp.ExpiryNever),
)
if err != nil {
log.Fatal("Creating sender link:", err)
}

defer func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err := sender.Close(ctx)
if err != nil {
log.Errorf("failed to close sender:%s", err)
return
}
//log.Infof("sender close")
}()

ctx := context.Background()
msg := amqp.NewMessage(msgData)

b.N = pubMsgCount
b.ReportAllocs()
b.ResetTimer()

for i := 0; i < b.N; i++ {
// Send message
err = sender.Send(ctx, msg)
if err != nil {
log.Fatal("Sending message:", err)
}
if err != nil {
log.Fatal(err)
return
}
}
}

func BenchmarkMqttPublish(b *testing.B) {
opt := mqtt.NewClientOptions().SetClientID("pubClient").SetCleanSession(false)
opt.AddBroker("tcp://127.0.0.1:1883")
client := mqtt.NewClient(opt)
t := client.Connect()
err := t.Error()
if err != nil {
log.Fatal(err)
return
}
if t.Wait() {
err := t.Error()
if err != nil {
log.Fatal(err)
return
}
}
defer func() {
client.Disconnect(10000)
}()

b.N = pubMsgCount
b.ReportAllocs()
b.ResetTimer()

for i := 0; i < b.N; i++ {
token := client.Publish(mqttDestination, 2, true, msgData)
err := token.Error()
if err != nil {
log.Fatal(err)
return
}
}
}

func BenchmarkStompSubscriber(b *testing.B) {
conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
if err != nil {
log.Fatal(err)
return
}
clientID := "1"
//defer conn.Disconnect()
sub, err := conn.Subscribe(stompDestination, stomp.AckClientIndividual, stomp.SubscribeOpt.Id(clientID))
if err != nil {
log.Fatal(err)
return
}
//defer func() {
//err := sub.Unsubscribe()
//if err != nil {
//log.Fatal(clientID, err)
//return
//}
//return
//}()
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
defer cancel()

b.N = subMsgCount
b.ReportAllocs()
b.ResetTimer()
defer b.StopTimer()
var i int64 = 0

go func() {
for range time.Tick(time.Second) {
if atomic.LoadInt64(&i) >= int64(b.N) {
cancel()
}
}
}()

defer func() {
//log.Info("close")
}()
for {
select {
case m := <-sub.C:
if m.Err != nil {
log.Fatal(m.Err)
return
}
m.Conn.Ack(m)
i++
if atomic.LoadInt64(&i) > int64(b.N) {
return
}
case <-ctx.Done():
return
}
}
}

func BenchmarkAmqpSubscriber(b *testing.B) {
// Create client
client, err := amqp.Dial("amqp://127.0.0.1",
amqp.ConnSASLPlain("system", "manager"),
)
if err != nil {
log.Fatal("Dialing AMQP server:", err)
}
//defer client.Close()

// Open a session
session, err := client.NewSession()
if err != nil {
log.Fatal("Creating AMQP session:", err)
}

clientID := "1"
defer func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err := session.Close(ctx)
if err != nil {
log.Errorf("%s failed to close session:%s", clientID, err)
return
}
//log.Errorf("%s session close", clientID)
}()

// Continuously read messages
// Create a receiver
receiver, err := session.NewReceiver(
amqp.LinkSourceAddress(amqpDestination),
amqp.LinkCredit(10),
)
if err != nil {
log.Fatal("Creating receiver link:", err)
}
defer func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err := receiver.Close(ctx)
if err != nil {
log.Errorf("%s failed to close receiver:%s", clientID, err)
return
}
//log.Errorf("%s receiver close", clientID)
}()
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
defer cancel()

b.N = subMsgCount
b.ReportAllocs()
b.ResetTimer()
defer b.StopTimer()
var i int64 = 0

go func() {
for range time.Tick(time.Second) {
if atomic.LoadInt64(&i) >= int64(b.N) {
cancel()
}
}
}()
for {
// Receive next message
msg, err := receiver.Receive(ctx)
if err != nil {
if err == context.Canceled {
log.Infof("Reading message from AMQP:%s", err)
break
}
log.Errorf("Reading message from AMQP:%s", err)
break
}
//log.Infof("%s msg body:%s value:%T %s", clientID, msg.GetData(), msg.Value, msg.Value)
// Accept message
msg.Accept()
atomic.AddInt64(&i, 1)
if atomic.LoadInt64(&i) > int64(b.N) {
//log.Info("return")
return
}
}
}

func BenchmarkMqttSubscriber(b *testing.B) {
opt := mqtt.NewClientOptions().SetClientID("subClient").SetCleanSession(false)
opt.AddBroker("tcp://127.0.0.1:1883")
client := mqtt.NewClient(opt)
t := client.Connect()
if t.Wait() {
err := t.Error()
if err != nil {
log.Fatal(err)
return
}
}
defer func() {
client.Disconnect(1000)
}()

ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
defer cancel()

b.N = subMsgCount
b.ReportAllocs()
b.ResetTimer()
defer b.StopTimer()
var i int64 = 0

go func() {
for range time.Tick(time.Second) {
if atomic.LoadInt64(&i) >= int64(b.N) {
cancel()
}
}
}()
client.Subscribe(mqttDestination, 2, func(c mqtt.Client, m mqtt.Message) {
//log.Infof("%s msg body:%s", "1", m.Payload())
m.Ack()
atomic.AddInt64(&i, 1)
if atomic.LoadInt64(&i) > int64(b.N) {
//log.Info("return")
return
}
})
select {
case <-ctx.Done():
break
}
log.Info("close sub")
}

--user=www-data // 习惯将web相关的服务以www-data用户运行, 如没有此用户可以创建一个也可不加此项按默认nobody用户
--group=www-data
--with-http_v2_module // 默认选项不带http2
--with-http_ssl_module // 默认选项不带ssl, 上http2必须要上ssl的
--with-stream // https://nginx.org/en/docs/stream/ngx_stream_core_module.html
--with-openssl // 指定OpenSSL
--with-pcre=./pcre-8.40 // 需要(version 4.4 — 8.40)的pcre,注意Nginx不支持pcre2
--with-pcre-jit // 打开pcre JIT支持
--with-zlib=./zlib-1.2.11 // 需要(version 1.1.3 — 1.2.11)的zlib以支持gzip

cd /usr/local
wget https://nginx.org/download/nginx-1.14.2.tar.gz
tar -zxf nginx-1.14.2.tar.gz
cd nginx-1.14.2

cd nginx-1.14.2
wget https://github.com/openssl/openssl/archive/OpenSSL_1_1_0e.tar.gz
tar -zxf OpenSSL_1_1_0e.tar.gz

cd nginx-1.14.2
wget https://ftp.pcre.org/pub/pcre/pcre-8.40.tar.gz
tar -zxf pcre-8.40.tar.gz

cd nginx-1.14.2
wget http://zlib.net/zlib-1.2.11.tar.gz
tar -zxf zlib-1.2.11.tar.gz

./configure \
--user=nginx \
--group=nginx \
--conf-path=/etc/nginx/nginx.conf \
--with-http_v2_module \
--with-http_ssl_module \
--with-stream \
--with-openssl=./openssl-OpenSSL_1_1_0e \
--with-pcre=./pcre-8.40 --with-pcre-jit \
--with-zlib=./zlib-1.2.11
make && make install

ln -sf /usr/local/nginx/sbin/nginx /usr/local/bin

warning: Error disabling address space randomization: Operation not permitted
Cannot create process: Operation not permitted
During startup program exited with code 127.
(gdb)

>By default, Docker containers are “unprivileged” and cannot, for example, run a Docker daemon inside a Docker container. This is because by default a container is not allowed to access any devices, but a “privileged” container is given access to all devices (see the documentation on cgroups devices).

>When the operator executes docker run --privileged, Docker will enable access to all devices on the host as well as set some configuration in AppArmor or SELinux to allow the container nearly all the same access to the host as processes running outside containers on the host. Additional information about running with --privileged is available on the Docker Blog.

>If you want to limit access to a specific device or devices you can use the --device flag. It allows you to specify one or more devices that will be accessible within the container.

>

BenchmarkStructSliceWithoutPool-8                 200000              5458 ns/op           16320 B/op          8 allocs/op
BenchmarkStructPointerSliceWithoutPool-8          200000              6045 ns/op            8504 B/op        109 allocs/op
BenchmarkStructSliceWithPool-8                   1000000              1287 ns/op              32 B/op          1 allocs/op
BenchmarkStructPointerSliceWithPool-8             300000              4910 ns/op            6498 B/op        102 allocs/op

package main

import (
"sync"
"testing"
)

var structSlicePool = sync.Pool{
New: func() interface{} {
return make([]Basic, 100)
},
}

var structPointerSlicePool = sync.Pool{
New: func() interface{} {
return make([]*Basic, 100)
},
}

type Basic struct {
Id, N1, N2, N3, N4, N5 int
Name                   string
}

func BenchmarkStructSliceWithoutPool(b *testing.B) {
for i := 0; i < b.N; i++ {
var list []Basic
for j := 0; j < 101; j++ {
var data = Basic{Id: j, Name: "Name"}
list = append(list, data)
}
}
}

func BenchmarkStructPointerSliceWithoutPool(b *testing.B) {
for i := 0; i < b.N; i++ {
var list []*Basic
for j := 0; j < 101; j++ {
var data = Basic{Id: j, Name: "Name"}
list = append(list, &data)
}
}
}

func BenchmarkStructSliceWithPool(b *testing.B) {
for i := 0; i < b.N; i++ {
list := structSlicePool.Get().([]Basic)
initLen := len(list)
for j := 0; j < 101; j++ {
var data = Basic{Id: j, Name: "Name"}
if j < initLen {
list[j] = data
} else {
list = append(list, data)
}
}
structSlicePool.Put(list)
}
}

func BenchmarkStructPointerSliceWithPool(b *testing.B) {
for i := 0; i < b.N; i++ {
list := structPointerSlicePool.Get().([]*Basic)
initLen := len(list)
for j := 0; j < 101; j++ {
var data = Basic{Id: j, Name: "Name"}
if j < initLen {
list[j] = &data
} else {
list = append(list, &data)
}
}
structPointerSlicePool.Put(list)
}

}

go func(){
defer func(){
if err:=recover(); err!=nil{
}
}()
}()

package main

import (
"fmt"
"log"
"time"
)

func main() {
ch := make(chan int, 10)

for i := 0; i < 2; i++ {
go worker(ch, i)
}

for i := 0; i < 3; i++ {
ch <- i
ch <- -1
}

time.Sleep(time.Second * 5)
}

func worker(ch <-chan int, goId int) {

log.Printf("worker%d running", goId)
for data := range ch {
func() {
defer func() {
if err := recover(); err != nil {
log.Printf("worker%d recover error:%s", goId, err)
}
}()
log.Printf("worker%d received data:%d", goId, data)
if data == -1 {
panic(fmt.Errorf("worker%d panic", goId))
}
}()
}
}

package zaplog

import (
"bytes"
"fmt"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"log"
"runtime"
"strings"
)

// CallerEncoder will add caller to log. format is "filename:lineNum:funcName", e.g:"zaplog/zaplog_test.go:15:zaplog.TestNewLogger"
func CallerEncoder(caller zapcore.EntryCaller, enc zapcore.PrimitiveArrayEncoder) {
enc.AppendString(strings.Join([]string{caller.TrimmedPath(), runtime.FuncForPC(caller.PC).Name()}, ":"))
}

func newLoggerConfig(debugLevel bool) (loggerConfig zap.Config) {
loggerConfig = zap.NewProductionConfig()
loggerConfig.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
loggerConfig.EncoderConfig.EncodeCaller = CallerEncoder
if debugLevel {
loggerConfig.Level = zap.NewAtomicLevelAt(zap.DebugLevel)
}
return
}

// NewCustomLoggers is a shortcut to get normal logger, noCallerLogger.
func NewCustomLoggers(debugLevel bool) (logger, noCallerLogger *zap.Logger) {
loggerConfig := newLoggerConfig(debugLevel)
logger, err := loggerConfig.Build()
if err != nil {
panic(err)
}
loggerConfig.DisableCaller = true
noCallerLogger, err = loggerConfig.Build()
if err != nil {
panic(err)
}
return
}

// NewLogger return a normal logger
func NewLogger(debugLevel bool) (logger *zap.Logger) {
loggerConfig := newLoggerConfig(debugLevel)
logger, err := loggerConfig.Build()
if err != nil {
panic(err)
}
return
}

// NewNoCallerLogger return a no caller key value, will be faster
func NewNoCallerLogger(debugLevel bool) (noCallerLogger *zap.Logger) {
loggerConfig := newLoggerConfig(debugLevel)
loggerConfig.DisableCaller = true
noCallerLogger, err := loggerConfig.Build()
if err != nil {
panic(err)
}
return
}

// CompatibleLogger is a logger which compatible to logrus/std log/prometheus.
// it implements Print() Println() Printf() Dbug() Debugln() Debugf() Info() Infoln() Infof() Warn() Warnln() Warnf()
// Error() Errorln() Errorf() Fatal() Fataln() Fatalf() Panic() Panicln() Panicf() With() WithField() WithFields()
type CompatibleLogger struct {
_log *zap.Logger
}

// NewCompatibleLogger return CompatibleLogger with caller field
func NewCompatibleLogger(debugLevel bool) *CompatibleLogger {
return &CompatibleLogger{NewLogger(debugLevel).WithOptions(zap.AddCallerSkip(1))}
}

// Print logs a message at level Info on the compatibleLogger.
func (l CompatibleLogger) Print(args ...interface{}) {
l._log.Info(fmt.Sprint(args...))
}

// Println logs a message at level Info on the compatibleLogger.
func (l CompatibleLogger) Println(args ...interface{}) {
l._log.Info(fmt.Sprint(args...))
}

// Printf logs a message at level Info on the compatibleLogger.
func (l CompatibleLogger) Printf(format string, args ...interface{}) {
l._log.Info(fmt.Sprintf(format, args...))
}

// Debug logs a message at level Debug on the compatibleLogger.
func (l CompatibleLogger) Debug(args ...interface{}) {
l._log.Debug(fmt.Sprint(args...))
}

// Debugln logs a message at level Debug on the compatibleLogger.
func (l CompatibleLogger) Debugln(args ...interface{}) {
l._log.Debug(fmt.Sprint(args...))
}

// Debugf logs a message at level Debug on the compatibleLogger.
func (l CompatibleLogger) Debugf(format string, args ...interface{}) {
l._log.Debug(fmt.Sprintf(format, args...))
}

// Info logs a message at level Info on the compatibleLogger.
func (l CompatibleLogger) Info(args ...interface{}) {
l._log.Info(fmt.Sprint(args...))
}

// Infoln logs a message at level Info on the compatibleLogger.
func (l CompatibleLogger) Infoln(args ...interface{}) {
l._log.Info(fmt.Sprint(args...))
}

// Infof logs a message at level Info on the compatibleLogger.
func (l CompatibleLogger) Infof(format string, args ...interface{}) {
l._log.Info(fmt.Sprintf(format, args...))
}

// Warn logs a message at level Warn on the compatibleLogger.
func (l CompatibleLogger) Warn(args ...interface{}) {
l._log.Warn(fmt.Sprint(args...))
}

// Warnln logs a message at level Warn on the compatibleLogger.
func (l CompatibleLogger) Warnln(args ...interface{}) {
l._log.Warn(fmt.Sprint(args...))
}

// Warnf logs a message at level Warn on the compatibleLogger.
func (l CompatibleLogger) Warnf(format string, args ...interface{}) {
l._log.Warn(fmt.Sprintf(format, args...))
}

// Error logs a message at level Error on the compatibleLogger.
func (l CompatibleLogger) Error(args ...interface{}) {
l._log.Error(fmt.Sprint(args...))
}

// Errorln logs a message at level Error on the compatibleLogger.
func (l CompatibleLogger) Errorln(args ...interface{}) {
l._log.Error(fmt.Sprint(args...))
}

// Errorf logs a message at level Error on the compatibleLogger.
func (l CompatibleLogger) Errorf(format string, args ...interface{}) {
l._log.Error(fmt.Sprintf(format, args...))
}

// Fatal logs a message at level Fatal on the compatibleLogger.
func (l CompatibleLogger) Fatal(args ...interface{}) {
l._log.Fatal(fmt.Sprint(args...))
}

// Fatalln logs a message at level Fatal on the compatibleLogger.
func (l CompatibleLogger) Fatalln(args ...interface{}) {
l._log.Fatal(fmt.Sprint(args...))
}

// Fatalf logs a message at level Fatal on the compatibleLogger.
func (l CompatibleLogger) Fatalf(format string, args ...interface{}) {
l._log.Fatal(fmt.Sprintf(format, args...))
}

// Panic logs a message at level Painc on the compatibleLogger.
func (l CompatibleLogger) Panic(args ...interface{}) {
l._log.Panic(fmt.Sprint(args...))
}

// Panicln logs a message at level Painc on the compatibleLogger.
func (l CompatibleLogger) Panicln(args ...interface{}) {
l._log.Panic(fmt.Sprint(args...))
}

// Panicf logs a message at level Painc on the compatibleLogger.
func (l CompatibleLogger) Panicf(format string, args ...interface{}) {
l._log.Panic(fmt.Sprintf(format, args...))
}

// With return a logger with an extra field.
func (l *CompatibleLogger) With(key string, value interface{}) *CompatibleLogger {
return &CompatibleLogger{l._log.With(zap.Any(key, value))}
}

// WithField return a logger with an extra field.
func (l *CompatibleLogger) WithField(key string, value interface{}) *CompatibleLogger {
return &CompatibleLogger{l._log.With(zap.Any(key, value))}
}

// WithFields return a logger with extra fields.
func (l *CompatibleLogger) WithFields(fields map[string]interface{}) *CompatibleLogger {
i := 0
var clog *CompatibleLogger
for k, v := range fields {
if i == 0 {
clog = l.WithField(k, v)
} else {
clog = clog.WithField(k, v)
}
i++
}
return clog
}

// FormatStdLog set the output of stand package log to zaplog
func FormatStdLog() {
log.SetFlags(log.Llongfile)
log.SetOutput(&logWriter{NewNoCallerLogger(false)})
}

type logWriter struct {
logger *zap.Logger
}

// Write implement io.Writer, as std log's output
func (w logWriter) Write(p []byte) (n int, err error) {
i := bytes.Index(p, []byte(":")) + 1
j := bytes.Index(p[i:], []byte(":")) + 1 + i
caller := bytes.TrimRight(p[:j], ":")
// find last index of /
i = bytes.LastIndex(caller, []byte("/"))
// find penultimate index of /
i = bytes.LastIndex(caller[:i], []byte("/"))
w.logger.Info("stdLog", zap.ByteString("caller", caller[i+1:]), zap.ByteString("log", bytes.TrimSpace(p[j:])))
return len(p), nil
}

[mysql] 2017/04/26 10:01:05 packets.go:130: write tcp 127.0.0.1:56346->127.0.0.1:3306: write: broken pipe
[mysql] 2017/04/26 10:01:05 packets.go:130: write tcp 127.0.0.1:56346->127.0.0.1:3306: write: broken pipe
[mysql] 2017/04/26 10:01:05 packets.go:130: write tcp 127.0.0.1:56350->127.0.0.1:3306: write: broken pipe
[mysql] 2017/04/26 10:01:05 packets.go:130: write tcp 127.0.0.1:56350->127.0.0.1:3306: write: broken pipe

mysql> show variables like '%wait_timeout%';                                                                                                                                                                                                                      
+--------------------------+----------+
| Variable_name            | Value    |
+--------------------------+----------+
| innodb_lock_wait_timeout | 50       |
| lock_wait_timeout        | 31536000 |
| wait_timeout             | 28800    |
+--------------------------+----------+
3 rows in set (0.00 sec)

db.SetConnMaxLifetime(time.Hour*7)