Chemmy's Blog

chengming0916@gmail.com

0%

发表于 更新于 分类于 Valine：

DotNetty完全教程（八）

Excerpt

EventLoop介绍我们先回顾一下，EventLoop就是我们在最开始的示意图中的Selector，每个EventLoop和一个线程绑定，用于处理多个Channel。任务调度如果我们想实现延时任务的调度，比如连接成功5s之后发送一包数据，就可以用到EventLoop的计划任务ctx.Channel.EventLoop.Schedule(() =>{ Console.Wr…

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

EventLoop

介绍

我们先回顾一下，EventLoop就是我们在最开始的示意图中的Selector，每个EventLoop和一个线程绑定，用于处理多个Channel。

任务调度

  1. 如果我们想实现延时任务的调度，比如连接成功5s之后发送一包数据，就可以用到EventLoop的计划任务

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    ctx.Channel.EventLoop.Schedule(() =>
    {
        Console.WriteLine("delay 1s");
    }, new TimeSpan(1000));
    // 如果需要提前取消，可以调用Cancel方法
    IScheduledTask task = ctx.Channel.EventLoop.Schedule(() =>
    {
        Console.WriteLine("delay 1s");
    }, new TimeSpan(1000));
    tsak.Cancel();

  2. 一个任务引发后，会判断当前是否在需要处理这个任务的EventLoop中（程序知道自己目前在执行哪个线程，线程又跟EventLoop对应），如果在就直接执行该任务，如果不在该任务中，则任务入队稍后处理

  3. 永远不要把一个需要耗费长时间的任务放到EventLoop执行队列来执行，需要使用我们前面介绍的EventExecutor的方法。

Group

许多Channel对应一个EventLoop，但是EventLoop能分配给她的Channel个数是有限的，要处理可以扩展的无数个Channel就需要EventLoopGroup。他们的结构关系如下图：

我们之前讲过，Netty不仅能够完成NIO系统的搭建，也能通过一些简单的配置，变成OIO阻塞IO系统，阻塞IO的话，就不能多个Channel共享一个EventLoop了，就需要一个Channel分配一个EventLoop。总的来说，EventLoop跟线程的关系是不会改变的。

需要注意的是：

  1. 给Channel分配EventLoop的是EventLoopGroup。而他将尽量均衡的将Channel进行分配。

发表于 更新于 分类于 Valine：

DotNetty完全教程（六）

Excerpt

资源管理目的在处理数据的过程中，我们需要确保没有任何的资源泄漏。这时候我们就得很关心资源管理。引用计数的处理使用完ByteBuf之后，需要调整其引用计数以确保资源的释放内存内漏探测Netty提供了ResourceLeakDetector来检测内存泄漏，因为其是采样检测的，所以相关开销并不大。泄露日志检测级别手动释放消息ReferenceCountUtil.SafeRelea…

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

资源管理

目的

在处理数据的过程中，我们需要确保没有任何的资源泄漏。这时候我们就得很关心资源管理。

引用计数的处理

使用完ByteBuf之后，需要调整其引用计数以确保资源的释放

内存内漏探测

Netty提供了ResourceLeakDetector来检测内存泄漏，因为其是采样检测的，所以相关开销并不大。

  1. 泄露日志

  2. 检测级别

  3. 手动释放消息

    1
    ReferenceCountUtil.SafeRelease(this.Message);

  4. 分析SimpleChannelInboundHandler

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    public override void ChannelRead(IChannelHandlerContext ctx, object msg)
    {
        bool release = true;
        try
        {
            if (this.AcceptInboundMessage(msg))
            {
                I imsg = (I)msg;
                this.ChannelRead0(ctx, imsg);
            }
            else
            {
                release = false;
                ctx.FireChannelRead(msg);
            }
        }
        finally
        {
            if (autoRelease && release)
            {
                ReferenceCountUtil.Release(msg);
            }
        }
    }

由上面的源码可以看出，Read0事实上是Read的封装，区别就是Read0方法在调用的时候，消息一定是被释放了，这就是手动释放的例子。

发表于 更新于 分类于 Valine：

DotNetty完全教程（十一）

Excerpt

编码器和解码器定义编码器负责将应用程序可以识别的数据结构转化为可传输的数据流，解码器反之。对于应用程序来说，编码器操作出站数据，解码器操作入站数据。解码器和Handler解码器因为是处理入站数据的，所以继承了ChannelInBoundHandler.我们理解的时候可以认为解码器就是一种特殊的Handler，用于处理信息。解码器的类型ByteToMessageDecoderRepl…

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

编码器和解码器

定义

编码器负责将应用程序可以识别的数据结构转化为可传输的数据流，解码器反之。对于应用程序来说，编码器操作出站数据，解码器操作入站数据。

解码器和Handler

解码器因为是处理入站数据的，所以继承了ChannelInBoundHandler.我们理解的时候可以认为解码器就是一种特殊的Handler，用于处理信息。

解码器的类型

  • ByteToMessageDecoder
  • ReplayingDecoder
  • MessageToMessageDecoder

解码器实例

ByteToMessageDecoder

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
// ByteToMessageDecoder
public class ToIntDecoder : ByteToMessageDecoder
{
    protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)
    {
        if (input.ReadableBytes >= 4) output.Add(input.ReadInt());
    }
}
// 测试代码
[Fact]
public void TestIntDecoder()
{
    IByteBuffer buf = Unpooled.Buffer();
    for (int i = 0; i < 8; i++)
    {
        buf.WriteByte(i);
    }
    // 构建Channel
    EmbeddedChannel channel = new EmbeddedChannel(new ToIntDecoder());
    // 测试
    Assert.True(channel.WriteInbound(buf));
    Assert.True(channel.Finish());

    // 比如 0 1 2 3
    // 3*2^0+2*2^8+1*2^16+0*2^24
    Assert.Equal(66051, channel.ReadInbound<int>());

}

ReplayingDecoder

1
2
3
4
5
6
7
8
9
10
11
12
// 不需要判断ReadableBytes的ReplayingDecoder
public class ToIntDecoder2 : ReplayingDecoder<int>
{
    public ToIntDecoder2(int initialState) : base(initialState)
    {
    }

    protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)
    {
        output.Add(input.ReadInt());
    }
}

MessageToMessageDecoder

1
2
3
4
5
6
7
public class IntToStringDecoder : MessageToMessageDecoder<int>
{
    protected override void Decode(IChannelHandlerContext context, int message, List<object> output)
    {
        output.Add(message.ToString());
    }
}

更多解码器

  • LineBaseFrameDecoder 使用行尾控制符解析数据，可以把数据一行一行解析出来
  • HttpObjectDecoder HTTP解码器

编码器

根据我们之前的知识可以轻易的推导出，Encoder继承了ChannelOutBoundHandler

  • MessageToByteEncoder
  • MessageToMessageEncoder

编码器实例

MessageToByteEncoder

1
2
3
4
5
6
7
public class ShortToByteEncoder : MessageToByteEncoder<short>
{
    protected override void Encode(IChannelHandlerContext context, short message, IByteBuffer output)
    {
        output.WriteShort(message);
    }
}

MessageToMessageEncoder

1
2
3
4
5
6
7
public class IntToStringEncoder : MessageToMessageEncoder<int>
{
    protected override void Encode(IChannelHandlerContext context, int message, List<object> output)
    {
        output.Add(message.ToString());
    }
}

编解码器

MessageToMessageCodec,它拥有encode和decode两个方法，用于实现来回的转换数据，这种编解码器我们在后面实例的时候再举例说明。
这种编解码器可以把数据的转换，逆转换过程封装，但是同时他的缺点是，不如分开写重用方便。那我们就会想了，既然如此的话，为什么我们不能把一个编码器，一个解码器结合起来，作为一个编解码器呢？这样的话，编码器解码器分别可以重用，结合出来的编解码器也可以方便的使用 CombinedChannelDuplexHandler 就可以实现这样的作用。

1
2
3
4
5
// 提供结合的编解码器
public class CombinedCodec : CombinedChannelDuplexHandler<ToIntDecoder, ShortToByteEncoder>
{

}

发表于 更新于 分类于 Valine：

DotNetty完全教程（十）

Excerpt

单元测试EmbeddedChannel介绍EmbeddedChannel是专门为了测试ChannelHandler的传输。我们先看一下他的API用一张图来描述这样的一个模拟过程编写基于xUnit的单元测试新建一个xUnit工程 UnitTest新建一个用于测试EmbededChannel的工程 EmbededChannelTestEmbededChannelTest工程需要引用…

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

单元测试

EmbeddedChannel介绍

EmbeddedChannel是专门为了测试ChannelHandler的传输。我们先看一下他的API

用一张图来描述这样的一个模拟过程

编写基于xUnit的单元测试

  1. 新建一个xUnit工程 UnitTest
  2. 新建一个用于测试EmbededChannel的工程 EmbededChannelTest
  3. EmbededChannelTest工程需要引用DotNetty的类库，这里因为我们需要测试一个解码器，所以除了原先的Buffer Common Transport之外我们还需要引用Codecs
  4. xUnit工程需要引用EmbededChannelTest工程
  5. 在EmbededChannelTest工程之下新建FixedLengthFrameDecoder待测试类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
using DotNetty.Buffers;
using DotNetty.Codecs;
using DotNetty.Transport.Channels;
using System;
using System.Collections.Generic;
using System.Text;

namespace EmbededChannelTest
{
    public class FixedLengthFrameDecoder : ByteToMessageDecoder
    {
        private int _frameLength;
        public FixedLengthFrameDecoder(int frameLength)
        {
            if (frameLength <= 0)
            {
                throw new Exception("不合法的参数。");
            }
            _frameLength = frameLength;
        }
        protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)
        {
            // 解码器实现固定的帧长度
            while (input.ReadableBytes >= _frameLength)
            {
                IByteBuffer buf = input.ReadBytes(_frameLength);
                output.Add(buf);
            }
        }
    }
}

我们可以看到这个解码器将buffer中的字节流转化为每3个一帧。接下来我们需要编写测试类，我们在UnitTest工程下新建一个类，名字叫做UnitTester，编写如下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
using DotNetty.Buffers;
using DotNetty.Transport.Channels.Embedded;
using EmbededChannelTest;
using System;
using System.Collections.Generic;
using System.Text;
using Xunit;

namespace UnitTest
{
    public class UnitTester
    {
        [Fact]
        public void testFrameDecoder()
        {
            IByteBuffer buf = Unpooled.Buffer();
            for (int i = 0; i < 9; i++)
            {
                buf.WriteByte(i);
            }
            IByteBuffer input = buf.Duplicate();
            EmbeddedChannel channel = new EmbeddedChannel(new FixedLengthFrameDecoder(3));
            // 写数据
            // retain能够将buffer的引用计数加1，并且返回这个buffer本身
            Assert.True(channel.WriteInbound(input.Retain()));
            Assert.True(channel.Finish());
            // 读数据
            IByteBuffer read = channel.ReadInbound<IByteBuffer>();
            Assert.Equal(buf.ReadSlice(3), read);
            read.Release();

            read = channel.ReadInbound<IByteBuffer>();
            Assert.Equal(buf.ReadSlice(3), read);
            read.Release();

            read = channel.ReadInbound<IByteBuffer>();
            Assert.Equal(buf.ReadSlice(3), read);
            read.Release();

            Assert.Null(channel.ReadInbound<object>());
            buf.Release();
        }
    }
}

