NanoMsg框架|C#中REQREP、PUBSUB和SURVEY使用（附Demo））

Vaccae

发布于 2020-02-25 15:09:48

1.6K0

发布于 2020-02-25 15:09:48

导语

前一篇中《NanoMsg框架|C#中Nanomsg的PAIR和BUS使用》已经介绍了PAIR和BUS两个模式，这一篇我们把剩下几个常用的一起说了，像REQREP、PUBSUB和SURVEY，主要是因为NNanoMsg里面已经把这些都封装的差不多了，调用方式基本都一样，所以不就浪费章节了，这篇介绍完后我们就要来说Android这块怎么使用nanomsg，那个相对来说就比较麻烦多了。

REQREP模式

微卡智享

REQREP模式：允许构建集群的无状态服务来处理用户请求。每个req请求都需要rep响应，由请求端发起请求，然后等待回应端应答，一般是一个REP多个REQ类似http协议(一应一答)。关键字：NN_REP,NN_REQ。

Server端代码

        /// <summary>
        /// REQREP
        /// </summary>
        private void REQREPSocketSrvConnect()
        {
            try
            {
                replySocket = new Lazy<ReplySocket>(() => new ReplySocket());
                replySocket.Value.Bind("tcp://*:8001");
                new Task(() =>
                {
                    while (replySocket.IsValueCreated)
                    {
                        Thread.Sleep(1000);
                        //接收数据
                        byte[] buffer = replySocket.Value.Receive();
                        string recvstr = Encoding.UTF8.GetString(buffer);
                        TextShow(recvstr);
                        //发送数据
                        replySocket.Value.Send(Encoding.UTF8.GetBytes("已收到" + recvstr));
                    }
                }).Start();
            }
            catch (Exception ex)
            {
                TextShow(ex.Message);
            }
        }

Client端代码

 requestSocket = new Lazy<RequestSocket>(() => new RequestSocket());
 requestSocket.Value.Connect("tcp://localhost:8001");
 
 if (requestSocket.IsValueCreated)
 {
    //发送数据
    requestSocket.Value.Send(Encoding.UTF8.GetBytes(inputstr));

    //接收数据
    byte[] buffer = requestSocket.Value.Receive();
    TextShow("REQREP模式返回数据：" + Encoding.UTF8.GetString(buffer));
 }
 else
 {
    TextShow("REQREP未创建！");
 }

REQREP效果

PUBSUB模式

微卡智享

PUBSUB模式：结合使用可实现消息广播模式（Topics && Broadcast）服务端只管发布，不管客户端是否连接，也不管是不是丢消息，但客户端连接上来以后就不会丢消息。

Server端发送时可以在前缀的字符串定义不同的主题类型，Client端可以通过Subscribe设置接收到的订阅主题，如果Subscribe设置为空，即接收所有的主题。

Server端代码

        /// <summary>
        /// PUBSUB连接
        /// </summary>
        private void PUBSUBSrvConnect()
        {
            try
            {
                publishSocket = new Lazy<PublishSocket>(() => new PublishSocket());
                publishSocket.Value.Bind("tcp://*:8002");
            }
            catch (Exception ex)
            {
                TextShow(ex.Message);
            }
        }

Client端代码

  subscribeSocket = new Lazy<SubscribeSocket>(() => new SubscribeSocket());
  //一定要指定订阅的主题前缀，指定为空订阅所有主题,否则收不到
  subscribeSocket.Value.Subscribe("PUBSUB");

  subscribeSocket.Value.Connect("tcp://localhost:8002");

  new Task(() =>
  {
      while (subscribeSocket.IsValueCreated)
      {
          Thread.Sleep(1000);
          //接收数据
          byte[] buffer = subscribeSocket.Value.Receive();
          string recvstr = Encoding.UTF8.GetString(buffer);
          TextShow(recvstr);
      }
  }).Start();

PUBSUB效果

SURVEY模式

微卡智享

SURVEY模式：可实现调查模式，允许向多个地点广播调查并收集响应，surveyort负责发出调查问题，而respondent客户端则负责回应。关键字：NN_SURVEYOR，NN_RESPONDENT。

Server端代码

/// <summary>
/// SURVEY模式
/// </summary>
private void SURVEYSrcConnect()
{
    try
    {
        surveyorSocket = new Lazy<SurveyorSocket>(() => new SurveyorSocket());
        surveyorSocket.Value.Bind("tcp://*:8002");

        new Task(() =>
        {
            while (surveyorSocket.IsValueCreated)
            {
                Thread.Sleep(1000);

                //发送数据
                string sendstr = "SURVEY消息，请查收！";
                surveyorSocket.Value.Send(Encoding.UTF8.GetBytes(sendstr));
                TextShow(sendstr);

                //接收消息
                byte[] buffer = surveyorSocket.Value.Receive();
                if (buffer != null)
                {
                    string recvstr = Encoding.UTF8.GetString(buffer);
                    TextShow(recvstr);
                }                     
            }
        }).Start();
    }
    catch (Exception ex)
    {
        TextShow(ex.Message);
    }
}

Client端代码

respondentSocket = new Lazy<RespondentSocket>(() => new RespondentSocket());
respondentSocket.Value.Connect("tcp://localhost:8002");
new Task(() =>
{
    while (respondentSocket.IsValueCreated)
    {
        Thread.Sleep(1000);
        //接收数据
        byte[] buffer = respondentSocket.Value.Receive();
        string recvstr = Encoding.UTF8.GetString(buffer);
        TextShow(recvstr);
        //发送数据
        string sendstr = "已收到!";
        respondentSocket.Value.Send(Encoding.UTF8.GetBytes(sendstr));
        TextShow("发送返回数据：" + sendstr);
    }
}).Start();

SURVEY效果