代码拉取完成,页面将自动刷新
本库基于 https://gitee.com/tym_hmm/rabbitmq-pool-go 开发,拓展了一些特性,增加了发送失败的监听器,重发机制的优化,生产更加稳定
开发语言 golang 依赖库
go get -u gitee.com/aohanhongzhi/go-rabbitmq-pool
已在线上生产环镜运行, 5200W请求 qbs 3000 时, 连接池显示无压力
rabbitmq部署为线上集群
名称 | 说明 |
---|---|
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()
//使用默认虚拟host "/"
err := instanceRPool.Connect("192.168.1.202", 5672, "guest", "guest")
//使用自定义虚
//err:=instanceConsumePool.ConnectVirtualHost("192.168.1.202", 5672, "guest", "guest", "/testHost")
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,//是否重试
IsAutoAck: false, //是否自动确认消息
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 则无需重试
_ = retryClient.Ack()//确认消息
fmt.Printf("data:%s\n", string(data))
return true
},
}
instanceConsumePool.RegisterConsumeReceive(nomrl)
err := instanceConsumePool.RunConsume()
if err != nil {
fmt.Println(err)
}
- 参数说明
名称 | 类型 | 说明 |
---|---|---|
ExchangeName | string | 交换机名称 |
ExchangeType | string | 交换机类型: EXCHANGE_TYPE_FANOUT EXCHANGE_TYPE_DIRECT EXCHANGE_TYPE_TOPIC |
Route | string | 路由键 |
QueueName | string | 队列名称 |
IsTry | bool | 是否重试 如果开启重试后, 在成功回调用返回true会对消息进行重试, 重试时间为 5000~15000 MS |
IsAutoAck | bool | 是否自动确认消息, true: 组件底层会自动对消息进行确认 false: 手动进行消息确认,在成功会调中需进行手动确认 _ = retryClient.Ack()
|
MaxReTry | int | 重试最大次数s, 需isTry为true |
EventFail | func | 失败回调 |
EventSuccess | func | 成功回调 |
错误码为
- 生产者push时返回的 *RabbitMqError
- 消费者事件监听回返的 code
错误码 | 说明 |
---|---|
501 | 生产者发送超过最大重试次数 |
502 | 获取信道失败, 一般为认道队列数用尽 |
503 | 交换机/队列/绑定失败 |
504 | 连接失败 |
506 | 信道创建失败 |
507 | 超过最大重试次数 |
git tag v0.0.2
git push --tags
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。