编写完成之后直接右键点击运行测试即可。同理我们可以测试用于出站数据的Encoder，这里不贴代码了，感兴趣的可以去工程中自己看源码进行学习。

发表于 更新于 分类于 Valine：

DotNetty完全教程（四）

Excerpt

ByteBufferNetty中ByteBuffer的介绍Netty 的数据处理 API 通过两个组件暴露——abstract class ByteBuf 和 interfaceByteBufHolderDotNetty中有AbstractByteBuffer IByteBuffer IByteBufferHolder优点：它可以被用户自定义的缓冲区类型扩展；通过内置的复合缓冲区…

ByteBuffer

Netty中ByteBuffer的介绍

Netty 的数据处理 API 通过两个组件暴露——abstract class ByteBuf 和 interface
ByteBufHolder

DotNetty中有AbstractByteBuffer IByteBuffer IByteBufferHolder

优点：

  • 它可以被用户自定义的缓冲区类型扩展；
  • 通过内置的复合缓冲区类型实现了透明的零拷贝；
  • 容量可以按需增长（类似于 JDK 的 StringBuilder）；
  • 在读和写这两种模式之间切换不需要调用 ByteBuffer 的 flip()方法；
  • 读和写使用了不同的索引；
  • 支持方法的链式调用；
  • 支持引用计数；
  • 支持池化

原理：

每一个ByteBuf都有两个索引，读索引和写索引，read和write会移动索引，set和get不会引动索引。

使用ByteBuf

  1. 堆缓冲区(使用数组的方式展示和操作数据)

使用支撑数组给ByteBuf提供快速的分配和释放的能力。适用于有遗留数据需要处理的情况。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public override void ChannelRead(IChannelHandlerContext ctx, object msg)
{
    IByteBuffer message = msg as IByteBuffer;
    // 检查是否有支撑数组
    if (message.HasArray)
    {
        // 获取数组
        byte[] array = message.Array;
        // 计算第一个字节的偏移
        int offset = message.ArrayOffset + message.ReaderIndex;
        // 获取可读字节数
        int length = message.ReadableBytes;
        // 调用方法，处理数据
        HandleArray(array, offset, length);
    }
    Console.WriteLine("收到信息：" + message.ToString(Encoding.UTF8));
    ctx.WriteAsync(message);
}
  1. 直接缓冲区
1
2
3
4
5
6
7
8
9
10
11
12
13
public override void ChannelRead(IChannelHandlerContext ctx, object msg)
{
    IByteBuffer message = msg as IByteBuffer;
    if (message.HasArray)
    {
        int length = message.ReadableBytes;
        byte[] array = new byte[length];
        message.GetBytes(message.ReaderIndex, array);
        HandleArray(array, 0, length);
    }
    Console.WriteLine("收到信息：" + message.ToString(Encoding.UTF8));
    ctx.WriteAsync(message);
}
  1. CompositeByteBuffer 复合缓冲区

如果要发送的命令是由两个ByteBuf拼接构成的，那么就需要复合缓冲区，比如Http协议中一个数据流由头跟内容构成这样的逻辑。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public override void ChannelRead(IChannelHandlerContext ctx, object msg)
{
    IByteBuffer message = msg as IByteBuffer;
    // 创建一个复合缓冲区
    CompositeByteBuffer messageBuf = Unpooled.CompositeBuffer();
    // 创建两个ByteBuffer
    IByteBuffer headBuf = Unpooled.CopiedBuffer(message);
    IByteBuffer bodyBuf = Unpooled.CopiedBuffer(message);
    // 添加到符合缓冲区中
    messageBuf.AddComponents(headBuf, bodyBuf);
    // 删除
    messageBuf.RemoveComponent(0);

    Console.WriteLine("收到信息：" + message.ToString(Encoding.UTF8));
    ctx.WriteAsync(message);
}

字节级操作

  1. 读取(不移动索引)
1
2
3
4
5
6
7
8
9
10
11
12
13
public override void ChannelRead(IChannelHandlerContext ctx, object msg)
{
    IByteBuffer message = msg as IByteBuffer;
    for (int i = 0; i < message.Capacity; i++)
    {
        // 如此使用索引访问不会改变读索引也不会改变写索引
        byte b = message.GetByte(i);
        Console.WriteLine(b);
    }

    Console.WriteLine("收到信息：" + message.ToString(Encoding.UTF8));
    ctx.WriteAsync(message);
}
  1. 丢弃可丢弃字节
    所谓可丢弃字节就是调用read方法之后，readindex已经移动过了的区域，这段区域的字节称为可丢弃字节。
1
message.DiscardReadBytes();

只有在内存十分宝贵需要清理的时候再调用这个方法，随便调用有可能会造成内存的复制，降低效率。
3. 读取所有可读字节（移动读索引）

1
2
3
4
while (message.IsReadable())
{
    Console.WriteLine(message.ReadByte());
}
  1. 写入数据
1
2
3
4
5
// 使用随机数填充可写区域
while (message.WritableBytes > 4)
{
    message.WriteInt(new Random().Next(0, 100));
}
  1. 管理索引
  • MarkReaderIndex ResetReaderIndex 标记和恢复读索引
  • MarkWriterIndex ResetWriterIndex 标记和恢复写索引
  • SetReaderIndex(int) SetWriterIndex(int) 直接移动索引
  • clear() 重置两个索引都为0，但是不会清除内容
  1. 查找
  • IndexOf()
  • 使用Processor
1
2
// 查找\r
message.ForEachByte(ByteProcessor.FindCR);
  1. 派生

派生的意思是创建一个新的ByteBuffer，这个ByteBuf派生于其他的ByteBuf，派生出来的子ByteBuf具有自己的读写索引，但是本质上指向同一个对象，这样就导致了改变一个，另一个也会改变。

  • duplicate()；
  • slice()；
  • slice(int, int)；
  • Unpooled.unmodifiableBuffer(…)；
  • order(ByteOrder)；
  • readSlice(int)。
  1. 复制
    复制不同于派生，会复制出一个独立的ByteBuf，修改其中一个不会改变另一个。
  • copy
  1. 释放
1
2
// 显式丢弃消息
ReferenceCountUtil.release(msg);
  1. 增加引用计数防止释放
1
ReferenceCountUtil.retain(message)
  1. 其他api
    在这里插入图片描述

ByteBufHolder

  1. 目的
    再数据处理的过程中不仅仅有字节数据内容本身，还会有一些附加信息，比如HTTP响应的状态码，Cookie等。给ByteBuf附加信息就要用到ByteBufHolder.
  2. API

管理ByteBuffer

  1. 按需分配 ByteBufAllocator

    注意分配是池化的，最大程度上降低分配和释放内存的开销。

    1
    2
    3
    4
    5
    6
    7
    // 获取Allocator
    // 1
    IChannelHandlerContext ctx = null;
    IByteBufferAllocator allocator = ctx.Allocator;
    // 2
    IChannel channel = null;
    allocator = channel.Allocator;

有两种ByteBufAllocator的实现：PooledByteBufAllocator和UnpooledByteBufAllocator，前者池化了ByteBuf的实例，极大限度的提升了性能减少了内存碎片。
2. Unpooled缓冲区
获取不到 ByteBufAllocator的引用的时候我们可以使用Unpooled工具类来操作ByteBuf。
在这里插入图片描述

  1. ByteBufUtil
    这个类提供了一些通用的API，都是静态的辅助方法，例如hexdump方法可以以十六进制的方式打印ByteBuf的内容。还有equal方法判断bytebuf是否相等。

引用计数

  1. 目的

    ByteBuf和ByteBufHolder都有计数的机制。引用计数都从1开始，如果计数大于0则不被释放，如果等于0就会被释放。它的目的是为了支持池化的实现，降低了内存分配的开销。

  2. 异常

    如果访问一个计数为0的对象就会引发IllegalReferenceCountException。

发表于 更新于 分类于 Valine：

DotNetty系列三：编码解码器,IdleStateHandler心跳机制

Excerpt

在上一节基础上，实现编码解码器。1.创建一个类库项目。用于实现编码解码器。编码器： public class CommonServerEncoder : MessageToByteEncoder<string> { protected override void Encode(IChannelHandlerContext context, s…

在上一节基础上，实现编码解码器。

1.创建一个类库项目。用于实现编码解码器。

编码器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class CommonServerEncoder : MessageToByteEncoder<string>    
{
    protected override void Encode(IChannelHandlerContext context, string message, IByteBuffer output)        
    {
        byte[] messageBytes = Encoding.UTF8.GetBytes(message);            
        IByteBuffer initialMessage = Unpooled.Buffer(messageBytes.Length);            
        initialMessage.WriteBytes(messageBytes);            
        output.WriteBytes(initialMessage);        
    }    
}

public class CommonClientEncoder : MessageToByteEncoder<string>    
{
    protected override void Encode(IChannelHandlerContext context, string message, IByteBuffer output)        
    {
        byte[] messageBytes = Encoding.UTF8.GetBytes(message);            
        IByteBuffer initialMessage = Unpooled.Buffer(messageBytes.Length);            
        initialMessage.WriteBytes(messageBytes);            
        output.WriteBytes(initialMessage);        
    }    
}

解码器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class CommonServerDecoder : ByteToMessageDecoder    
{
    protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)        
    {
        byte[] array = new byte[input.ReadableBytes];            
        input.GetBytes(input.ReaderIndex, array, 0, input.ReadableBytes);            
        input.Clear();            
        output.Add(array);        
    }    
}

public class CommonClientDecoder : ByteToMessageDecoder    
{
    protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)        
    {
        byte[] array = new byte[input.ReadableBytes];            
        input.GetBytes(input.ReaderIndex, array, 0, input.ReadableBytes);            
        input.Clear();            
        output.Add(array);        
    }    
}

2.服务端里添加：

                        //配置编码解码器
                        pipeline.AddLast(new CommonServerEncoder());
                        pipeline.AddLast(new CommonServerDecoder());

客户端里添加：

                        //配置编码解码器
                        pipeline.AddLast(new CommonClientEncoder());
                        pipeline.AddLast(new CommonClientDecoder());

3.服务端接收和发送：

1
2
3
4
5
6
7
8
9
public override void ChannelRead(IChannelHandlerContext context, object message)        
{
    if (message is byte[] o)            
    {                
        Console.WriteLine($"解码器方式，从客户端接收:{Encoding.UTF8.GetString(o)}:{DateTime.Now}");           
    }
    string msg = "服务端从客户端接收到内容后返回，我是服务端";            
    context.WriteAsync(msg);        
}

客户端接收和发送：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public override void ChannelActive(IChannelHandlerContext context)        
{           
    Console.WriteLine("我是客户端.");           
    Console.WriteLine($"连接至服务端{context}.");
    string message = "客户端1";            
    context.WriteAndFlushAsync(message);        
}

public override void ChannelRead(IChannelHandlerContext context, object message)        
{
    if (message is byte[] o)            
    {                
        Console.WriteLine($"解码器方式，从服务端接收:{Encoding.UTF8.GetString(o)}:{DateTime.Now}");            
    }        
}

实现了上一节一样的效果。

4.IdleStateHandler心跳机制:

4.1服务端添加IdleStateHandler心跳检测处理器,添加自定义处理Handler类实现userEventTriggered()方法作为超时事件的逻辑处理.

