Skip to content

Commit d472d93

Browse files
kevwankevwan
committed
refactor rabbitmq
1 parent 254cf9c commit d472d93

File tree

8 files changed

+80
-91
lines changed

8 files changed

+80
-91
lines changed

‎example/rabbitmq/admin/admin.go‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,19 @@
11
package main
22

33
import (
4-
"github.com/zeromicro/go-queue/rabbitmq"
54
"log"
5+
6+
"github.com/zeromicro/go-queue/rabbitmq"
67
)
78

89
func main() {
9-
10-
conf := rabbitmq.RabbitMqConf{
10+
conf := rabbitmq.RabbitConf{
1111
Host: "192.168.253.100",
1212
Port: 5672,
1313
Username: "guest",
1414
Password: "guest",
1515
}
16-
admin := rabbitmq.MustNewRabbitMqAdmin(conf)
16+
admin := rabbitmq.MustNewAdmin(conf)
1717
exchangeConf := rabbitmq.ExchangeConf{
1818
ExchangeName: "jiang",
1919
Type: "direct",
@@ -27,6 +27,7 @@ func main() {
2727
if err != nil {
2828
log.Fatal(err)
2929
}
30+
3031
queueConf := rabbitmq.QueueConf{
3132
Name: "jxj",
3233
Durable: true,
@@ -38,6 +39,7 @@ func main() {
3839
if err != nil {
3940
log.Fatal(err)
4041
}
42+
4143
err = admin.Bind("jxj", "jxj", "jiang", false, nil)
4244
if err != nil {
4345
log.Fatal(err)

‎example/rabbitmq/listener/config/config.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,5 +3,5 @@ package config
33
import "github.com/zeromicro/go-queue/rabbitmq"
44

55
type Config struct {
6-
ListenerConf rabbitmq.RabbitMqListenerConf
6+
ListenerConf rabbitmq.RabbitListenerConf
77
}

‎example/rabbitmq/listener/main.go‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@ package main
33
import (
44
"flag"
55
"fmt"
6+
67
"github.com/zeromicro/go-queue/example/rabbitmq/listener/config"
78
"github.com/zeromicro/go-queue/rabbitmq"
8-
99
"github.com/zeromicro/go-zero/core/conf"
1010
"github.com/zeromicro/go-zero/core/service"
1111
)
@@ -17,13 +17,11 @@ func main() {
1717
var c config.Config
1818
conf.MustLoad(*configFile, &c)
1919

20-
listener := rabbitmq.MustNewRabbitMqListener(c.ListenerConf, Handler{})
21-
20+
listener := rabbitmq.MustNewListener(c.ListenerConf, Handler{})
2221
serviceGroup := service.NewServiceGroup()
2322
serviceGroup.Add(listener)
2423
defer serviceGroup.Stop()
2524
serviceGroup.Start()
26-
2725
}
2826

2927
type Handler struct {

‎example/rabbitmq/sender/main.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,19 +2,19 @@ package main
22

33
import (
44
"encoding/json"
5-
"github.com/zeromicro/go-queue/rabbitmq"
65
"log"
6+
7+
"github.com/zeromicro/go-queue/rabbitmq"
78
)
89

910
func main() {
10-
11-
conf := rabbitmq.RabbitMqSenderConf{RabbitMqConf: rabbitmq.RabbitMqConf{
11+
conf := rabbitmq.RabbitSenderConf{RabbitConf: rabbitmq.RabbitConf{
1212
Host: "192.168.253.100",
1313
Port: 5672,
1414
Username: "guest",
1515
Password: "guest",
1616
}, ContentType: "application/json"}
17-
sender := rabbitmq.MustNewRabbitMqSender(conf)
17+
sender := rabbitmq.MustNewSender(conf)
1818
data := map[string]interface{}{
1919
"msg": "json test 111",
2020
}
@@ -23,8 +23,9 @@ func main() {
2323
if err != nil {
2424
log.Fatal(err)
2525
}
26+
2627
conf.ContentType = "text/plain"
27-
sender = rabbitmq.MustNewRabbitMqSender(conf)
28+
sender = rabbitmq.MustNewSender(conf)
2829
message := "test message"
2930
err = sender.Send("exchange.direct", "gozero", []byte(message))
3031
if err != nil {

‎rabbitmq/config.go‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,29 +2,32 @@ package rabbitmq
22

33
import "fmt"
44

5-
type RabbitMqConf struct {
5+
type RabbitConf struct {
66
Username string
77
Password string
88
Host string
99
Port int
1010
VHost string `json:",optional"`
1111
}
1212

13-
type RabbitMqListenerConf struct {
14-
RabbitMqConf
13+
type RabbitListenerConf struct {
14+
RabbitConf
1515
ListenerQueues []ConsumerConf
1616
}
1717

1818
type ConsumerConf struct {
1919
Name string
2020
AutoAck bool `json:",default=true"`
2121
Exclusive bool `json:",default=false"`
22-
NoLocal bool `json:",default=false"` // Set to true, which means that messages sent by producers in the same connection cannot be delivered to consumers in this connection
23-
NoWait bool `json:",default=false"` // Whether to block processing
22+
// Set to true, which means that messages sent by producers in the same connection
23+
// cannot be delivered to consumers in this connection.
24+
NoLocal bool `json:",default=false"`
25+
// Whether to block processing
26+
NoWait bool `json:",default=false"`
2427
}
2528

26-
type RabbitMqSenderConf struct {
27-
RabbitMqConf
29+
type RabbitSenderConf struct {
30+
RabbitConf
2831
ContentType string `json:",default=text/plain"` // MIME content type
2932
}
3033

@@ -46,6 +49,7 @@ type ExchangeConf struct {
4649
Queues []QueueConf
4750
}
4851

49-
func getRabbitMqURL(rabbitConf RabbitMqConf) string {
50-
return fmt.Sprintf("amqp://%s:%s@%s:%d/%s", rabbitConf.Username, rabbitConf.Password, rabbitConf.Host, rabbitConf.Port, rabbitConf.VHost)
52+
func getRabbitURL(rabbitConf RabbitConf) string {
53+
return fmt.Sprintf("amqp://%s:%s@%s:%d/%s", rabbitConf.Username, rabbitConf.Password,
54+
rabbitConf.Host, rabbitConf.Port, rabbitConf.VHost)
5155
}

‎rabbitmq/listener.go‎

Lines changed: 16 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
11
package rabbitmq
22

33
import (
4-
"fmt"
4+
"log"
5+
56
"github.com/streadway/amqp"
67
"github.com/zeromicro/go-zero/core/logx"
78
"github.com/zeromicro/go-zero/core/queue"
8-
"log"
99
)
1010

1111
type (
@@ -20,40 +20,28 @@ type (
2020
channel *amqp.Channel
2121
forever chan bool
2222
handler ConsumeHandler
23-
queues RabbitMqListenerConf
23+
queues RabbitListenerConf
2424
}
2525
)
2626

27-
func MustNewRabbitMqListener(listenerConf RabbitMqListenerConf, handler ConsumeHandler) queue.MessageQueue {
28-
q, err := newRabbitMq(listenerConf, handler)
27+
func MustNewListener(listenerConf RabbitListenerConf, handler ConsumeHandler) queue.MessageQueue {
28+
listener := RabbitListener{queues: listenerConf, handler: handler, forever: make(chan bool)}
29+
conn, err := amqp.Dial(getRabbitURL(listenerConf.RabbitConf))
2930
if err != nil {
30-
log.Fatal(err)
31+
log.Fatalf("failed to connect rabbitmq, error: %v", err)
3132
}
3233

33-
return q
34-
}
35-
36-
func newRabbitMq(listenerConf RabbitMqListenerConf, handler ConsumeHandler) (queue.MessageQueue, error) {
37-
listener := RabbitListener{queues: listenerConf, handler: handler, forever: make(chan bool)}
38-
conn, err := amqp.Dial(getRabbitMqURL(listenerConf.RabbitMqConf))
39-
listener.ErrorHandler(err, "failed to connect rabbitmq!")
4034
listener.conn = conn
41-
4235
channel, err := listener.conn.Channel()
43-
listener.ErrorHandler(err, "failed to open a channel")
44-
listener.channel = channel
45-
return listener, nil
46-
}
47-
48-
func (q RabbitListener) ErrorHandler(err error, message string) {
4936
if err != nil {
50-
logx.Errorf("%s:%s", message, err)
51-
panic(fmt.Sprintf("%s:%s", message, err))
37+
log.Fatalf("failed to open a channel: %v", err)
5238
}
39+
40+
listener.channel = channel
41+
return listener
5342
}
5443

5544
func (q RabbitListener) Start() {
56-
5745
for _, que := range q.queues.ListenerQueues {
5846
msg, err := q.channel.Consume(
5947
que.Name,
@@ -64,7 +52,10 @@ func (q RabbitListener) Start() {
6452
que.NoWait,
6553
nil,
6654
)
67-
q.ErrorHandler(err, "failed to listener")
55+
if err != nil {
56+
log.Fatalf("failed to listener, error: %v", err)
57+
}
58+
6859
go func() {
6960
for d := range msg {
7061
if err := q.handler.Consume(string(d.Body)); err != nil {
@@ -73,6 +64,7 @@ func (q RabbitListener) Start() {
7364
}
7465
}()
7566
}
67+
7668
<-q.forever
7769
}
7870

‎rabbitmq/rabbitmqadmin.go‎

Lines changed: 22 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,39 +1,35 @@
11
package rabbitmq
22

33
import (
4-
"fmt"
4+
"log"
5+
56
"github.com/streadway/amqp"
6-
"github.com/zeromicro/go-zero/core/logx"
77
)
88

9-
type (
10-
RabbitMqAdmin struct {
11-
conn *amqp.Connection
12-
channel *amqp.Channel
9+
type Admin struct {
10+
conn *amqp.Connection
11+
channel *amqp.Channel
12+
}
13+
14+
func MustNewAdmin(rabbitMqConf RabbitConf) *Admin {
15+
var admin Admin
16+
conn, err := amqp.Dial(getRabbitURL(rabbitMqConf))
17+
if err != nil {
18+
log.Fatalf("failed to connect rabbitmq, error: %v", err)
1319
}
14-
)
1520

16-
func MustNewRabbitMqAdmin(rabbitMqConf RabbitMqConf) *RabbitMqAdmin {
17-
admin := &RabbitMqAdmin{}
18-
conn, err := amqp.Dial(getRabbitMqURL(rabbitMqConf))
19-
admin.ErrorHandler(err, "failed to connect rabbitmq!")
2021
admin.conn = conn
21-
2222
channel, err := admin.conn.Channel()
23-
admin.ErrorHandler(err, "failed to open a channel")
24-
admin.channel = channel
25-
return admin
26-
}
27-
28-
func (q *RabbitMqAdmin) ErrorHandler(err error, message string) {
2923
if err != nil {
30-
logx.Errorf("%s:%s", message, err)
31-
panic(fmt.Sprintf("%s:%s", message, err))
24+
log.Fatalf("failed to open a channel, error: %v", err)
3225
}
26+
27+
admin.channel = channel
28+
return &admin
3329
}
3430

35-
func (q *RabbitMqAdmin) DeclareExchange(conf ExchangeConf, args amqp.Table) error {
36-
err := q.channel.ExchangeDeclare(
31+
func (q *Admin) DeclareExchange(conf ExchangeConf, args amqp.Table) error {
32+
return q.channel.ExchangeDeclare(
3733
conf.ExchangeName,
3834
conf.Type,
3935
conf.Durable,
@@ -42,11 +38,9 @@ func (q *RabbitMqAdmin) DeclareExchange(conf ExchangeConf, args amqp.Table) erro
4238
conf.NoWait,
4339
args,
4440
)
45-
46-
return err
4741
}
4842

49-
func (q *RabbitMqAdmin) DeclareQueue(conf QueueConf, args amqp.Table) error {
43+
func (q *Admin) DeclareQueue(conf QueueConf, args amqp.Table) error {
5044
_, err := q.channel.QueueDeclare(
5145
conf.Name,
5246
conf.Durable,
@@ -55,16 +49,16 @@ func (q *RabbitMqAdmin) DeclareQueue(conf QueueConf, args amqp.Table) error {
5549
conf.NoWait,
5650
args,
5751
)
52+
5853
return err
5954
}
6055

61-
func (q *RabbitMqAdmin) Bind(queueName string, routekey string, exchange string, notWait bool, args amqp.Table) error {
62-
err := q.channel.QueueBind(
56+
func (q *Admin) Bind(queueName string, routekey string, exchange string, notWait bool, args amqp.Table) error {
57+
return q.channel.QueueBind(
6358
queueName,
6459
routekey,
6560
exchange,
6661
notWait,
6762
args,
6863
)
69-
return err
7064
}

‎rabbitmq/sender.go‎

Lines changed: 14 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
package rabbitmq
22

33
import (
4-
"fmt"
4+
"log"
5+
56
"github.com/streadway/amqp"
6-
"github.com/zeromicro/go-zero/core/logx"
77
)
88

99
type (
@@ -18,26 +18,25 @@ type (
1818
}
1919
)
2020

21-
func MustNewRabbitMqSender(rabbitMqConf RabbitMqSenderConf) Sender {
21+
func MustNewSender(rabbitMqConf RabbitSenderConf) Sender {
2222
sender := &RabbitMqSender{ContentType: rabbitMqConf.ContentType}
23-
conn, err := amqp.Dial(getRabbitMqURL(rabbitMqConf.RabbitMqConf))
24-
sender.ErrorHandler(err, "failed to connect rabbitmq!")
25-
sender.conn = conn
23+
conn, err := amqp.Dial(getRabbitURL(rabbitMqConf.RabbitConf))
24+
if err != nil {
25+
log.Fatalf("failed to connect rabbitmq, error: %v", err)
26+
}
2627

28+
sender.conn = conn
2729
channel, err := sender.conn.Channel()
28-
sender.ErrorHandler(err, "failed to open a channel")
29-
sender.channel = channel
30-
return sender
31-
}
32-
func (q *RabbitMqSender) ErrorHandler(err error, message string) {
3330
if err != nil {
34-
logx.Errorf("%s:%s", message, err)
35-
panic(fmt.Sprintf("%s:%s", message, err))
31+
log.Fatalf("failed to open a channel, error: %v", err)
3632
}
33+
34+
sender.channel = channel
35+
return sender
3736
}
38-
func (q *RabbitMqSender) Send(exchange string, routeKey string, msg []byte) error {
3937

40-
err := q.channel.Publish(
38+
func (q *RabbitMqSender) Send(exchange string, routeKey string, msg []byte) error {
39+
return q.channel.Publish(
4140
exchange,
4241
routeKey,
4342
false,
@@ -47,5 +46,4 @@ func (q *RabbitMqSender) Send(exchange string, routeKey string, msg []byte) erro
4746
Body: msg,
4847
},
4948
)
50-
return err
5149
}

0 commit comments

Comments
 (0)