rabbitmq 操作库
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

single_queue_test.go 2.7 KiB

2 年之前
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111
  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 = "10.10.140.138"
  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 > 100 {
  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 = "10.10.140.138"
  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(5)
  55. //接收消息时,指定
  56. messages := receiveMq.Consume()
  57. forever := make(chan bool) //开辟管道
  58. //启用协程处理消息
  59. go func() {
  60. for msg := range messages {
  61. log.Printf("Received a message: %s", msg.Body)
  62. msg.Ack(false)
  63. //msg.Reject(true)
  64. }
  65. }()
  66. log.Printf("[*] Waiting for message,To exit press CTRL+C")
  67. <-forever
  68. }
  69. func TestReceive1(t *testing.T) {
  70. var host, port, user, pwd string
  71. log.SetFlags(log.Lshortfile | log.LstdFlags)
  72. log.Println("input host,port,user,pwd")
  73. fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
  74. if host == "" && port == "" && user == "" && pwd == "" {
  75. host = "10.10.140.138"
  76. port = "5672"
  77. user = "admin"
  78. pwd = "123456"
  79. }
  80. rabbit.Init(host, port, user, pwd)
  81. ch, err := rabbit.Cfg.Pool.GetChannel()
  82. if err != nil {
  83. fmt.Println(err.Error())
  84. }
  85. defer ch.Release()
  86. ch.Qos(5)
  87. delivery := ch.Consume(SingleQueueName)
  88. var res amqp.Delivery
  89. var ok bool
  90. //for {
  91. res, ok = <-delivery
  92. if ok == true {
  93. fmt.Println(string(res.Body))
  94. _ = res.Ack(false)
  95. } else {
  96. fmt.Println("error getting message")
  97. //break
  98. }
  99. //}
  100. fmt.Println("get msg done")
  101. }