IdleStateHandler心跳检测每十五秒进行一次读检测，如果十五秒内ChannelRead()方法未被调用则触发一次userEventTrigger()方法.

                        // IdleStateHandler 心跳
                        //服务端为读IDLE
                        pipeline.AddLast(new IdleStateHandler(15, 0, 0));//第一个参数为读，第二个为写，第三个为读写全部

4.2服务端Handler重载UserEventTriggered：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private int lossConnectCount = 0;
public override void UserEventTriggered(IChannelHandlerContext context, object evt)        {            
    Console.WriteLine("已经15秒未收到客户端的消息了！");
    if (evt is IdleStateEvent eventState)            
    {
        if (eventState.State == IdleState.ReaderIdle)               
        {                   
            lossConnectCount++;if (lossConnectCount > 2)                    
            {                        
                Console.WriteLine("关闭这个不活跃通道！");                        
                context.CloseAsync();                   
            }                }            
    }
    else            
    {
        base.UserEventTriggered(context, evt);            
    }        
}

接收部分改为判断心跳：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public override void ChannelRead(IChannelHandlerContext context, object message)        
{
    if (message is byte[] o)            
    {                
        Console.WriteLine($"解码器方式，从客户端接收:{Encoding.UTF8.GetString(o)}:{DateTime.Now}");
        if (Encoding.UTF8.GetString(o).Contains("biubiu:"))                
        {
            string temp = "服务端接收到心跳连接";                    
            context.WriteAsync(temp);return;                
        }            
    }
    string msg = "服务端从客户端接收到内容后返回，我是服务端";            
    context.WriteAsync(msg);       
}

4.3客户端添加IdleStateHandler心跳检测处理器，并添加自定义处理Handler类实现userEventTriggered()方法作为超时事件的逻辑处理；

设定IdleStateHandler心跳检测每十秒进行一次写检测，如果十秒内write()方法未被调用则触发一次userEventTrigger()方法，实现客户端每十秒向服务端发送一次消息；

                        // IdleStateHandler 心跳
                        //客户端为写IDLE
                        pipeline.AddLast(new IdleStateHandler(0, 10, 0));//第一个参数为读，第二个为写，第三个为读写全部

4.4客户端Handler重载UserEventTriggered：

1
2
3
4
5
6
7
8
9
10
public override void UserEventTriggered(IChannelHandlerContext context, object evt)        {           
    Console.WriteLine("客户端循环心跳监测发送: " + DateTime.Now);
    if (evt is IdleStateEvent eventState)            
    {
        if (eventState.State == IdleState.WriterIdle)                
        {                    
            context.WriteAndFlushAsync($"biubiu:{DateTime.Now}");         
        }           
    }       
}

4.5实现效果：

5.群发：将客户端上下线通知，群发至所有客户端。只在服务端修改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
static volatile IChannelGroup groups;
public override void HandlerAdded(IChannelHandlerContext context)        
{            
    Console.WriteLine($"客户端{context}上线.");
                                                                          
    base.HandlerAdded(context);            
    IChannelGroup g = groups;if (g == null)            
    {
        lock (this) 
        {
            if (groups == null)                    
            {                        
                g = groups = new DefaultChannelGroup(context.Executor);
            }               
        }            
    }           
    g.Add(context.Channel);            
    groups.WriteAndFlushAsync($"欢迎{context.Channel.RemoteAddress}加入.");        
}

public override void HandlerRemoved(IChannelHandlerContext context)        
{            
    Console.WriteLine($"客户端{context}下线.");
    base.HandlerRemoved(context);            
    groups.Remove(context.Channel);            
    groups.WriteAndFlushAsync($"恭送{context.Channel.RemoteAddress}离开.");       
}

实现效果：

项目下载地址：项目下载

发表于 更新于 分类于 ， Valine：

一、下载链接https://portal.influxdata.com/downloads，选windows版

二、解压到安装盘，目录如下

三、修改conf文件，代码如下，直接复制粘贴（1.4.2版本），注意修改路径，带D盘的改为你的安装路径就好，一共三个，注意网上有配置admin进行web管理，但新版本配置文件里没有admin因为官方给删除了，需下载Chronograf，后文会介绍

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
### Welcome to the InfluxDB configuration file.

# The values in this file override the default values used by the system if
# a config option is not specified. The commented out lines are the configuration
# field and the default value used. Uncommenting a line and changing the value
# will change the value used at runtime when the process is restarted.

# Once every 24 hours InfluxDB will report usage data to usage.influxdata.com
# The data includes a random ID, os, arch, version, the number of series and other
# usage data. No data from user databases is ever transmitted.
# Change this option to true to disable reporting.
# reporting-disabled = false

# Bind address to use for the RPC service for backup and restore.
# bind-address = "127.0.0.1:8088"

###
### [meta]
###
### Controls the parameters for the Raft consensus group that stores metadata
### about the InfluxDB cluster.
###

[meta]
  # Where the metadata/raft database is stored
  dir = "D:/influxdb-1.4.2-1/meta"

  # Automatically create a default retention policy when creating a database.
    retention-autocreate = true

  # If log messages are printed for the meta service
    logging-enabled = true

###
### [data]
###
### Controls where the actual shard data for InfluxDB lives and how it is
### flushed from the WAL. "dir" may need to be changed to a suitable place
### for your system, but the WAL settings are an advanced configuration. The
### defaults should work for most systems.
###

[data]
  # The directory where the TSM storage engine stores TSM files.
  dir = "D:/influxdb-1.4.2-1/data"

  # The directory where the TSM storage engine stores WAL files.
  wal-dir = "D:/influxdb-1.4.2-1/wal"

  # The amount of time that a write will wait before fsyncing.  A duration
  # greater than 0 can be used to batch up multiple fsync calls.  This is useful for slower
  # disks or when WAL write contention is seen.  A value of 0s fsyncs every write to the WAL.
  # Values in the range of 0-100ms are recommended for non-SSD disks.
  # wal-fsync-delay = "0s"


  # The type of shard index to use for new shards.  The default is an in-memory index that is
  # recreated at startup.  A value of "tsi1" will use a disk based index that supports higher
  # cardinality datasets.
  # index-version = "inmem"

  # Trace logging provides more verbose output around the tsm engine. Turning
  # this on can provide more useful output for debugging tsm engine issues.
  # trace-logging-enabled = false

  # Whether queries should be logged before execution. Very useful for troubleshooting, but will
  # log any sensitive data contained within a query.
    query-log-enabled = true

  # Settings for the TSM engine

  # CacheMaxMemorySize is the maximum size a shard's cache can
  # reach before it starts rejecting writes.
  # Valid size suffixes are k, m, or g (case insensitive, 1024 = 1k).
  # Vaues without a size suffix are in bytes.
  # cache-max-memory-size = "1g"

  # CacheSnapshotMemorySize is the size at which the engine will
  # snapshot the cache and write it to a TSM file, freeing up memory
  # Valid size suffixes are k, m, or g (case insensitive, 1024 = 1k).
  # Values without a size suffix are in bytes.
  # cache-snapshot-memory-size = "25m"

  # CacheSnapshotWriteColdDuration is the length of time at
  # which the engine will snapshot the cache and write it to
  # a new TSM file if the shard hasn't received writes or deletes
  # cache-snapshot-write-cold-duration = "10m"

  # CompactFullWriteColdDuration is the duration at which the engine
  # will compact all TSM files in a shard if it hasn't received a
  # write or delete
  # compact-full-write-cold-duration = "4h"

  # The maximum number of concurrent full and level compactions that can run at one time.  A
  # value of 0 results in 50% of runtime.GOMAXPROCS(0) used at runtime.  Any number greater
  # than 0 limits compactions to that value.  This setting does not apply
  # to cache snapshotting.
  # max-concurrent-compactions = 0

  # The maximum series allowed per database before writes are dropped.  This limit can prevent
  # high cardinality issues at the database level.  This limit can be disabled by setting it to
  # 0.
  # max-series-per-database = 1000000

  # The maximum number of tag values per tag that are allowed before writes are dropped.  This limit
  # can prevent high cardinality tag values from being written to a measurement.  This limit can be
  # disabled by setting it to 0.
  # max-values-per-tag = 100000

###
### [coordinator]
###
### Controls the clustering service configuration.
###

[coordinator]
  # The default time a write request will wait until a "timeout" error is returned to the caller.
  # write-timeout = "10s"

  # The maximum number of concurrent queries allowed to be executing at one time.  If a query is
  # executed and exceeds this limit, an error is returned to the caller.  This limit can be disabled
  # by setting it to 0.
  # max-concurrent-queries = 0

  # The maximum time a query will is allowed to execute before being killed by the system.  This limit
  # can help prevent run away queries.  Setting the value to 0 disables the limit.
  # query-timeout = "0s"

  # The time threshold when a query will be logged as a slow query.  This limit can be set to help
  # discover slow or resource intensive queries.  Setting the value to 0 disables the slow query logging.
  # log-queries-after = "0s"

  # The maximum number of points a SELECT can process.  A value of 0 will make
  # the maximum point count unlimited.  This will only be checked every second so queries will not
  # be aborted immediately when hitting the limit.
  # max-select-point = 0

  # The maximum number of series a SELECT can run.  A value of 0 will make the maximum series
  # count unlimited.
  # max-select-series = 0

  # The maxium number of group by time bucket a SELECT can create.  A value of zero will max the maximum
  # number of buckets unlimited.
  # max-select-buckets = 0

###
### [retention]
###
### Controls the enforcement of retention policies for evicting old data.
###

[retention]
  # Determines whether retention policy enforcement enabled.
    enabled = true

  # The interval of time when retention policy enforcement checks run.
    check-interval = "30m"

###
### [shard-precreation]
###
### Controls the precreation of shards, so they are available before data arrives.
### Only shards that, after creation, will have both a start- and end-time in the
### future, will ever be created. Shards are never precreated that would be wholly
### or partially in the past.

[shard-precreation]
  # Determines whether shard pre-creation service is enabled.
    enabled = true

  # The interval of time when the check to pre-create new shards runs.
    check-interval = "10m"

  # The default period ahead of the endtime of a shard group that its successor
  # group is created.
    advance-period = "30m"

###
### Controls the system self-monitoring, statistics and diagnostics.
###
### The internal database for monitoring data is created automatically if
### if it does not already exist. The target retention within this database
### is called 'monitor' and is also created with a retention period of 7 days
### and a replication factor of 1, if it does not exist. In all cases the
### this retention policy is configured as the default for the database.

[monitor]
  # Whether to record statistics internally.
    store-enabled = true

  # The destination database for recorded statistics
    store-database = "_internal"

  # The interval at which to record statistics
    store-interval = "10s"

###
### [http]
###
### Controls how the HTTP endpoints are configured. These are the primary
### mechanism for getting data into and out of InfluxDB.
###

