Java&Go高性能队列之channel性能测试

Java&Go高性能队列之channel性能测试

之前写了两篇Java的高性能队列性能测试实践文章,发现了一些比较通用的规律,总体上Disruptor性能是要领先LinkedBlockingQueue的。先回顾一下Java&Go高性能队列之LinkedBlockingQueue性能测试Java&Go高性能队列之Disruptor性能测试

那么理论上性能更高的Go语言中的channel (下文中的也称为队列)性能如何呢,下面我将对它进行同样的性能测试。

测试场景设计的思路与前两篇文章相同,通过三个场景对变量的修改进行对比压测,包括不限于数量、大小、goroutine的数量。

结论

总体来说channel性能还是在性能足够高,完全满足现在压测需求。总结起来几点比较通用的参考:

  • Go语言channel性能足够好,首先与生产者生产能力,工作中需要提升生产能力
  • 消息体越小越好
  • channel的size长度并不重要
  • 创建请求对象上fasthttp.Request居然还不如net/http.Request,可能跟没有release掉有关。

简介

Go 语言中的通道(channel)是一种特殊的类型。通道像一个传送带或者队列,总是遵循先入先出(First In First Out)的规则,保证收发数据的顺序。每一个通道都是一个具体类型的导管,也就是声明channel的时候需要为其指定元素类型。如果说goroutine是Go程序并发的执行体,channel就是它们之间的连接。channel是可以让一个goroutine发送特定值到另一个goroutine的通信机制。

在我查资料的过程中,发现Go语言在锁解决(多协程/多goroutine安全)的层面有很多很优秀的功能,显示在不同场景下会比channel性能更高。但是我在阅读goreplay源码的过程中,看到的更多还是channel的实践。等我逐步提高自己Go语言多协程编程能力之后再来测试其他实现。

测试结果

这里性能只记录每毫秒处理消息(对象)个数作为评价性能的唯一标准。在我测试Disruptor框架的过程中,发现这个单一指标有点有失偏颇,后续如果还有下一轮的测试的话,我再优化这个地方。

数据说明

这里我用了三种net/http中的Request,创建方法均使用原生API,为了区分大小的区别,我会响应增加一些header和URL长度。

小对象:

				get, _ := http.NewRequest("GET", base.Empty, nil)

中对象:

				get,_ := http.NewRequest("GET",base.Empty, nil)
				get.Header.Add("token", token)
				get.Header.Add("Connection", base.Connection_Alive)
				get.Header.Add("User-Agent", base.UserAgent)

大对象:

				get,_ := http.NewRequest("GET",base.Empty, nil)
				get.Header.Add("token", token)
				get.Header.Add("token1", token)
				get.Header.Add("token2", token)
				get.Header.Add("token3", token)
				get.Header.Add("token4", token)
				get.Header.Add("token5", token)
				get.Header.Add("Connection", base.Connection_Alive)
				get.Header.Add("User-Agent", base.UserAgent)

生产者

对象大小 队列长度 (百万) 线程数 速率(/ms)
1 1 2173
1 5 4385
1 10 4273
5 1 2048
10 1 1964
1 1 831
1 5 1792
1 10 2450
1 20 2481
5 1 898
10 1 848
0.5 1 865
0.5 5 1760
1 1 560
1 5 1633
1 10 2092
0.5 1 571
0.5 5 1677
0.5 10 1984

针对net/http中的Request消息体结论如下: 1. 长度在50万 ~ 1000万没有明显差异 2. 生产者越多越好(20以内,再多增益效果不明显) 3. 消息体尽可能小

消费者

对象大小 队列长度 (百万) 线程数 速率(/ms)
1 1 2092
1 5 3322
1 10 3472
1 20 3246
2 1 2030
2 5 4081
5 1 2150
5 5 3980
1 1 1851
1 5 3460
1 10 3289
1 20 2832
2 1 1733
2 5 2652
1 1 1697
1 5 2564
1 10 3436
0.5 1 2032
0.5 5 3311
0.5 10 3597

针对net/http中的Request消息体结论如下: 1. 长度在50万 ~ 500万没有明显差异 2. 消费者10 ~ 20以内到达峰值 2. 消息体尽可能小

消费者并发越多越好,这个在实际工作中消费者消费消息会有耗时,消费者goroutine会很多,要根据实际情况设置消费者数量,或者在压测过程中灵活增减消费者数量,这点跟Disruptor不同。

生产者 & 消费者

这里的线程数指的是生产者或者消费者的数量,总体线程数是此数值的2倍。

对象大小 次数 (百万) 线程数 速率(/ms)
1 1 0.1
1 1 0.2
1 1 0.5
1 5 0.1
1 10 0.1
2 1 0.1
2 1 0.2
2 5 0.2
5 5 0.1
5 10 0.1
1 1 0.1
1 1 0.2
1 5 0.2
1 10 0.2
2 1 0.2
2 5 0.2
2 10 0.2
2 15 0.2
1 1 0.1
1 1 0.2
1 5 0.2
1 10 0.2
2 1 0.2
2 5 0.2
2 10 0.2

针对net/http中的Request消息体结论如下: 1. 消息队列积累消息对性能影响不大 2. 消费次数越多,性能反而有点下降,应该是生产者速率不足导致 3. 消息体尽可能小,不过性能下降不多

测试用例

总体代码逻辑与Java和Groovy用例一样,有几处差别如下:

