rabbmitmq连接池[已过生产]

源码地址https://gitee.com/tym_hmm/rabbitmq-pool-go

rabbitmq 连接池channel复用

开发语言 golang 依赖库

go get -u gitee.com/tym_hmm/rabbitmq-pool-go

go get -u gitee.com/tym_hmm/rabbitmq-pool-go

已在线上生产环镜运行， 5200W请求 qbs 3000 时，连接池显示无压力
rabbitmq部署为线上集群

功能说明

自定义连接池大小及最大处理channel数
消费者底层断线自动重连
底层使用轮循方式复用tcp
生产者每个tcp对应一个channel,防止channel写入阻塞造成内存使用过量
支持rabbitmq exchangeType
默认值

名称	说明
tcp最大连接数	5
生产者消费发送失败最大重试次数	5
消费者最大channel信道数(每个连接自动平分)	100(每个tcp10个)

使用

初始化

var oncePool sync.Once
var instanceRPool *kelleyRabbimqPool.RabbitPool
func initrabbitmq() *kelleyRabbimqPool.RabbitPool {oncePool.Do(func() {//初始化生产者instanceRPool = kelleyRabbimqPool.NewProductPool()//初始化消费者instanceConsumePool = kelleyRabbimqPool.NewConsumePool()err := instanceRPool.Connect("192.168.1.202", 5672, "guest", "guest")if err != nil {fmt.Println(err)}})return instanceRPool
}

生产者

var wg sync.WaitGroup
for i:=0;i<100000; i++ {wg.Add(1)go func(num int) {defer wg.Done()data:=kelleyRabbimqPool.GetRabbitMqDataFormat("testChange5", kelleyRabbimqPool.EXCHANGE_TYPE_TOPIC, "textQueue5", "/", fmt.Sprintf("这里是数据%d", num))_=instanceRPool.Push(data)}(i)
}
wg.Wait()

消费者

可定义多个消息者事件, 不通交换机, 队列, 路由

每个事件独立

nomrl := &rabbitmq.ConsumeReceive{
#定义消费者事件ExchangeName: "testChange31",//队列名称ExchangeType: kelleyRabbimqPool.EXCHANGE_TYPE_DIRECT,Route:        "",QueueName:    "testQueue31",IsTry:true,//是否重试MaxReTry: 5,//最大重试次数EventFail: func(code int, e error, data []byte) {fmt.Printf("error:%s", e)},/**** 参数说明* @param data []byte 接收的rabbitmq数据* @param header map[string]interface{} 原rabbitmq header* @param retryClient RabbitmqPool.RetryClientInterface 自定义重试数据接口，重试需return true 防止数据重复提交***/EventSuccess: func(data []byte, header map[string]interface{},retryClient kelleyRabbimqPool.RetryClientInterface)bool {//如果返回true 则无需重试fmt.Printf("data:%s\n", string(data))return true},
}
instanceConsumePool.RegisterConsumeReceive(nomrl)err := instanceConsumePool.RunConsume()
if err != nil {fmt.Println(err)
}

错误码说明

错误码为

生产者push时返回的 *RabbitMqError

消费者事件监听回返的 code

错误码	说明
501	生产者发送超过最大重试次数
502	获取信道失败, 一般为认道队列数用尽
503	交换机/队列/绑定失败
504	连接失败
506	信道创建失败
507	超过最大重试次数

rabbmitmq连接池[已过生产]相关推荐

IIS应用程序池相关问题及连接池已满的解决方法
关于应用程序池在 IIS 6.0 中,引入了应用程序池,应用程序池是将一个或多个应用程序链接到一个或多个工作进程集合的配置.因为应用程序池中的应用程序与其他应用程序被工作进程边界分隔 ...
mybatis 原理_深入理解MyBatis原理 MyBatis数据源与连接池
点击上方"程序开发者社区"关注,选择"设为星标" 第一时间送达实用干货对于ORM框架而言,数据源的组织是一个非常重要的一部分,这直接影响到框架的性能问题.本文 ...
[数据库druid连接池实现]--Java版本
前言之前有介绍一篇关于自定义的数据库链接池实现,但是质量并不高.而且遭到博友的一脸嫌弃!确实,拿出来的东西就应该保证是正确的,或者质量高,否则很容易就误导了他人,显得TMD也不道德,所以今天专门重写 ...
《深入理解mybatis原理三》 Mybatis数据源与连接池
对于ORM框架而言,数据源的组织是一个非常重要的一部分,这直接影响到框架的性能问题.本文将通过对MyBatis框架的数据源结构进行详尽的分析,并且深入解析MyBatis的连接池. 本文首先会讲述MyB ...
连接池和 Timeout expired异常
转自:博客园宁静.致远:http://www.cnblogs.com/zhangzhu/archive/2013/10/10/3361197.html 异常信息: MySql.Data.MySqlCl ...
连接池超时配置_HttpClient连接池的一些思考
前言使用apache的httpclient进行http的交互处理已经很长时间了,而httpclient实例则使用了http连接池,想必大家也没有关心过连接池的管理.事实上,通过分析httpclien ...
连接池和 Timeout expired异常【转】
异常信息: MySql.Data.MySqlClient.MySqlException (0x80004005): error connecting: Timeout expired. The tim ...
[译]连接池和 Timeout expired异常
原文标题:Connection Pooling and the "Timeout expired" exception FAQ 来源:http://blogs.msdn.com/a ...
HikariCP连接池教程
HikariCP连接池已作为当前流行的SpringBoot框架默认连接池,那如何在一个非框架服务使用HikariCP? HikariCP官方地址:https://github.com/brettwoo ...

rabbmitmq连接池[已过生产]

rabbitmq 连接池channel复用

功能说明

使用

rabbmitmq连接池[已过生产]相关推荐

最新文章

热门文章