[http]
  # Determines whether HTTP endpoint is enabled.
    enabled = true

  # The bind address used by the HTTP service.
    bind-address = ":8086"

  # Determines whether user authentication is enabled over HTTP/HTTPS.
  # auth-enabled = false

  # The default realm sent back when issuing a basic auth challenge.
  # realm = "InfluxDB"

  # Determines whether HTTP request logging is enabled.
  # log-enabled = true

  # Determines whether detailed write logging is enabled.
  # write-tracing = false

  # Determines whether the pprof endpoint is enabled.  This endpoint is used for
  # troubleshooting and monitoring.
  # pprof-enabled = true

  # Determines whether HTTPS is enabled.
  # https-enabled = false

  # The SSL certificate to use when HTTPS is enabled.
  # https-certificate = "/etc/ssl/influxdb.pem"

  # Use a separate private key location.
  # https-private-key = ""

  # The JWT auth shared secret to validate requests using JSON web tokens.
  # shared-secret = ""

  # The default chunk size for result sets that should be chunked.
  # max-row-limit = 0

  # The maximum number of HTTP connections that may be open at once.  New connections that
  # would exceed this limit are dropped.  Setting this value to 0 disables the limit.
  # max-connection-limit = 0

  # Enable http service over unix domain socket
  # unix-socket-enabled = false

  # The path of the unix domain socket.
  # bind-socket = "/var/run/influxdb.sock"

  # The maximum size of a client request body, in bytes. Setting this value to 0 disables the limit.
  # max-body-size = 25000000


###
### [ifql]
###
### Configures the ifql RPC API.
###

[ifql]
  # Determines whether the RPC service is enabled.
  # enabled = true

  # Determines whether additional logging is enabled.
  # log-enabled = true

  # The bind address used by the ifql RPC service.
  # bind-address = ":8082"


###
### [subscriber]
###
### Controls the subscriptions, which can be used to fork a copy of all data
### received by the InfluxDB host.
###

[subscriber]
  # Determines whether the subscriber service is enabled.
  # enabled = true

  # The default timeout for HTTP writes to subscribers.
  # http-timeout = "30s"

  # Allows insecure HTTPS connections to subscribers.  This is useful when testing with self-
  # signed certificates.
  # insecure-skip-verify = false

  # The path to the PEM encoded CA certs file. If the empty string, the default system certs will be used
  # ca-certs = ""

  # The number of writer goroutines processing the write channel.
  # write-concurrency = 40

  # The number of in-flight writes buffered in the write channel.
  # write-buffer-size = 1000


###
### [[graphite]]
###
### Controls one or many listeners for Graphite data.
###

[[graphite]]
  # Determines whether the graphite endpoint is enabled.
  # enabled = false
  # database = "graphite"
  # retention-policy = ""
  # bind-address = ":2003"
  # protocol = "tcp"
  # consistency-level = "one"

  # These next lines control how batching works. You should have this enabled
  # otherwise you could get dropped metrics or poor performance. Batching
  # will buffer points in memory if you have many coming in.

  # Flush if this many points get buffered
  # batch-size = 5000

  # number of batches that may be pending in memory
  # batch-pending = 10

  # Flush at least this often even if we haven't hit buffer limit
  # batch-timeout = "1s"

  # UDP Read buffer size, 0 means OS default. UDP listener will fail if set above OS max.
  # udp-read-buffer = 0

  ### This string joins multiple matching 'measurement' values providing more control over the final measurement name.
  # separator = "."

  ### Default tags that will be added to all metrics.  These can be overridden at the template level
  ### or by tags extracted from metric
  # tags = ["region=us-east", "zone=1c"]

  ### Each template line requires a template pattern.  It can have an optional
  ### filter before the template and separated by spaces.  It can also have optional extra
  ### tags following the template.  Multiple tags should be separated by commas and no spaces
  ### similar to the line protocol format.  There can be only one default template.
  # templates = [
  #   "*.app env.service.resource.measurement",
  #   # Default template
  #   "server.*",
  # ]

###
### [collectd]
###
### Controls one or many listeners for collectd data.
###

[[collectd]]
  # enabled = false
  # bind-address = ":25826"
  # database = "collectd"
  # retention-policy = ""
  #
  # The collectd service supports either scanning a directory for multiple types
  # db files, or specifying a single db file.
  # typesdb = "/usr/local/share/collectd"
  #
  # security-level = "none"
  # auth-file = "/etc/collectd/auth_file"

  # These next lines control how batching works. You should have this enabled
  # otherwise you could get dropped metrics or poor performance. Batching
  # will buffer points in memory if you have many coming in.

  # Flush if this many points get buffered
  # batch-size = 5000

  # Number of batches that may be pending in memory
  # batch-pending = 10

  # Flush at least this often even if we haven't hit buffer limit
  # batch-timeout = "10s"

  # UDP Read buffer size, 0 means OS default. UDP listener will fail if set above OS max.
  # read-buffer = 0

  # Multi-value plugins can be handled two ways.
  # "split" will parse and store the multi-value plugin data into separate measurements
  # "join" will parse and store the multi-value plugin as a single multi-value measurement.
  # "split" is the default behavior for backward compatability with previous versions of influxdb.
  # parse-multivalue-plugin = "split"
###
### [opentsdb]
###
### Controls one or many listeners for OpenTSDB data.
###

[[opentsdb]]
  # enabled = false
  # bind-address = ":4242"
  # database = "opentsdb"
  # retention-policy = ""
  # consistency-level = "one"
  # tls-enabled = false
  # certificate= "/etc/ssl/influxdb.pem"

  # Log an error for every malformed point.
  # log-point-errors = true

  # These next lines control how batching works. You should have this enabled
  # otherwise you could get dropped metrics or poor performance. Only points
  # metrics received over the telnet protocol undergo batching.

  # Flush if this many points get buffered
  # batch-size = 1000

  # Number of batches that may be pending in memory
  # batch-pending = 5

  # Flush at least this often even if we haven't hit buffer limit
  # batch-timeout = "1s"

###
### [[udp]]
###
### Controls the listeners for InfluxDB line protocol data via UDP.
###

[[udp]]
  # enabled = false
  # bind-address = ":8089"
  # database = "udp"
  # retention-policy = ""

  # These next lines control how batching works. You should have this enabled
  # otherwise you could get dropped metrics or poor performance. Batching
  # will buffer points in memory if you have many coming in.

  # Flush if this many points get buffered
  # batch-size = 5000

  # Number of batches that may be pending in memory
  # batch-pending = 10

  # Will flush at least this often even if we haven't hit buffer limit
  # batch-timeout = "1s"

  # UDP Read buffer size, 0 means OS default. UDP listener will fail if set above OS max.
  # read-buffer = 0

###
### [continuous_queries]
###
### Controls how continuous queries are run within InfluxDB.
###

[continuous_queries]
  # Determines whether the continuous query service is enabled.
  # enabled = true

  # Controls whether queries are logged when executed by the CQ service.
  # log-enabled = true

  # Controls whether queries are logged to the self-monitoring data store.
  # query-stats-enabled = false

  # interval for how often continuous queries will be checked if they need to run
  # run-interval = "1s"

四、使配置生效并打开数据库连接，双击influxd.exe就好，然后双击influx.exe进行操作，网上有操作教程，注意操作数据库时不能关闭influxd.exe，我不知道为什么总有这么个提示：There was an error writing history file: open : The system cannot find the file specified.不过好像没啥影响

五、要使用web管理需要下载Chronograf，https://portal.influxdata.com/downloads第三个就是，下载完直接解压，双击exe程序，在浏览器输入http://localhost:8888/，一开始登录要账户密码，我都用admin就进去了

这个是查看建立的数据库

这个是查看数据库的数据

没了

发表于 更新于 分类于 Valine：

关于子仓库或者说是仓库共用，git官方推荐的工具是git subtree。 我自己也用了一段时间的git subtree，感觉比git submodule好用，但是也有一些缺点，在可接受的范围内。
所以对于仓库共用，在git subtree 与 git submodule之中选择的话，我推荐git subtree。

git subtree 可以实现一个仓库作为其他仓库的子仓库。

![](git subtree教程.assets/webp.webp)

image

使用git subtree 有以下几个原因：

  • 旧版本的git也支持(最老版本可以到 v1.5.2).
  • git subtree与git submodule不同，它不增加任何像.gitmodule这样的新的元数据文件.
  • git subtree对于项目中的其他成员透明，意味着可以不知道git subtree的存在.

当然，git subtree也有它的缺点，但是这些缺点还在可以接受的范围内：

  • 必须学习新的指令(如：git subtree).
  • 子仓库的更新与推送指令相对复杂。

git subtree的主要命令有：

1
2
3
4
5
6
git subtree add   --prefix=<prefix> <commit>
git subtree add   --prefix=<prefix> <repository> <ref>
git subtree pull  --prefix=<prefix> <repository> <ref>
git subtree push  --prefix=<prefix> <repository> <ref>
git subtree merge --prefix=<prefix> <commit>
git subtree split --prefix=<prefix> [OPTIONS] [<commit>]

准备

我们先准备一个仓库叫photoshop，一个仓库叫libpng，然后我们希望把libpng作为photoshop的子仓库。
photoshop的路径为https://github.com/test/photoshop.git，仓库里的文件有：

1
2
3
4
5
6
photoshop
    |
    |-- photoshop.c
    |-- photoshop.h
    |-- main.c
    \-- README.md

libPNG的路径为https://github.com/test/libpng.git，仓库里的文件有：

1
2
3
4
5
libpng
    |
    |-- libpng.c
    |-- libpng.h
    \-- README.md

以下操作均位于父仓库的根目录中。

在父仓库中新增子仓库

我们执行以下命令把libpng添加到photoshop中：

1
git subtree add --prefix=sub/libpng https://github.com/test/libpng.git master --squash

(--squash参数表示不拉取历史信息，而只生成一条commit信息。)

执行git status可以看到提示新增两条commit：

![](git subtree教程.assets/webp-1711075013770.webp)

image

git log查看详细修改：

![](git subtree教程.assets/webp-1711075016725.webp)

image

执行git push把修改推送到远端photoshop仓库，现在本地仓库与远端仓库的目录结构为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
photoshop
    |
    |-- sub/
    |   |
    |   \--libpng/
    |       |
    |       |-- libpng.c
    |       |-- libpng.h
    |       \-- README.md
    |
    |-- photoshop.c
    |-- photoshop.h
    |-- main.c
    \-- README.md

注意，现在的photoshop仓库对于其他项目人员来说，可以不需要知道libpng是一个子仓库。什么意思呢？
当你git clone或者git pull的时候，你拉取到的是整个photoshop(包括libpng在内，libpng就相当于photoshop里的一个普通目录)；当你修改了libpng里的内容后执行git push，你将会把修改push到photoshop上。
也就是说photoshop仓库下的libpng与其他文件无异。

从源仓库拉取更新

如果源libpng仓库更新了，photoshop里的libpng如何拉取更新？使用git subtree pull，例如：

1
git subtree pull --prefix=sub/libpng https://github.com/test/libpng.git master --squash

推送修改到源仓库

如果在photoshop仓库里修改了libpng，然后想把这个修改推送到源libpng仓库呢？使用git subtree push，例如：

1
git subtree push --prefix=sub/libpng https://github.com/test/libpng.git master

简化git subtree命令

我们已经知道了git subtree 的命令的基本用法，但是上述几个命令还是显得有点复杂，特别是子仓库的源仓库地址，特别不方便记忆。
这里我们把子仓库的地址作为一个remote，方便记忆：

1
git remote add -f libpng https://github.com/test/libpng.git

然后可以这样来使用git subtree命令：

1
2
3
git subtree add --prefix=sub/libpng libpng master --squash
git subtree pull --prefix=sub/libpng libpng master --squash
git subtree push --prefix=sub/libpng libpng master

发表于 更新于 分类于 ， Valine：