这里我用了sync.WaitGroup代替了java.util.concurrent.CountDownLatch,暂时没有找到合适的功能替换java.util.concurrent.CyclicBarrier,经过测试并不影响测试结果,所以略过此项。

Go语言的channel有个先天的优势,就是必需得设置size,相当于提前分配内存了。这点是我之前没想到的,当我回去复测LinkedBlockingQueue的时候发现并没有明显的性能差异,对于测试结果影响可忽略。

我还用了atomic.AddInt32解决计数安全的问题,这里不多分享了,有兴趣可以搜一下官方文档学习使用。

生产者场景

func TestQueue(t *testing.T) {
	var index int32 = 0
	rs := make(chan *http.Request, total+10000)
	var group sync.WaitGroup
	group.Add(threadNum)
	milli := futil.Milli()
	funtester := func() {
		go func() {
			for {
				l := atomic.AddInt32(&index, 1)
				if l%piece == 0 {
					m := futil.Milli()
					log.Println(m - milli)
					milli = m
				}
				if l > total {
					break
				}
				get := getRequest()
				rs <- get
			}
			group.Done()
		}()
	}
	start := futil.Milli()
	for i := 0; i < threadNum; i++ {
		funtester()
	}
	group.Wait()
	end := futil.Milli()

	log.Println(atomic.LoadInt32(&index))
	log.Printf("平均每毫秒速率%d", total/(end-start))
}

消费者场景


func TestConsumer(t *testing.T) {
	rs := make(chan *http.Request, total+10000)
	var group sync.WaitGroup
	group.Add(10)
	funtester := func() {
		go func() {
			for {
				if len(rs) > total {
					break
				}
				get := getRequest()

				rs <- get
			}
			group.Done()
		}()
	}
	for i := 0; i < 10; i++ {
		funtester()
	}
	group.Wait()
	log.Printf("造数据完成! 总数%d", len(rs))
	totalActual := int64(len(rs))
	var conwait sync.WaitGroup
	conwait.Add(threadNum)
	consumer := func() {
		go func() {
		FUN:
			for {
				select {
				case <-rs:
				case <-time.After(10 * time.Millisecond):
					break FUN
				}
			}
			conwait.Done()
		}()
	}
	start := futil.Milli()
	for i := 0; i < threadNum; i++ {
		consumer()
	}
	conwait.Wait()
	end := futil.Milli()
	log.Printf("平均每毫秒速率%d", totalActual/(end-start))

}

生产者 & 消费者 场景

这里我引入了另外一个变量:初始队列长度length,用例运行之前将队列按照这个长度进行单线程填充。

func TestConsumer(t *testing.T) {
	rs := make(chan *http.Request, total+10000)
	var group sync.WaitGroup
	group.Add(10)
	funtester := func() {
		go func() {
			for {
				if len(rs) > total {
					break
				}
				get := getRequest()

				rs <- get
			}
			group.Done()
		}()
	}
	for i := 0; i < 10; i++ {
		funtester()
	}
	group.Wait()
	log.Printf("造数据完成! 总数%d", len(rs))
	totalActual := int64(len(rs))
	var conwait sync.WaitGroup
	conwait.Add(threadNum)
	consumer := func() {
		go func() {
		FUN:
			for {
				select {
				case <-rs:
				case <-time.After(10 * time.Millisecond):
					break FUN
				}
			}
			conwait.Done()
		}()
	}
	start := futil.Milli()
	for i := 0; i < threadNum; i++ {
		consumer()
	}
	conwait.Wait()
	end := futil.Milli()
	log.Printf("平均每毫秒速率%d", totalActual/(end-start))

}

生产对象

func getRequest() *http.Request {
	//get, _ := http.NewRequest("GET", base.Empty, nil)

	//get,_ := http.NewRequest("GET",url, nil)
	//get.Header.Add("token", token)
	//get.Header.Add("Connection", base.Connection_Alive)
	//get.Header.Add("User-Agent", base.UserAgent)

	get,_ := http.NewRequest("GET",url, nil)
	get.Header.Add("token", token)
	get.Header.Add("token1", token)
	get.Header.Add("token2", token)
	get.Header.Add("token3", token)
	get.Header.Add("token4", token)
	get.Header.Add("token5", token)
	get.Header.Add("Connection", base.Connection_Alive)
	get.Header.Add("User-Agent", base.UserAgent)

	return get
}

基准测试

下面是我使用FunTester(Go语言版本)性能测试框架对三种消息对象的生产代码进行的测试结果。没想到net/http的性能还不如Java的,有点奇怪。

测试对象 线程数 个数(百万) 速率(/ms)
1 1 3311
5 1 3725
5 5 7382
1 1 1132
5 1 1205
5 5 3064
1 1 732
5 1 738
5 5 2061

下面是fasthttp.Request的基准测试结果:

测试对象 线程数 个数(百万) 速率(/ms)
1 1 2673
5 1 2881
5 5 4983
1 1 1197
5 1 1137
5 5 2784
1 1 621
5 1 631
5 5 1438

fasthttp.Request居然还不如net/http.Request,有点奇怪。

测试用例如下:

// TestBase
// @Description: 基准测试
// @param t
func TestBase(t *testing.T) {
	execute.ExecuteRoutineTimes(func() {
		getRequest()
	},total,threadNum)
}

Have Fun ~ Tester !