作為一個MQ，pub/sub是非常常見的一個用法，nanomsg作為ZMQ的繼任，很少有部落格講述nanomsg的訂閱/釋出模式。

官方DEMO

#include <assert.h>
#include <libc.h>
#include <stdio.h>
#include <nanomsg/nn.h>
#include <nanomsg/pubsub.h>

#define SERVER "server"
#define CLIENT "client"

char *date ()
{
  time_t raw = time (&raw);
  struct tm *info = localtime (&raw);
  char *text = asctime (info);
  text[strlen(text)-1] = '\0'; // remove '\n'
  return text;
}

int server (const char *url)
{
  int sock = nn_socket (AF_SP, NN_PUB);
  assert (sock >= 0);
  assert (nn_bind (sock, url) >= 0);
  while (1)
    {
      char *d = date();
      int sz_d = strlen(d) + 1; // '\0' too
      printf ("SERVER: PUBLISHING DATE %s\n", d);
      int bytes = nn_send (sock, d, sz_d, 0);
      assert (bytes == sz_d);
      sleep(1);
    }
  return nn_shutdown (sock, 0);
}

int client (const char *url, const char *name)
{
  int sock = nn_socket (AF_SP, NN_SUB);
  assert (sock >= 0);
  // TODO learn more about publishing/subscribe keys
  assert (nn_setsockopt (sock, NN_SUB, NN_SUB_SUBSCRIBE, "", 0) >= 0);
  assert (nn_connect (sock, url) >= 0);
  while (1)
    {
      char *buf = NULL;
      int bytes = nn_recv (sock, &buf, NN_MSG, 0);
      assert (bytes >= 0);
      printf ("CLIENT (%s): RECEIVED %s\n", name, buf);
      nn_freemsg (buf);
    }
  return nn_shutdown (sock, 0);
}

int main (const int argc, const char **argv)
{
  if (strncmp (SERVER, argv[1], strlen (SERVER)) == 0 && argc >= 2)
    return server (argv[2]);
  else if (strncmp (CLIENT, argv[1], strlen (CLIENT)) == 0 && argc >= 3)
    return client (argv[2], argv[3]);
  else
    {
      fprintf (stderr, "Usage: pubsub %s|%s <URL> <ARG> ...\n",
               SERVER, CLIENT);
      return 1;
    }
}
 

訂閱相關

assert (nn_setsockopt (sock, NN_SUB, NN_SUB_SUBSCRIBE, topic, length(topic)) >= 0);

說明:

    這裡推薦的TOPIC 型別是字串型別。一次訂閱只能訂閱一個TOPIC。

    如果你想訂閱兩個TOPIC，如"SHANGHAI", "BEIJING"，訂閱兩次。

assert (nn_setsockopt (sock, NN_SUB, NN_SUB_SUBSCRIBE, "SHANGHAI", strlen(SHANGHAI)) >= 0);
assert (nn_setsockopt (sock, NN_SUB, NN_SUB_SUBSCRIBE, "BEIJING", strlen(BEIJING)) >= 0);
 

釋出相關

如何釋出一個內容？

int bytes = nn_send (sock, d, content, length(content));

content是你想釋出的內容。

length(content)是釋出的內容的長度

如釋出上海的天氣

int bytes = nn_send (sock, d, "SHANGHAI|20'-25',SUN", strlen("SHANGHAI|20'-25',SUN");

上面釋出的內容接收方接收的資料的形式？

只有訂閱了

SHANGHAI***的訂閱方才能受到資料，我沒閱讀原始碼，估計釋出的內容會遍歷訂閱的話題，通配開頭內容相匹配的話題。---這樣的做法效能應該是不會很高的，但是可以適應多種訂閱業務的資料型別。

我感覺以字串做TOPIC，構建顯式的釋出介面，以如下形式提供更合理

pub(topic, content)

接收方以實現。

OnSub(topic, content)

這樣的方式可以用HASHMAP來管理TOPIC。

綜述

NANOMSG的訂閱釋出不是很高階。或者說，不是純粹的訂閱釋出業務型別。至少，沒有強制化樹立訂閱主題。每次釋出主題，需要遍歷通配所有的訂閱主題（效能有待提高）。

如果需要這樣強型別的主題，可以考慮ZMQ，需要注意的是，ZMQ的FILTER是在SUB端處理的，意味著所有的MSG都會推給所有的Suber方（非常欠缺），或者有空可以考慮自己實現一個。

參考連結