InfluxDB是一个开源的时序数据库，使用GO语言开发，特别适合用于处理和分析资源监控数据这种时序相关数据。而InfluxDB自带的各种特殊函数如求标准差，随机取样数据，统计数据变化比等，使数据统计和实时分析变得十分方便。在我们的容器资源监控系统中，就采用了InfluxDB存储cadvisor的监控数据。本文对InfluxDB的基本概念和一些特色功能做一个详细介绍，内容主要是翻译整理自官网文档，如有错漏，请指正。

这里说一下使用docker容器运行influxdb的步骤，物理机安装请参照官方文档。拉取镜像文件后运行即可，当前最新版本是1.3.5。启动容器时设置挂载的数据目录和开放端口。InfluxDB的操作语法InfluxQL与SQL基本一致，也提供了一个类似mysql-client的名为influx的CLI。InfluxDB本身是支持分布式部署多副本存储的，本文介绍都是针对的单节点单副本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17


f216e9be15bff545befecb30d1d275552026216a939cc20c042b17419e3bde31

root@f216e9be15bf:/
Connected to http:
InfluxDB shell version: 1.3.5
> create database cadvisor  
> show databases           
name: databases
name
----
_internal
cadvisor
> CREATE USER testuser WITH PASSWORD 'testpwd' 
> GRANT ALL PRIVILEGES ON cadvisor TO testuser 
> CREATE RETENTION POLICY "cadvisor_retention" ON "cadvisor" DURATION 30d REPLICATION 1 DEFAULT

influxdb里面有一些重要概念：database，timestamp，field key， field value， field set，tag key，tag value，tag set，measurement， retention policy ，series，point。结合下面的例子数据来说明这几个概念：

1
2
3
4
5
6
7
8
9
10
11
name: census
-————————————
time                     butterflies     honeybees     location   scientist
2015-08-18T00:00:00Z      12                23           1         langstroth
2015-08-18T00:00:00Z      1                 30           1         perpetua
2015-08-18T00:06:00Z      11                28           1         langstroth
2015-08-18T00:06:00Z      3                 28           1         perpetua
2015-08-18T05:54:00Z      2                 11           2         langstroth
2015-08-18T06:00:00Z      1                 10           2         langstroth
2015-08-18T06:06:00Z      8                 23           2         perpetua
2015-08-18T06:12:00Z      7                 22           2         perpetua

timestamp

既然是时间序列数据库，influxdb的数据都有一列名为time的列，里面存储UTC时间戳。

field key，field value，field set

butterflies和honeybees两列数据称为字段(fields)，influxdb的字段由field key和field value组成。其中butterflies和honeybees为field key，它们为string类型，用于存储元数据。

而butterflies这一列的数据12-7为butterflies的field value，同理，honeybees这一列的23-22为honeybees的field value。field value可以为string，float，integer或boolean类型。field value通常都是与时间关联的。

field key和field value对组成的集合称之为field set。如下：

1
2
3
4
5
6
7
8
butterflies = 12 honeybees = 23
butterflies = 1 honeybees = 30
butterflies = 11 honeybees = 28
butterflies = 3 honeybees = 28
butterflies = 2 honeybees = 11
butterflies = 1 honeybees = 10
butterflies = 8 honeybees = 23
butterflies = 7 honeybees = 22

在influxdb中，字段必须存在。注意，字段是没有索引的。如果使用字段作为查询条件，会扫描符合查询条件的所有字段值，性能不及tag。类比一下，fields相当于SQL的没有索引的列。

tag key，tag value，tag set

location和scientist这两列称为标签(tags)，标签由tag key和tag value组成。location这个tag key有两个tag value：1和2，scientist有两个tag value：langstroth和perpetua。tag key和tag value对组成了tag set，示例中的tag set如下：

1
2
3
4
location = 1, scientist = langstroth
location = 2, scientist = langstroth
location = 1, scientist = perpetua
location = 2, scientist = perpetua

tags是可选的，但是强烈建议你用上它，因为tag是有索引的，tags相当于SQL中的有索引的列。tag value只能是string类型 如果你的常用场景是根据butterflies和honeybees来查询，那么你可以将这两个列设置为tag，而其他两列设置为field，tag和field依据具体查询需求来定。

measurement

measurement是fields，tags以及time列的容器，measurement的名字用于描述存储在其中的字段数据，类似mysql的表名。如上面例子中的measurement为census。measurement相当于SQL中的表，本文中我在部分地方会用表来指代measurement。

retention policy

retention policy指数据保留策略，示例数据中的retention policy为默认的autogen。它表示数据一直保留永不过期，副本数量为1。你也可以指定数据的保留时间，如30天。

series

series是共享同一个retention policy，measurement以及tag set的数据集合。示例中数据有4个series，如下:

Arbitrary series number

Retention policy

Measurement

Tag set

series 1

autogen

census

location = 1,scientist = langstroth

series 2

autogen

census

location = 2,scientist = langstroth

series 3

autogen

census

location = 1,scientist = perpetua

series 4

autogen

census

location = 2,scientist = perpetua

point

point则是同一个series中具有相同时间的field set，points相当于SQL中的数据行。如下面就是一个point：

1
2
3
4
name: census
-----------------
time                  butterflies    honeybees   location    scientist
2015-08-18T00:00:00Z       1            30           1        perpetua

database

上面提到的结构都存储在数据库中，示例的数据库为my_database。一个数据库可以有多个measurement，retention policy， continuous queries以及user。influxdb是一个无模式的数据库，可以很容易的添加新的measurement，tags，fields等。而它的操作却和传统的数据库一样，可以使用类SQL语言查询和修改数据。

influxdb不是一个完整的CRUD数据库，它更像是一个CR-ud数据库。它优先考虑的是增加和读取数据而不是更新和删除数据的性能，而且它阻止了某些更新和删除行为使得创建和读取数据更加高效。

influxdb函数分为聚合函数，选择函数，转换函数，预测函数等。除了与普通数据库一样提供了基本操作函数外，还提供了一些特色函数以方便数据统计计算，下面会一一介绍其中一些常用的特色函数。

  • 聚合函数：FILL(), INTEGRAL()，SPREAD()， STDDEV()，MEAN(), MEDIAN()等。
  • 选择函数: SAMPLE(), PERCENTILE(), FIRST(), LAST(), TOP(), BOTTOM()等。
  • 转换函数: DERIVATIVE(), DIFFERENCE()等。
  • 预测函数：HOLT_WINTERS()。

先从官网导入测试数据（注：这里测试用的版本是1.3.1，最新版本是1.3.5）:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
$ curl https://s3.amazonaws.com/noaa.water-database/NOAA_data.txt -o NOAA_data.txt
$ influx -import -path=NOAA_data.txt -precision=s -database=NOAA_water_database
$ influx -precision rfc3339 -database NOAA_water_database
Connected to http://localhost:8086 version 1.3.1
InfluxDB shell 1.3.1
> show measurements
name: measurements
name
----
average_temperature
distincts
h2o_feet
h2o_pH
h2o_quality
h2o_temperature

> show series from h2o_feet;
key
---
h2o_feet,location=coyote_creek
h2o_feet,location=santa_monica

下面的例子都以官方示例数据库来测试，这里只用部分数据以方便观察。measurement为h2o_feet，tag key为location，field key有level description和water_level两个。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
> SELECT * FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z'
name: h2o_feet
time                 level description    location     water_level
----                 -----------------    --------     -----------
2015-08-18T00:00:00Z between 6 and 9 feet coyote_creek 8.12
2015-08-18T00:00:00Z below 3 feet         santa_monica 2.064
2015-08-18T00:06:00Z between 6 and 9 feet coyote_creek 8.005
2015-08-18T00:06:00Z below 3 feet         santa_monica 2.116
2015-08-18T00:12:00Z between 6 and 9 feet coyote_creek 7.887
2015-08-18T00:12:00Z below 3 feet         santa_monica 2.028
2015-08-18T00:18:00Z between 6 and 9 feet coyote_creek 7.762
2015-08-18T00:18:00Z below 3 feet         santa_monica 2.126
2015-08-18T00:24:00Z between 6 and 9 feet coyote_creek 7.635
2015-08-18T00:24:00Z below 3 feet         santa_monica 2.041
2015-08-18T00:30:00Z between 6 and 9 feet coyote_creek 7.5
2015-08-18T00:30:00Z below 3 feet         santa_monica 2.051

GROUP BY，FILL()

如下语句中GROUP BY time(12m),* 表示以每12分钟和tag(location)分组(如果是GROUP BY time(12m)则表示仅每12分钟分组，GROUP BY 参数只能是time和tag)。然后fill(200)表示如果这个时间段没有数据，以200填充，mean(field_key)求该范围内数据的平均值(注意：这是依据series来计算。其他还有SUM求和，MEDIAN求中位数)。LIMIT 7表示限制返回的point(记录数)最多为7条，而SLIMIT 1则是限制返回的series为1个。

注意这里的时间区间，起始时间为整点前包含这个区间第一个12m的时间，比如这里为 2015-08-17T:23:48:00Z，第一条为 2015-08-17T23:48:00Z <= t < 2015-08-18T00:00:00Z这个区间的location=coyote_creek的water_level的平均值，这里没有数据，于是填充的200。第二条为 2015-08-18T00:00:00Z <= t < 2015-08-18T00:12:00Z区间的location=coyote_creek的water_level平均值，这里为 （8.12+8.005）/ 2 = 8.0625，其他以此类推。

而GROUP BY time(10m)则表示以10分钟分组，起始时间为包含这个区间的第一个10m的时间，即 2015-08-17T23:40:00Z。默认返回的是第一个series，如果要计算另外那个series，可以在SQL语句后面加上 SOFFSET 1。

那如果时间小于数据本身采集的时间间隔呢，比如GROUP BY time(10s)呢？这样的话，就会按10s取一个点，没有数值的为空或者FILL填充，对应时间点有数据则保持不变。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
## GROUP BY time(12m)
> SELECT mean("water_level") FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(12m),* fill(200) LIMIT 7 SLIMIT 1
name: h2o_feet
tags: location=coyote_creek
time                 mean
----                 ----
2015-08-17T23:48:00Z 200
2015-08-18T00:00:00Z 8.0625
2015-08-18T00:12:00Z 7.8245
2015-08-18T00:24:00Z 7.5675

## GROUP BY time(10m)，SOFFSET设置为1
> SELECT mean("water_level") FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(10m),* fill(200) LIMIT 7 SLIMIT 1 SOFFSET 1
name: h2o_feet
tags: location=santa_monica
time                 mean
----                 ----
2015-08-17T23:40:00Z 200
2015-08-17T23:50:00Z 200
2015-08-18T00:00:00Z 2.09
2015-08-18T00:10:00Z 2.077
2015-08-18T00:20:00Z 2.041
2015-08-18T00:30:00Z 2.051

INTEGRAL(field_key, unit)

计算数值字段值覆盖的曲面的面积值并得到面积之和。测试数据如下：

1
2
3
4
5
6
7
8
9
10
11
> SELECT "water_level" FROM "h2o_feet" WHERE "location" = 'santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z'

name: h2o_feet
time                   water_level
----                   -----------
2015-08-18T00:00:00Z   2.064
2015-08-18T00:06:00Z   2.116
2015-08-18T00:12:00Z   2.028
2015-08-18T00:18:00Z   2.126
2015-08-18T00:24:00Z   2.041
2015-08-18T00:30:00Z   2.051

