概述：

      這次是nanomsg庫實現的一個類似於MQTT通訊的一種方式，廣播訂閱的一個一對多的通訊方式。一個主的廣播訊息，其他可以訂閱自己想要的主題資訊，然後就會只接收訂閱的主題的資訊。

PubSub一對多主題訂閱通訊Demo

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <nanomsg/pubsub.h>
#include <nanomsg/nn.h>

/*
  此程式為nanomsg多執行緒一對多單向通訊demo,類似MQTT通訊，一個廣播，其他為訂閱相應主題
  客戶端只接收到自己訂閱的對應主題的內容。
*/

//inproc 標識用於多執行緒通訊
char *url = "inproc://sky_test";

//傳送資料的socket初始化
int send_sock_init(int *sock)
{
  *sock = nn_socket (AF_SP, NN_PUB);
  if (*sock < 0) {
    printf("create send data sock failed\r\n");
    return 1;
  }
  if (nn_bind(*sock, url) < 0) {
    printf("bind send data sock failed\r\n");
    return 1;
  }
  printf("send data socket init success...\r\n");
  return 0;
}

//接收資料的socket初始化
int recieve_sock_init(int *sock, char *topic)
{
  *sock = nn_socket (AF_SP, NN_SUB);
  if (*sock < 0) {
    printf("create recieve data sock failed\r\n");
    return 1;
  }
  if (NULL == topic) {
    //設定訂閱主題為全部
    nn_setsockopt(*sock, NN_SUB, NN_SUB_SUBSCRIBE, "", 0);
  } else {
    //設定訂閱的主題及主題長度，主要是對比傳送內容的開始位元組
    //如設定主題為"sky:",那麼就會對比資訊前面位元組是否相同，相同則可以收到該資訊
    nn_setsockopt(*sock, NN_SUB, NN_SUB_SUBSCRIBE, topic, strlen(topic));
  }
  if (nn_connect(*sock, url) < 0) {
    printf("connect recieve data sock failed\r\n");
    return 1;
  }
  printf("recieve data socket init success...\r\n");
  return 0;
}

//執行緒1測試
void *thread_test(void *arg)
{
  int c_sock;
  if (0 != recieve_sock_init(&c_sock, "sky:")) {
    return;
  }
  while (1) {
    //輪詢接收訂閱主題"sky:"資訊
    char *rx_msg = NULL;
    int result = nn_recv(c_sock, &rx_msg, NN_MSG, NN_DONTWAIT);
    if (result > 0) {
      printf("Thread 1 Recieve: %s\r\n\r\n", rx_msg);
      nn_freemsg (rx_msg);
    }
    sleep(1);
  }
}

//執行緒2測試
void *thread_test2(void *arg)
{
  int c_sock;
  if (0 != recieve_sock_init(&c_sock, "born:")) {
    return;
  }
  while (1) {
    //輪詢接收訂閱主題"born:"資訊
    char *rx_msg = NULL;
    int result = nn_recv(c_sock, &rx_msg, NN_MSG, NN_DONTWAIT);
    if (result > 0) {
      printf("Thread 2 Recieve: %s\r\n\r\n", rx_msg);
      nn_freemsg (rx_msg);
    }
    sleep(1);
  }
}

//傳送資料
int send_data(int sock, char *data)
{
  if (data == NULL) {
    return 1;
  }
  if (nn_send(sock, data, strlen(data)+1, 0) < 0) {
    return 1;
  }
  printf("Main Server Send:%s\r\n\r\n", data);
  return 0;
}

int main()
{
  int s_sock, ret, i = 0;
  pthread_t ps, ps2;
  char *tx_msg = "sky:Hello Thread Sky";
  char *tx_msg1 = "born:Hello Thread Born";
  char *tx_msg2 = "Storm:Hello Thread Storm";
  if (0 != send_sock_init(&s_sock)) {
    return 1;
  }
  //建立子執行緒，接收資訊
  pthread_create(&ps, NULL, thread_test, NULL);
  pthread_create(&ps2, NULL, thread_test2, NULL);
  sleep(1);
  //間隔兩秒，傳送資訊到子執行緒接收資料端
  while (1) {
    //測試傳送廣播
    if (0 == i) {
      ret = send_data(s_sock, tx_msg);
      if (0 == ret) {
        i ++;
      }
    } else if (1 == i) {
      ret = send_data(s_sock, tx_msg1);
      if (0 == ret) {
        i ++;
      }
    } else if (2 == i) {
      ret = send_data(s_sock, tx_msg2);
      if (0 == ret) {
        i = 0;
      }
    }
    sleep(2);
  }
  return 0;
}
 

編譯
gcc -o nanomsg_pubsub nanomsg_pubsub.c -lnanomsg -lpthread

執行結果：
[email protected]:~/Study/nanomsg/code_test/inproc/pubsub$ ./nanomsg_pubsub
send data socket init success...
recieve data socket init success...
recieve data socket init success...
Main Server Send:sky:Hello Thread Sky

Thread 1 Recieve: sky:Hello Thread Sky

Main Server Send:born:Hello Thread Born

Thread 2 Recieve: born:Hello Thread Born

Main Server Send:Storm:Hello Thread Storm

Main Server Send:sky:Hello Thread Sky

Thread 1 Recieve: sky:Hello Thread Sky
...

根據結果可以看到，執行緒1，2分別訂閱了sky born主題，所以可以收到，而storm主題沒訂閱所以都沒有收到。