关于ZeroMQ(又称?MQ)有很多传神的解说,我这里给各位看官上一段比较凡尔赛的描述。
Underneath the brown paper wrapping of ZeroMQ’s socket API lies the world of messaging patterns.
"在ZeroMQ发黄的封皮下面,流淌的是全世界的消息通信模式",一翻译就没有文艺范儿了,总之口气很大(敢自称Zero就已经体现了作者的雄心),声称ZeroMQ涵盖了所有的通信模式,包括进程内/进程间通讯、TCP、广播等等模式。
闲言少叙,书归正传。网上讲go语言ZeroMQ编程的例子比较少,今天列一些常见的通信模式,给初学者一个指引,希望大家少走弯路。
REQ-REP
即request-response模式,典型的client-server架构,client发送request给server,server返回response给client,一个server可以同时跟多个client建立连接。
在ZeroMQ中凡是涉及到“1对多”的模式,都是“1”的那一方调用Bind()函数在某个端口上开始监听,“多”的那一方调用Connect()函数请求跟“1”方建立连接。在普通的socket编程中,必须先启动Server,再启动Client,否则Client的Connect()函数会失败从而退出程序。然而ZeroMQ的Connect()函数实际上是异步的,如果如果Server端还没有启起来它不会以失败告终,而是立即返回,你可以紧接着调用Send()函数。Send()又支持阻塞模式和非阻塞模式,阻塞模式会一直等待对方就绪(比如至少要等连接建立好吧)再发送数据,非阻塞模式会先把消息存到本的缓冲队列里然后立即返回。Server端由于维护了多条连接,它会为每个连接都单独建立一个缓冲队列。
在req-rep模式中,每台机器send(发送数据)和revc(接收数据)必须交替调用,否则会报错。
import (
"fmt"
"strconv"
"time"
zmq "github.com/pebbe/zmq4"
)
func startServer(port int) {
//REP 表示server端
socket, _ := zmq.NewSocket(zmq.REP)
//Bind 绑定端口,并指定传输层协议
socket.Bind("tcp://127.0.0.1:" + strconv.Itoa(port))
fmt.Printf("bind to port %d\n", port)
defer socket.Close()
for {
//Recv和Send必须交替进行
resp, _ := socket.Recv(0) //0表示阻塞模式
socket.Send("Hello "+resp, 0) //同步发送
}
}
func startClient(port int, msg string) {
//REQ 表示client端
socket, _ := zmq.NewSocket(zmq.REQ)
//Connect 请求建立连接,并指定传输层协议
socket.Connect("tcp://127.0.0.1:" + strconv.Itoa(port))
fmt.Println("connect to server")
defer socket.Close()
for i := 0; i < 10; i++ {
//Send和Recv必须交替进行
socket.Send(msg, zmq.DONTWAIT) //非阻塞模式,异步发送(只是将数据写入本地buffer,并没有真正发送到网络上)
resp, _ := socket.Recv(0)
fmt.Printf("receive [%s]\n", resp)
time.Sleep(5 * time.Second)
}
}
DEALER-ROUTER
DEALER-ROUTER跟REQ-REP模式比较接近,都是client-server构架,但是DEALER-ROUTER有它的2个特别之处:
- send()和revc()不需要交替调用
- ROUTER端的消息需要分成多帧发送,至少2帧,因为第一帧需要发送对端的地址。消息正文如果需要分成多帧发送,那么除后一帧外其他帧在send时都需要设置SNDMORE标识。对于DEALER发过来的一条消息,ROUTER需要调用2次Recv(),第一次Recv()读出来的是对端的地址
import (
"fmt"
"strconv"
"time"
zmq "github.com/pebbe/zmq4"
)
func router(port int) {
//ROUTER 表示server端
socket, _ := zmq.NewSocket(zmq.ROUTER)
//Bind 绑定端口,并指定传输层协议
socket.Bind("tcp://127.0.0.1:" + strconv.Itoa(port))
fmt.Printf("bind to port %d\n", port)
defer socket.Close()
for {
//Send和Recv没必要交替进行
addr, _ := socket.RecvBytes(0) //接收到的第一帧表示对方的地址UUID
resp, _ := socket.Recv(0)
socket.SendBytes(addr, zmq.SNDMORE) //第一帧需要指明对方的地址,SNDMORE表示消息还没发完
socket.Send("Hello", zmq.SNDMORE) //如果不用SNDMORE表示这已经是最后一帧了,下一次Send就是下一段消息的第一帧了,需要指明对方的地址
socket.Send(resp, 0)
}
}
func dealer(port int, msg string) {
//DEALER 表示client端
socket, _ := zmq.NewSocket(zmq.DEALER)
//Connect 请求建立连接,并指定传输层协议
socket.Connect("tcp://127.0.0.1:" + strconv.Itoa(port))
fmt.Println("connect to server")
defer socket.Close()
for i := 0; i < 10; i++ {
//Send和Recv没必要交替进行
socket.Send(msg, 0) //非阻塞模式,异步发送(只是将数据写入本地buffer,并没有真正发送到网络上)
resp1, _ := socket.Recv(0)
resp2, _ := socket.Recv(0)
fmt.Printf("receive [%s %s]\n", resp1, resp2)
time.Sleep(5 * time.Second)
}
}
SUB-PUB
PUB代表发布者publisher,SUB代表订阅者subscriber,发布者发送一条消息,所有的订阅者都会收到,是典型的广播模式。
订阅者只接收特定主题的消息,发布者在每条消息前面加一个特定的前缀表示主题。订阅者通过调用SetSubscribe(prefix string)函数过滤出特定前缀的消息,如果给SetSubscribe()传的是空字符串则订阅者不过滤任何消息,全部接收。
显然发布者只能调用Send,订阅者只能调用Recv。
import (
"fmt"
"strconv"
"time"
zmq "github.com/pebbe/zmq4"
)
func publish(port int, prefix string) {
ctx, _ := zmq.NewContext()
defer ctx.Term()
//PUB 表示publisher角色
publisher, _ := ctx.NewSocket(zmq.PUB)
defer publisher.Close()
//Bind 绑定端口,并指定传输层协议
publisher.Bind("tcp://127.0.0.1:" + strconv.Itoa(port))
//publisher会把消息发送给所有subscriber,subscriber可以动态加入
for i := 0; i < 5; i++ {
//publisher只能调用send方法
publisher.Send(prefix+"Hello my followers", 0)
publisher.Send(prefix+"How are you", 0)
fmt.Printf("loop %d send over\n", i+1)
time.Sleep(10 * time.Second)
}
publisher.Send(prefix+"END", 0)
}
func subscribe(port int, prefix string) {
//SUB 表示subscriber角色
subscriber, _ := zmq.NewSocket(zmq.SUB)
defer subscriber.Close()
//Bind 绑定端口,并指定传输层协议
subscriber.Connect("tcp://127.0.0.1:" + strconv.Itoa(port))
subscriber.SetSubscribe(prefix) //只接收前缀为prefix的消息
fmt.Printf("listen to port %d\n", port)
for {
//接收广播
if resp, err := subscriber.Recv(0); err == nil {
resp = resp[len(prefix):] //去掉前缀
fmt.Printf("receive [%s]\n", resp)
if resp == "END" {
break
}
} else {
fmt.Println(err)
break
}
}
}
PUSH-PULL
跟PUB-SUB模式一样,PUSH-PULL也只能实现消息的单向传输,但是pusher会采用轮询法选择一个worker把消息发送给它,其他worker都收不到这条消息,所以PUSH-PULL模式常用来做任务的分发。
import (
"fmt"
"strconv"
"time"
zmq "github.com/pebbe/zmq4"
)
func push(port int) {
ctx, _ := zmq.NewContext()
defer ctx.Term()
//PUSH 表示pusher角色
pusher, _ := ctx.NewSocket(zmq.PUSH)
defer pusher.Close()
//Bind 绑定端口,并指定传输层协议
pusher.SetSndhwm(110)
pusher.Bind("tcp://127.0.0.1:" + strconv.Itoa(port))
//pusher把消息送给一个puller(采用公平轮转的方式选择一个puller),puller可以动态加入
for i := 0; i < 5; i++ {
pusher.Send("Hello my followers", 0)
pusher.Send("How are you", 0)
fmt.Printf("loop %d send over\n", i+1)
time.Sleep(5 * time.Second)
}
pusher.Send("END", 0)
}
func pull(port int) {
//PULL 表示puller角色
puller, _ := zmq.NewSocket(zmq.PULL)
defer puller.Close()
//Bind 绑定端口,并指定传输层协议
puller.Connect("tcp://127.0.0.1:" + strconv.Itoa(port))
fmt.Printf("listen to port %d\n", port)
for {
//接收广播
if resp, err := puller.Recv(0); err == nil {
fmt.Printf("receive [%s]\n", resp)
if resp == "END" {
break
}
} else {
fmt.Println(err)
break
}
}
}
聊天室后端架构
最后来一个综合练习,实现一个聊天室的后端。采用hub-client架构,client代表聊天室中发言的用户,client通过DEALER-ROUTER模式把发言内容发给hub(hub不需要专门响应该client,所以不能采用REQ-REP模式,REQ-REP要求send和recv必须交替进行),hub通过PUB-SUB模式把消息广播给所有client。
import (
"bufio"
"encoding/base64"
"fmt"
"os"
"strconv"
"strings"
zmq "github.com/pebbe/zmq4"
)
func hub(subPort, pubPort int) {
//接收所有client的消息
socket, _ := zmq.NewSocket(zmq.ROUTER)
socket.Bind("tcp://127.0.0.1:" + strconv.Itoa(subPort))
fmt.Printf("bind to port %d\n", subPort)
defer socket.Close()
//把消息广播给所有client
ctx, _ := zmq.NewContext()
defer ctx.Term()
publisher, _ := ctx.NewSocket(zmq.PUB)
defer publisher.Close()
publisher.Bind("tcp://127.0.0.1:" + strconv.Itoa(pubPort))
for {
//把接收到的client的消息再广播给所有client
if addr, err := socket.RecvBytes(0); err == nil { //第一帧读出对端的地址
client := base64.StdEncoding.EncodeToString(addr) //用对端地址来标识消息是谁发出来的
if resp, err := socket.Recv(0); err == nil {
if _, err := publisher.Send(client+"say: "+resp, 0); err != nil { //在消息前加上发送者的标识
fmt.Println(err)
break
}
} else {
fmt.Println(err)
break
}
} else {
fmt.Println(err)
break
}
}
}
func client(pubPort, subPort int) {
//把消息广播给hub
socket, _ := zmq.NewSocket(zmq.DEALER)
socket.Connect("tcp://127.0.0.1:" + strconv.Itoa(pubPort))
fmt.Println("connect to server")
defer socket.Close()
//订阅hub的消息
subscriber, _ := zmq.NewSocket(zmq.SUB)
defer subscriber.Close()
subscriber.Connect("tcp://127.0.0.1:" + strconv.Itoa(subPort))
subscriber.SetSubscribe("")
go func() {
for {
//把接收到的client的消息再广播给所有client
if resp, err := subscriber.Recv(0); err == nil {
fmt.Println(resp)
} else {
fmt.Println(err)
break
}
}
}()
fmt.Println("please type message")
reader := bufio.NewReader(os.Stdin)
for {
text, _ := reader.ReadString(‘\n‘)
text = strings.Replace(text, "\n", "", -1)
socket.Send(text, 0)
}
}