使用INTERGRAL计算面积。注意，这个面积就是这些点连接起来后与时间围成的不规则图形的面积，注意unit默认是以1秒计算，所以下面语句计算结果为3732.66=2.028*1800+分割出来的梯形和三角形面积。如果unit改为1分，则结果为3732.66/60 = 62.211。unit为2分，则结果为3732.66/120 = 31.1055。以此类推。

1
2
3
4
5
6
7
8
9
10
11
12
13
# unit为默认的1秒
> SELECT INTEGRAL("water_level") FROM "h2o_feet" WHERE "location" = 'santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z'
name: h2o_feet
time                 integral
----                 --------
1970-01-01T00:00:00Z 3732.66

# unit为1分
> SELECT INTEGRAL("water_level", 1m) FROM "h2o_feet" WHERE "location" = 'santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z'
name: h2o_feet
time                 integral
----                 --------
1970-01-01T00:00:00Z 62.211

SPREAD(field_key)

计算数值字段的最大值和最小值的差值。

1
2
3
4
5
6
7
8
> SELECT SPREAD("water_level") FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(12m),* fill(18) LIMIT 3 SLIMIT 1 SOFFSET 1
name: h2o_feet
tags: location=santa_monica
time                 spread
----                 ------
2015-08-17T23:48:00Z 18
2015-08-18T00:00:00Z 0.052000000000000046
2015-08-18T00:12:00Z 0.09799999999999986

STDDEV(field_key)

计算字段的标准差。influxdb用的是贝塞尔修正的标准差计算公式 ，如下：

  • mean=(v1+v2+…+vn)/n;
  • stddev = math.sqrt(
    ((v1-mean)2 + (v2-mean)2 + …+(vn-mean)2)/(n-1)
    )
1
2
3
4
5
6
7
8
9
> SELECT STDDEV("water_level") FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(12m),* fill(18) SLIMIT 1;
name: h2o_feet
tags: location=coyote_creek
time                 stddev
----                 ------
2015-08-17T23:48:00Z 18
2015-08-18T00:00:00Z 0.08131727983645186
2015-08-18T00:12:00Z 0.08838834764831845
2015-08-18T00:24:00Z 0.09545941546018377

PERCENTILE(field_key, N)

选取某个字段中大于N%的这个字段值。

如果一共有4条记录，N为10，则10%*4=0.4，四舍五入为0，则查询结果为空。N为20，则 20% * 4 = 0.8，四舍五入为1，选取的是4个数中最小的数。如果N为40，40% * 4 = 1.6，四舍五入为2，则选取的是4个数中第二小的数。由此可以看出N=100时，就跟MAX(field_key)是一样的，而当N=50时，与MEDIAN(field_key)在字段值为奇数个时是一样的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
> SELECT PERCENTILE("water_level",20) FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(12m)
name: h2o_feet
time                 percentile
----                 ----------
2015-08-17T23:48:00Z 
2015-08-18T00:00:00Z 2.064
2015-08-18T00:12:00Z 2.028
2015-08-18T00:24:00Z 2.041

> SELECT PERCENTILE("water_level",40) FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(12m)
name: h2o_feet
time                 percentile
----                 ----------
2015-08-17T23:48:00Z 
2015-08-18T00:00:00Z 2.116
2015-08-18T00:12:00Z 2.126
2015-08-18T00:24:00Z 2.051

SAMPLE(field_key, N)

随机返回field key的N个值。如果语句中有GROUP BY time()，则每组数据随机返回N个值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
> SELECT SAMPLE("water_level",2) FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z';
name: h2o_feet
time                 sample
----                 ------
2015-08-18T00:00:00Z 2.064
2015-08-18T00:12:00Z 2.028

> SELECT SAMPLE("water_level",2) FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z' GROUP BY time(12m);
name: h2o_feet
time                 sample
----                 ------
2015-08-18T00:06:00Z 2.116
2015-08-18T00:06:00Z 8.005
2015-08-18T00:12:00Z 7.887
2015-08-18T00:18:00Z 7.762
2015-08-18T00:24:00Z 7.635
2015-08-18T00:30:00Z 2.051

CUMULATIVE_SUM(field_key)

计算字段值的递增和。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
> SELECT CUMULATIVE_SUM("water_level") FROM "h2o_feet" WHERE time >= '2015-08-17T23:48:00Z' AND time <= '2015-08-18T00:30:00Z';
name: h2o_feet
time                 cumulative_sum
----                 --------------
2015-08-18T00:00:00Z 8.12
2015-08-18T00:00:00Z 10.184
2015-08-18T00:06:00Z 18.189
2015-08-18T00:06:00Z 20.305
2015-08-18T00:12:00Z 28.192
2015-08-18T00:12:00Z 30.22
2015-08-18T00:18:00Z 37.982
2015-08-18T00:18:00Z 40.108
2015-08-18T00:24:00Z 47.742999999999995
2015-08-18T00:24:00Z 49.78399999999999
2015-08-18T00:30:00Z 57.28399999999999
2015-08-18T00:30:00Z 59.334999999999994

DERIVATIVE(field_key, unit) 和 NON_NEGATIVE_DERIVATIVE(field_key, unit)

计算字段值的变化比。unit默认为1s，即计算的是1秒内的变化比。

如下面的第一个数据计算方法是 (2.116-2.064)/(6*60) = 0.00014..，其他计算方式同理。虽然原始数据是6m收集一次，但是这里的变化比默认是按秒来计算的。如果要按6m计算，则设置unit为6m即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
> SELECT DERIVATIVE("water_level") FROM "h2o_feet" WHERE "location" = 'santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z'
name: h2o_feet
time                 derivative
----                 ----------
2015-08-18T00:06:00Z 0.00014444444444444457
2015-08-18T00:12:00Z -0.00024444444444444465
2015-08-18T00:18:00Z 0.0002722222222222218
2015-08-18T00:24:00Z -0.000236111111111111
2015-08-18T00:30:00Z 0.00002777777777777842

> SELECT DERIVATIVE("water_level", 6m) FROM "h2o_feet" WHERE "location" = 'santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z'
name: h2o_feet
time                 derivative
----                 ----------
2015-08-18T00:06:00Z 0.052000000000000046
2015-08-18T00:12:00Z -0.08800000000000008
2015-08-18T00:18:00Z 0.09799999999999986
2015-08-18T00:24:00Z -0.08499999999999996
2015-08-18T00:30:00Z 0.010000000000000231

而DERIVATIVE结合GROUP BY time，以及mean可以构造更加复杂的查询，如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
> SELECT DERIVATIVE(mean("water_level"), 6m) FROM "h2o_feet" WHERE time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z' group by time(12m), *
name: h2o_feet
tags: location=coyote_creek
time                 derivative
----                 ----------
2015-08-18T00:12:00Z -0.11900000000000022
2015-08-18T00:24:00Z -0.12849999999999984

name: h2o_feet
tags: location=santa_monica
time                 derivative
----                 ----------
2015-08-18T00:12:00Z -0.00649999999999995
2015-08-18T00:24:00Z -0.015499999999999847

这个计算其实是先根据GROUP BY time求平均值，然后对这个平均值再做变化比的计算。因为数据是按12分钟分组的，而变化比的unit是6分钟，所以差值除以2(12/6)才得到变化比。如第一个值是 (7.8245-8.0625)/2 = -0.1190。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
> SELECT mean("water_level") FROM "h2o_feet" WHERE time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z' group by time(12m), *
name: h2o_feet
tags: location=coyote_creek
time                 mean
----                 ----
2015-08-18T00:00:00Z 8.0625
2015-08-18T00:12:00Z 7.8245
2015-08-18T00:24:00Z 7.5675

name: h2o_feet
tags: location=santa_monica
time                 mean
----                 ----
2015-08-18T00:00:00Z 2.09
2015-08-18T00:12:00Z 2.077
2015-08-18T00:24:00Z 2.0460000000000003

NON_NEGATIVE_DERIVATIVE与DERIVATIVE不同的是它只返回的是非负的变化比:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
> SELECT DERIVATIVE(mean("water_level"), 6m) FROM "h2o_feet" WHERE location='santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z' group by time(6m), *
name: h2o_feet
tags: location=santa_monica
time                 derivative
----                 ----------
2015-08-18T00:06:00Z 0.052000000000000046
2015-08-18T00:12:00Z -0.08800000000000008
2015-08-18T00:18:00Z 0.09799999999999986
2015-08-18T00:24:00Z -0.08499999999999996
2015-08-18T00:30:00Z 0.010000000000000231

> SELECT NON_NEGATIVE_DERIVATIVE(mean("water_level"), 6m) FROM "h2o_feet" WHERE location='santa_monica' AND time >= '2015-08-18T00:00:00Z' AND time <= '2015-08-18T00:30:00Z' group by time(6m), *
name: h2o_feet
tags: location=santa_monica
time                 non_negative_derivative
----                 -----------------------
2015-08-18T00:06:00Z 0.052000000000000046
2015-08-18T00:18:00Z 0.09799999999999986
2015-08-18T00:30:00Z 0.010000000000000231

4.1 基本语法

连续查询(CONTINUOUS QUERY，简写为CQ)是指定时自动在实时数据上进行的InfluxQL查询，查询结果可以存储到指定的measurement中。基本语法格式如下：

1
2
3
4
5
6
7
8
CREATE CONTINUOUS QUERY <cq_name> ON <database_name>
BEGIN
  <cq_query>
END

cq_query格式：
SELECT <function[s]> INTO <destination_measurement> FROM <measurement> [WHERE <stuff>] GROUP BY time(<interval>)[,<tag_key[s]>]

CQ操作的是实时数据，它使用本地服务器的时间戳、GROUP BY time()时间间隔以及InfluxDB预先设置好的时间范围来确定什么时候开始查询以及查询覆盖的时间范围。注意CQ语句里面的WHERE条件是没有时间范围的，因为CQ会根据GROUP BY time()自动确定时间范围。

CQ执行的时间间隔和GROUP BY time()的时间间隔一样，它在InfluxDB预先设置的时间范围的起始时刻执行。如果GROUP BY time(1h)，则单次查询的时间范围为 now()-GROUP BY time(1h)到 now()，也就是说，如果当前时间为17点，这次查询的时间范围为 16:00到16:59.99999。

下面看几个示例，示例数据如下，这是数据库transportation中名为bus_data的measurement，每15分钟统计一次乘客数和投诉数。数据文件bus_data.txt如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# DDL
CREATE DATABASE transportation

# DML
# CONTEXT-DATABASE: transportation 

bus_data,complaints=9 passengers=5 1472367600
bus_data,complaints=9 passengers=8 1472368500
bus_data,complaints=9 passengers=8 1472369400
bus_data,complaints=9 passengers=7 1472370300
bus_data,complaints=9 passengers=8 1472371200
bus_data,complaints=7 passengers=15 1472372100
bus_data,complaints=7 passengers=15 1472373000
bus_data,complaints=7 passengers=17 1472373900
bus_data,complaints=7 passengers=20 1472374800

