rabbitmq 操作库
Ви не можете вибрати більше 25 тем Теми мають розпочинатися з літери або цифри, можуть містити дефіси (-) і не повинні перевищувати 35 символів.

147 рядки
3.6 KiB

  1. package test
  2. import (
  3. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  4. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbitmq"
  5. "fmt"
  6. "github.com/streadway/amqp"
  7. "log"
  8. "strconv"
  9. "strings"
  10. "testing"
  11. )
  12. const SingleQueueName = "test_single_queue"
  13. func TestSend(t *testing.T) {
  14. var host, port, user, pwd string
  15. log.SetFlags(log.Lshortfile | log.LstdFlags)
  16. log.Println("input host,port,user,pwd")
  17. fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
  18. if host == "" && port == "" && user == "" && pwd == "" {
  19. host = "119.23.182.117"
  20. port = "5672"
  21. user = "admin"
  22. pwd = "123456"
  23. }
  24. rabbitmq.Init(host, port, user, pwd)
  25. //第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
  26. sendMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
  27. i := 0
  28. for {
  29. if i > 99 {
  30. break
  31. }
  32. //time.Sleep(time.Second * 1)
  33. greetings := []string{"HelloWorld!", strconv.Itoa(i)}
  34. sendMq.Send(SingleQueueName, strings.Join(greetings, " "))
  35. i = i + 1
  36. }
  37. }
  38. func TestReceive(t *testing.T) {
  39. var host, port, user, pwd string
  40. log.SetFlags(log.Lshortfile | log.LstdFlags)
  41. log.Println("input host,port,user,pwd")
  42. fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
  43. if host == "" && port == "" && user == "" && pwd == "" {
  44. host = "119.23.182.117"
  45. port = "5672"
  46. user = "admin"
  47. pwd = "123456"
  48. }
  49. //ch, err := GetChannel
  50. rabbitmq.Init(host, port, user, pwd)
  51. //第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
  52. receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
  53. //for {
  54. receiveMq.Qos(1)
  55. //接收消息时,指定
  56. messages := receiveMq.Consume()
  57. forever := make(chan bool) //开辟管道
  58. var i = 0
  59. //启用协程处理消息
  60. go func() {
  61. for msg := range messages {
  62. log.Printf("Received a message: %s", msg.Body)
  63. i++
  64. if i >= 5 {
  65. msg.Reject(true)
  66. receiveMq.CloseChannel()
  67. return
  68. } else {
  69. msg.Ack(false)
  70. }
  71. }
  72. }()
  73. log.Printf("[*] Waiting for message,To exit press CTRL+C")
  74. <-forever
  75. }
  76. func TestReceive1(t *testing.T) {
  77. var host, port, user, pwd string
  78. log.SetFlags(log.Lshortfile | log.LstdFlags)
  79. log.Println("input host,port,user,pwd")
  80. fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
  81. if host == "" && port == "" && user == "" && pwd == "" {
  82. host = "119.23.182.117"
  83. port = "5672"
  84. user = "admin"
  85. pwd = "123456"
  86. }
  87. rabbit.Init(host, port, user, pwd)
  88. ch, err := rabbit.Cfg.Pool.GetChannel()
  89. if err != nil {
  90. fmt.Println(err.Error())
  91. }
  92. defer ch.Release()
  93. ch.Qos(5)
  94. delivery := ch.Consume(SingleQueueName, false)
  95. var res amqp.Delivery
  96. var ok bool
  97. //for {
  98. res, ok = <-delivery
  99. if ok == true {
  100. fmt.Println(string(res.Body))
  101. _ = res.Ack(false)
  102. } else {
  103. fmt.Println("error getting message")
  104. //break
  105. }
  106. //}
  107. fmt.Println("get msg done")
  108. }
  109. func TestReceive2(t *testing.T) {
  110. var host, port, user, pwd string
  111. log.SetFlags(log.Lshortfile | log.LstdFlags)
  112. log.Println("input host,port,user,pwd")
  113. fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
  114. if host == "" && port == "" && user == "" && pwd == "" {
  115. host = "119.23.182.117"
  116. port = "5672"
  117. user = "admin"
  118. pwd = "123456"
  119. }
  120. //ch, err := GetChannel
  121. rabbitmq.Init(host, port, user, pwd)
  122. //第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
  123. receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
  124. //for {
  125. receiveMq.Qos(1)
  126. //接收消息时,指定
  127. messages, ok, err := receiveMq.Get()
  128. if err != nil {
  129. t.Fatalf("Failed get: %v", err)
  130. }
  131. if !ok {
  132. t.Fatalf("Get on a queued message did not find the message")
  133. }
  134. fmt.Println(">>>>>>ok>>>>>>", ok)
  135. fmt.Println(">>>>>>messages>>>>>>", string(messages.Body))
  136. }