求助 go-nsq 消费者的问题

[dt2vba]

大佬，下午好，我使用 go-nsq 消息队列处理请求时遇到一个问题。

假设有以下三个文件
main.go publishMessage.go consumeMessage.go

启动服务
go run main.go publishMessage.go
发送请求
curl -d "localhost:1323" -d "MESSAGE"

现在的问题是，我可以发布消息，但是只能启动新的进程去消费消息，go run consumeMessage.go

请问如何在一个进程中处理请求呢？

[lockerhyz]

go run main.go publishMessage.go consumeMessage.go

[dt2vba]

@lockerhyz 谢谢关注。我试了一下，不能正常工作。

[dt2vba]

@lockerhyz

你好，如果你有时间的话，可以帮我看一下吗？以下是工作流程，

消息是 HTTP 请求表单数据

publishMessage.go 发布消息

consumeMessage.go 消费消息


FILE 1 main.go

package main

import (
"github.com/labstack/echo"
)

type User struct {Name string}

func main() {
e:=echo.New()
r:=e.Group("/api/v1/applicant/register/user")
r.POST("",createUserInNSQ)
e.Start(":1323")
}


FILE 2 publishMessage.go

package main

import (
...
"github.com/nsqio/go-nsq"
...
)

var tcpNsqdAddr="127.0.0.1:4150"

func publishMessage(topic string,command []uint8) error {
config:=nsq.NewConfig()
producer,_:=nsq.NewProducer(tcpNsqdAddr,config)
producer.Publish(topic,command)
return nil
}

func createUserInNSQ(c echo.Context) (err error) {
u:=new(User)

topic:="InsertUser"
command,err:=json.Marshal(u)
publishMessage(topic,command)

return c.String( http.StatusOK,"OK")
}


FILE 3 consumeMessage.go

package main

import (
...
"github.com/nsqio/go-nsq"
...
)

var tcpNsqdAddr="127.0.0.1:4150"

type NsqHandler struct {
}

func (s *NsqHandler) HandleMessage(message *nsq.Message) error {
//insert message into MySQL
...

return nil
}

func main() {
config:=nsq.NewConfig()
com,_:=nsq.NewConsumer("InsertUser","channel1",config)
com.AddHandler(&NsqHandler{NsqHandlerId:"one"})
com.ConnectToNSQD(tcpNsqdAddr)

var wg=&sync.WaitGroup{}
wg.Add(1)
wg.Wait()
}

[keepeye]

学会使用 goroutine

[dt2vba]

@keepeye 你好，感谢提供帮助！

[lockerhyz]

@dt2vba
1. 把 consumeMessage.go 的 main 函数修改成其他函数名，例如 consumer()
2. 修改 main.go 中的 main 函数

```
func main() {
go consumer()
e:=echo.New()
r:=e.Group("/api/v1/applicant/register/user")
r.POST("",createUserInNSQ)
e.Start(":1323")
}
```

[lockerhyz]

@lockerhyz
3. go run main.go publishMessage.go consumeMessage.go

[dt2vba]

@lockerhyz 非常感谢你的帮助。