导入数据，命令如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
root@f216e9be15bf:/# influx -import -path=bus_data.txt -precision=s
root@f216e9be15bf:/# influx -precision=rfc3339 -database=transportation
Connected to http://localhost:8086 version 1.3.5
InfluxDB shell version: 1.3.5
> select * from bus_data
name: bus_data
time                 complaints passengers
----                 ---------- ----------
2016-08-28T07:00:00Z 9          5
2016-08-28T07:15:00Z 9          8
2016-08-28T07:30:00Z 9          8
2016-08-28T07:45:00Z 9          7
2016-08-28T08:00:00Z 9          8
2016-08-28T08:15:00Z 7          15
2016-08-28T08:30:00Z 7          15
2016-08-28T08:45:00Z 7          17
2016-08-28T09:00:00Z 7          20

示例1 自动缩小取样存储到新的measurement中

对单个字段自动缩小取样并存储到新的measurement中。

1
2
3
4
CREATE CONTINUOUS QUERY "cq_basic" ON "transportation"
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(1h)
END

这个CQ的意思就是对bus_data每小时自动计算取样数据的平均乘客数并存储到 average_passengers中。那么在2016-08-28这天早上会执行如下流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
At 8:00 cq_basic 执行查询，查询时间范围 time >= '7:00' AND time < '08:00'.
cq_basic写入一条记录到 average_passengers:
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   7
At 9:00 cq_basic 执行查询，查询时间范围 time >= '8:00' AND time < '9:00'.
cq_basic写入一条记录到 average_passengers:
name: average_passengers
------------------------
time                   mean
2016-08-28T08:00:00Z   13.75

# Results
> SELECT * FROM "average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   7
2016-08-28T08:00:00Z   13.75

示例2 自动缩小取样并存储到新的保留策略（Retention Policy）中

1
2
3
4
CREATE CONTINUOUS QUERY "cq_basic_rp" ON "transportation"
BEGIN
  SELECT mean("passengers") INTO "transportation"."three_weeks"."average_passengers" FROM "bus_data" GROUP BY time(1h)
END

与示例1类似，不同的是保留的策略不是autogen，而是改成了three_weeks(创建保留策略语法 CREATE RETENTION POLICY "three_weeks" ON "transportation" DURATION 3w REPLICATION 1)。

1
2
3
4
5
6
> SELECT * FROM "transportation"."three_weeks"."average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   7
2016-08-28T08:00:00Z   13.75

示例3 使用后向引用(backreferencing)自动缩小取样并存储到新的数据库中

1
2
3
4
CREATE CONTINUOUS QUERY "cq_basic_br" ON "transportation"
BEGIN
  SELECT mean(*) INTO "downsampled_transportation"."autogen".:MEASUREMENT FROM /.*/ GROUP BY time(30m),*
END

使用后向引用语法自动缩小取样并存储到新的数据库中。语法 :MEASUREMENT 用来指代后面的表，而 /.*/则是分别查询所有的表。这句CQ的含义就是每30分钟自动查询transportation的所有表(这里只有bus_data一个表)，并将30分钟内数字字段(passengers和complaints)求平均值存储到新的数据库 downsampled_transportation中。

最终结果如下：

1
2
3
4
5
6
7
8
> SELECT * FROM "downsampled_transportation."autogen"."bus_data"
name: bus_data
--------------
time                   mean_complaints   mean_passengers
2016-08-28T07:00:00Z   9                 6.5
2016-08-28T07:30:00Z   9                 7.5
2016-08-28T08:00:00Z   8                 11.5
2016-08-28T08:30:00Z   7                 16

示例4 自动缩小取样以及配置CQ的时间范围

1
2
3
4
CREATE CONTINUOUS QUERY "cq_basic_offset" ON "transportation"
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(1h,15m)
END

与前面几个示例不同的是，这里的GROUP BY time(1h, 15m)指定了一个时间偏移，也就是说 cq_basic_offset执行的时间不再是整点，而是往后偏移15分钟。执行流程如下:

1
2
3
4
5
6
7
8
9
10
At 8:15 cq_basic_offset 执行查询的时间范围 time >= '7:15' AND time < '8:15'.
name: average_passengers
------------------------
time                   mean
2016-08-28T07:15:00Z   7.75
At 9:15 cq_basic_offset 执行查询的时间范围 time >= '8:15' AND time < '9:15'.
name: average_passengers
------------------------
time                   mean
2016-08-28T08:15:00Z   16.75

最终结果:

1
2
3
4
5
6
> SELECT * FROM "average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T07:15:00Z   7.75
2016-08-28T08:15:00Z   16.75

4.2 高级语法

InfluxDB连续查询的高级语法如下：

1
2
3
4
5
CREATE CONTINUOUS QUERY <cq_name> ON <database_name>
RESAMPLE EVERY <interval> FOR <interval>
BEGIN
  <cq_query>
END

与基本语法不同的是，多了RESAMPLE关键字。高级语法里CQ的执行时间和查询时间范围则与RESAMPLE里面的两个interval有关系。

高级语法中CQ以EVERY interval的时间间隔执行，执行时查询的时间范围则是FOR interval来确定。如果FOR interval为2h，当前时间为17:00，则查询的时间范围为15:00-16:59.999999。RESAMPLE的EVERY和FOR两个关键字可以只有一个。

示例的数据表如下，比之前的多了几条记录为了示例3和示例4的测试:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
name: bus_data
--------------
time                   passengers
2016-08-28T06:30:00Z   2
2016-08-28T06:45:00Z   4
2016-08-28T07:00:00Z   5
2016-08-28T07:15:00Z   8
2016-08-28T07:30:00Z   8
2016-08-28T07:45:00Z   7
2016-08-28T08:00:00Z   8
2016-08-28T08:15:00Z   15
2016-08-28T08:30:00Z   15
2016-08-28T08:45:00Z   17
2016-08-28T09:00:00Z   20

示例1 只配置执行时间间隔

1
2
3
4
5
CREATE CONTINUOUS QUERY "cq_advanced_every" ON "transportation"
RESAMPLE EVERY 30m
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(1h)
END

这里配置了30分钟执行一次CQ，没有指定FOR interval，于是查询的时间范围还是GROUP BY time(1h)指定的一个小时，执行流程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
At 8:00, cq_advanced_every 执行时间范围 time >= '7:00' AND time < '8:00'.
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   7
At 8:30, cq_advanced_every 执行时间范围 time >= '8:00' AND time < '9:00'.
name: average_passengers
------------------------
time                   mean
2016-08-28T08:00:00Z   12.6667
At 9:00, cq_advanced_every 执行时间范围 time >= '8:00' AND time < '9:00'.
name: average_passengers
------------------------
time                   mean
2016-08-28T08:00:00Z   13.75

需要注意的是，这里的 8点到9点这个区间执行了两次，第一次执行时时8:30，平均值是 (8+15+15）/ 3 = 12.6667，而第二次执行时间是9:00，平均值是 (8+15+15+17) / 4=13.75，而且最后第二个结果覆盖了第一个结果。InfluxDB如何处理重复的记录可以参见这个文档。

最终结果：

1
2
3
4
5
6
> SELECT * FROM "average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   7
2016-08-28T08:00:00Z   13.75

示例2 只配置查询时间范围

1
2
3
4
5
CREATE CONTINUOUS QUERY "cq_advanced_for" ON "transportation"
RESAMPLE FOR 1h
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(30m)
END

只配置了时间范围，而没有配置EVERY interval。这样，执行的时间间隔与GROUP BY time(30m)一样为30分钟，而查询的时间范围为1小时，由于是按30分钟分组，所以每次会写入两条记录。执行流程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
At 8:00 cq_advanced_for 查询时间范围：time >= '7:00' AND time < '8:00'.
写入两条记录。
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   6.5
2016-08-28T07:30:00Z   7.5
At 8:30 cq_advanced_for 查询时间范围：time >= '7:30' AND time < '8:30'.
写入两条记录。
name: average_passengers
------------------------
time                   mean
2016-08-28T07:30:00Z   7.5
2016-08-28T08:00:00Z   11.5
At 9:00 cq_advanced_for 查询时间范围：time >= '8:00' AND time < '9:00'.
写入两条记录。
name: average_passengers
------------------------
time                   mean
2016-08-28T08:00:00Z   11.5
2016-08-28T08:30:00Z   16

需要注意的是，cq_advanced_for每次写入了两条记录，重复的记录会被覆盖。

最终结果：

1
2
3
4
5
6
7
8
> SELECT * FROM "average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   6.5
2016-08-28T07:30:00Z   7.5
2016-08-28T08:00:00Z   11.5
2016-08-28T08:30:00Z   16

示例3 同时配置执行时间间隔和查询时间范围

1
2
3
4
5
CREATE CONTINUOUS QUERY "cq_advanced_every_for" ON "transportation"
RESAMPLE EVERY 1h FOR 90m
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(30m)
END

这里配置了执行间隔为1小时，而查询范围90分钟，最后分组是30分钟，每次插入了三条记录。执行流程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
At 8:00 cq_advanced_every_for 查询时间范围 time >= '6:30' AND time < '8:00'.
插入三条记录
name: average_passengers
------------------------
time                   mean
2016-08-28T06:30:00Z   3
2016-08-28T07:00:00Z   6.5
2016-08-28T07:30:00Z   7.5
At 9:00 cq_advanced_every_for 查询时间范围 time >= '7:30' AND time < '9:00'.
插入三条记录
name: average_passengers
------------------------
time                   mean
2016-08-28T07:30:00Z   7.5
2016-08-28T08:00:00Z   11.5
2016-08-28T08:30:00Z   16

最终结果：

1
2
3
4
5
6
7
8
9
> SELECT * FROM "average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T06:30:00Z   3
2016-08-28T07:00:00Z   6.5
2016-08-28T07:30:00Z   7.5
2016-08-28T08:00:00Z   11.5
2016-08-28T08:30:00Z   16

示例4 配置查询时间范围和FILL填充

1
2
3
4
5
CREATE CONTINUOUS QUERY "cq_advanced_for_fill" ON "transportation"
RESAMPLE FOR 2h
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(1h) fill(1000)
END

在前面值配置查询时间范围的基础上，加上FILL填充空的记录。执行流程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
At 6:00, cq_advanced_for_fill 查询时间范围：time >= '4:00' AND time < '6:00'，没有数据，不填充。

At 7:00, cq_advanced_for_fill 查询时间范围：time >= '5:00' AND time < '7:00'. 写入两条记录，没有数据的时间点填充1000。
------------------------
time                   mean
2016-08-28T05:00:00Z   1000          <------ fill(1000)
2016-08-28T06:00:00Z   3             <------ average of 2 and 4

[…] At 11:00, cq_advanced_for_fill 查询时间范围：time >= '9:00' AND time < '11:00'.写入两条记录，没有数据的点填充1000。
name: average_passengers
------------------------
2016-08-28T09:00:00Z   20            <------ average of 20
2016-08-28T10:00:00Z   1000          <------ fill(1000)     

At 12:00, cq_advanced_for_fill 查询时间范围：time >= '10:00' AND time < '12:00'。没有数据，不填充。

最终结果:

1
2
3
4
5
6
7
8
9
10
> SELECT * FROM "average_passengers"
name: average_passengers
------------------------
time                   mean
2016-08-28T05:00:00Z   1000
2016-08-28T06:00:00Z   3
2016-08-28T07:00:00Z   7
2016-08-28T08:00:00Z   13.75
2016-08-28T09:00:00Z   20
2016-08-28T10:00:00Z   1000

发表于 更新于 Valine：

gitea 安装

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
version: "2"
services:
  web:
    image: gitea/gitea
    volumes:
      - ./data:/data
    ports:
      - "3000:3000"
    restart: always
  db:
    image: mariadb:10
    restart: always
    environment:
      - MYSQL_ROOT_PASSWORD=xxxx
      - MYSQL_DATABASE=gitea
      - MYSQL_USER=gitea
      - MYSQL_PASSWORD=xxxxx
    volumes:
      - ./db/:/var/lib/mysql

drone 安装（server + runner）（请参考官方教程）

官方教程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
version: "3"

services: 
  server:
    image: drone/drone:1
    container_name: drone_server
    restart: always
    environment: 
      DRONE_GITEA_SERVER: https://git.xxxx.cn
      DRONE_GITEA_CLIENT_ID: "xxxxx"
      DRONE_GITEA_CLIENT_SECRET: "xxxxx"
      DRONE_RPC_SECRET: "xxxxx"
      DRONE_SERVER_HOST: ci.xxxxx.cn
      DRONE_SERVER_PROTO: https
      DRONE_USER_CREATE: username:xxxx,admin:true
    ports:
      - "3001:80"
    volumes: 
      - "./data:/var/lib/drone"

  runner:
    image: drone/drone-runner-docker
    container_name: drone_runner_docker
    restart: always
    environment: 
      DRONE_RPC_PROTO: https
      DRONE_RPC_HOST: ci.xxxx.cn
      DRONE_RPC_SECRET: xxxxx
      DRONE_RUNNER_CAPACITY: 5
      DRONE_UI_USERNAME: happyxhw
      DRONE_UI_PASSWORD: xxxxx
      DRONE_RUNNER_NAME: xxxxx
      DRONE_RUNNER_LABELS: role:xxxxx
    ports:
      - "8484:3000"
    volumes: 
      - "/var/run/docker.sock:/var/run/docker.sock"

流水线配置（参考示例项目 .drone.yml)

