go语言ZeroMQ编程(带聊天室的实现)

关于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建立连接。

go语言ZeroMQ编程(带聊天室的实现)

在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

go语言ZeroMQ编程(带聊天室的实现)

DEALER-ROUTER跟REQ-REP模式比较接近,都是client-server构架,但是DEALER-ROUTER有它的2个特别之处:

  1. send()和revc()不需要交替调用
  2. 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,发布者发送一条消息,所有的订阅者都会收到,是典型的广播模式。

go语言ZeroMQ编程(带聊天室的实现)

订阅者只接收特定主题的消息,发布者在每条消息前面加一个特定的前缀表示主题。订阅者通过调用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模式常用来做任务的分发。

go语言ZeroMQ编程(带聊天室的实现)

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。

go语言ZeroMQ编程(带聊天室的实现)

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)
	}
}

go语言ZeroMQ编程(带聊天室的实现)

上一篇:C#后台线程


下一篇:layui-数据表格排序