push, pull_request,merge：

  1. linter: 执行 golangci-lint
  2. test
  3. build
  4. notification
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
kind: pipeline
name: drone-golang-example-dev-build

# 指定编译平台
platform:
  os: linux
  arch: amd64

# 指定代码空间，git代码会被clone到指定的path
workspace:
  path: /drone/src

# type: docker, k8s, ssh, exec
type: docker
# 流水线
steps:
  - name: linter
    image: golang:latest  # 编译用的镜像，最好在runner主机上提前下载好，否则会很慢，墙，一生之敌
    pull: if-not-exists   # 默认always，指定if-not-exists或never可以大幅度提高速度，pull在中国很费时
    environment:
      GOPROXY: "https://goproxy.cn,direct"  # 懂的都懂
    volumes:                                # 缓存 go mod & pkg，可以大幅度提高速度，避免每次都下载
      - name: pkgdeps
        path: /go/pkg
    commands:
      - go get github.com/golangci/golangci-lint/cmd/golangci-lint@v1.27.0  # golang下好评最高的linter工具，强迫症最爱
      - golangci-lint run

  - name: test
    image: golang:latest
    pull: if-not-exists
    environment:
      GOPROXY: "https://goproxy.cn,direct"
    volumes:
      - name: pkgdeps
        path: /go/pkg
    commands:
      - go test -v ./...

  - name: build
    image: golang:latest
    pull: if-not-exists
    environment:
      GOPROXY: "https://goproxy.cn,direct"
    volumes:
      - name: pkgdeps
        path: /go/pkg
    commands:
      - CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build # 使用alpine务必要指定 GO_ENABLED=0

  - name: notification                                 # 钉钉通知，自行探索
    image: lddsb/drone-dingtalk-message
    pull: if-not-exists
    settings:
      token:
        from_secret: dingtalk_token                    # 敏感数据请使用 from_secret
      type: markdown
      secret:
        from_secret: dingtalk_secret
      sha_link: true
      message_color: true
    when:                                              # 即使流水线失败也能通知
      status:
        - success
        - failure

trigger:                # 触发条件，并关系
  branch:               # git 分支
    - develop
    - master
  event:                # 事件
    - push
    - pull_request

volumes:                # 挂载,持久化数据
  - name: pkgdeps
    host:
      path: /tmp/pkg

node:
  role: xxxx    # 指定标签为role:xxxx的runner执行流水线

Drone | Continuous Integration

![](基于 gitea + drone + docker 的 CI 流程实践/v2-f9e22bde7e07201d1568f2d952648a7f_b.jpg)

promote：

  1. linter: 执行 golangci-lint
  2. test
  3. build
  4. publish: 发布镜像
  5. scp: 复制部署文件到目标部署主机
  6. deploy: 拉取镜像，重启容器
  7. notification

注意：

我在docker-compose.yml中将镜像的tag设置为了环境变量${DOCKER_TAG}，在开发环境中我喜欢将DRONE_COMMIT作为镜像的tag；在生产环境中我喜欢将DRONE_TAG（tag 事件触发生产环境部署）作为镜像tag，这样的好处是可以快速回滚到指定版本。
此外也可以将部署流程自动化到git事件中（master merge后自动部署生产环境，不需要手动promote）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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

# 编译、部署开发镜像, 主动触发
kind: pipeline
name: drone-golang-example-dev-deploy

platform:
  os: linux
  arch: amd64

workspace:
  path: /drone/src

type: docker
steps:
  - name: linter
    image: golang:latest                            
    pull: if-not-exists                            
    environment:
      GOPROXY: "https://goproxy.cn,direct"          
    volumes: 
      - name: pkgdeps
        path: /go/pkg
    commands:
      - go get github.com/golangci/golangci-lint/cmd/golangci-lint@v1.27.0  
      - golangci-lint run

  - name: test
    image: golang:latest
    pull: if-not-exists
    environment:
      GOPROXY: "https://goproxy.cn,direct"
    volumes:
      - name: pkgdeps
        path: /go/pkg
    commands:
      - go test -v ./...

  - name: build
    image: golang:latest
    pull: if-not-exists
    environment:
      GOPROXY: "https://goproxy.cn,direct"
    volumes:
      - name: pkgdeps
        path: /go/pkg
    commands:
      - CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build

  - name: publish
    image: docker/compose
    pull: if-not-exists
    environment:
      USERNAME:
        from_secret: aliyun_registry_username
      PASSWORD:
        from_secret: aliyun_registry_password
    volumes:
      - name: dockersock
        path: /var/run/docker.sock
    commands:
      - tag=${DRONE_COMMIT}              # 使用 DRONE_COMMIT 作为tag
      - echo "publish"
      - echo dev_$tag
      - export DOCKER_TAG=dev_$tag
      - docker login --username=$USERNAME registry.cn-hangzhou.aliyuncs.com -p $PASSWORD
      - docker-compose build
      - docker-compose push
      -
  - name: scp
    image: appleboy/drone-scp
    pull: if-not-exists
    settings:
      host:
        from_secret: dev_host
      username:
        from_secret: dev_user
      password:
        from_secret: dev_password
      port: 22
      target: /home/xuhewen/drone-golang-example
      source:
        - docker-compose.yml

  - name: deploy
    image: appleboy/drone-ssh
    pull: if-not-exists
    settings:
      host:
        from_secret: dev_host
      username:
        from_secret: dev_user
      password:
        from_secret: dev_password
      port: 22
      script:
        - tag=${DRONE_COMMIT}               # 请预先在目标主机上执行 docker login
        - echo "deploy"
        - echo dev_$tag
        - cd /home/xuhewen/drone-golang-example
        - export DOCKER_TAG=dev_$tag
        - docker-compose pull
        - docker-compose stop
        - docker-compose up -d

  - name: notification
    image: lddsb/drone-dingtalk-message
    pull: if-not-exists
    settings:
      token:
        from_secret: dingtalk_token
      type: markdown
      secret:
        from_secret: dingtalk_secret
      sha_link: true
      message_color: true
    when: # 即使流水线失败也能通知
      status:
        - success
        - failure

volumes:
  - name: pkgdeps
    host:
      path: /tmp/pkg
  - name: dockersock
    host:
      path: /var/run/docker.sock

trigger:
  event:
    - promote
  target:
    - develop            # 指定部署环境： dev，stage，production

when:
  branch:
    - develop

node:
  role: xxxx

触发 promote：

![](基于 gitea + drone + docker 的 CI 流程实践/v2-c4a5ef1e6b0955dde4d9908d71ccd776_b.jpg)

Drone | Continuous Integration

![](基于 gitea + drone + docker 的 CI 流程实践/v2-296c70343b96b87b68340739ce602616_b.jpg)

成功部署：

![](基于 gitea + drone + docker 的 CI 流程实践/v2-88b6e0e8a778d8781224d27c5f66143a_b.jpg)

rollback：根据 DRONE_COMMIT 或 DRONE_TAG 快速回滚

  1. scp: 复制部署文件（clone时git已经切换到了需要回滚的版本）
  2. rollback: 根据 DRONE_COMMIT 或 DRONE_TAG 拉取指定tag的镜像进行回滚
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 # 根据commit id回滚
kind: pipeline
name: drone-golang-example-dev-rollback

platform:
  os: linux
  arch: amd64

workspace:
  path: /drone/src

type: docker
steps:
  - name: scp
    image: appleboy/drone-scp
    pull: if-not-exists
    settings:
      host:
        from_secret: dev_host
      username:
        from_secret: dev_user
      password:
        from_secret: dev_password
      port: 22
      target: /home/xuhewen/drone-golang-example
      source:
        - docker-compose.yml

  - name: rollback
    image: appleboy/drone-ssh
    pull: if-not-exists
    settings:
      host:
        from_secret: dev_host
      username:
        from_secret: dev_user
      password:
        from_secret: dev_password
      port: 22
      script:
        - tag=${DRONE_COMMIT}
        - echo "deploy"
        - echo dev_$tag
        - cd /home/xuhewen/drone-golang-example
        - export DOCKER_TAG=dev_$tag
        - docker-compose pull
        - docker-compose stop
        - docker-compose up -d

  - name: notification
    image: lddsb/drone-dingtalk-message
    pull: if-not-exists
    settings:
      token:
        from_secret: dingtalk_token
      type: markdown
      secret:
        from_secret: dingtalk_secret
      sha_link: true
      message_color: true
    when:
      status:
        - success
        - failure

volumes:
  - name: dockersock
    host:
      path: /var/run/docker.sock

trigger:
  event:
    - rollback
  target:
    - develop              # 指定部署环境： dev，stage，production

when:
  branch:
    - develop

node:
  role: xxxx

Drone | Continuous Integration

![](基于 gitea + drone + docker 的 CI 流程实践/v2-e7601b0c866cc762b64390f54a1751fb_b.jpg)

成功回滚到指定版本（’hello, to happyxhw’ 回滚到 ‘hi, from happyxhw’）：

![](基于 gitea + drone + docker 的 CI 流程实践/v2-b8f3556bccb50089bb41de74bef436a0_b.jpg)

secrets:

![](基于 gitea + drone + docker 的 CI 流程实践/v2-6e17660da65691d41816a75557bd7e64_b.jpg)

镜像：

![](基于 gitea + drone + docker 的 CI 流程实践/v2-50cdbdf6a180f54a2d2b79dca451e781_b.jpg)

补充：

上面我只写出了开发环境中的CI流程，大家可以自行补充，比如加入预发布、生产环境的编译、部署和回滚流程，加入 master 分支、tag 事件的触发流程，具体请参考 drone 官方文档。