自学教程：C++ zmq_ctx_new函数代码示例

int main(){    printf("Connection to hello world server..../n");    void *context = zmq_ctx_new();    void *socket = zmq_socket(context, ZMQ_REQ);    zmq_connect(socket, "tcp://localhost:8888");    int request_no;    for(request_no = 0; request_no != 10; ++request_no) {        char buffer[10];        printf("Sending Hello %d.../n", request_no);        if(-1 == zmq_send(socket, "Hello", 5, 0)){            printf("send message error, %d, %s/n", errno, zmq_strerror(errno));            break;        }        // zmq_recv(socket, buffer, 10, 0);        // printf("Received world %d/n", request_no);    }    zmq_close(socket);    zmq_ctx_destroy(context);    return 0;}

int main(void){    // Prepare our context and subscriber    void *context = zmq_ctx_new();    void *subscriber = zmq_socket(context, ZMQ_SUB);    zmq_connect(subscriber, "tcp://localhost:5563");    zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "B", 1);    while (1) {        // Read envelope with address        char *address = s_recv(subscriber);        // Read messages contents        char *contents = s_recv(subscriber);        printf("[%s] %s/n", address, contents);        free(address);        free(contents);    }    // We never get here, but clean up anyhow    zmq_close(subscriber);    zmq_ctx_destroy(context);    return 0;}

int main (void) {    void *context = zmq_ctx_new ();    //  Socket facing clients    void *frontend = zmq_socket (context, ZMQ_ROUTER);    int rc = zmq_bind (frontend, "tcp://*:5559");    assert (rc == 0);    //  Socket facing services    void *backend = zmq_socket (context, ZMQ_DEALER);    rc = zmq_bind (backend, "tcp://*:5560");    assert (rc == 0);    //  Start the proxy    zmq_proxy (frontend, backend, NULL);    //  We never get here...    zmq_close (frontend);    zmq_close (backend);    zmq_ctx_destroy (context);    return 0;}

int main (void)                                                                                        {                                                                                                          printf ("Connecting to hello world server…/n");                                                                                                                                                              /*创建一个新的上下文*/                                                                                 void *context = zmq_ctx_new ();                                                                        void *requester = zmq_socket (context, ZMQ_REQ);                                                       /*通过tcp协议，5555端口，连接本机服务端*/                                                              zmq_connect (requester, "tcp://localhost:5555");                                                      int request_nbr;                                                                                       for (request_nbr = 0; request_nbr != 10; request_nbr++) {                                                  char buffer [10];                                                                                      printf ("Sending Hello %d…/n", request_nbr);                                                          zmq_send (requester, "Hello", 5, 0);                                                                   zmq_recv (requester, buffer, 10, 0);                                                                   printf ("Received World %d/n", request_nbr);                                                       }                                                                                                                                                                                                             zmq_close (requester);                                                                                 zmq_ctx_destroy (context);                                                                                                                                                                                    return 0;                                                                                          }

struct router_context* router_init(int port, const char *host) {    struct router_context *ctx = (struct router_context*)malloc(sizeof(struct router_context));    if (ctx == NULL) {        syslog(LOG_CRIT, "router_init: cannot allocate memory");        return NULL;    }    ctx->zmq_ctx = zmq_ctx_new();    ctx->zmq_sock = zmq_socket(ctx->zmq_ctx, ZMQ_PAIR);    if (ctx->zmq_sock == NULL) {        syslog(LOG_CRIT, "router_init: failed to create syscall socket");        return NULL;    }    syslog(LOG_DEBUG, "router_init: created syscall socket");    char zmq_port_str[256];    sprintf(zmq_port_str, "tcp://%s:%d", host, port);    syslog(LOG_INFO, "Starting router at %s", zmq_port_str);    int ret = zmq_bind(ctx->zmq_sock, zmq_port_str);    if (ret != 0) {        syslog(LOG_CRIT, "router_init: failed to bind router socket. Zmq_bind returned %d", ret);    }    //TODO: check ctx_new, socket and bind return codes    syslog(LOG_DEBUG, "router_init: initializing process list lock");    if (pthread_mutex_init(&ctx->process_list_lock, NULL) != 0) {        syslog(LOG_ERR, "router_init: cannot initialize socket list mutex");        return NULL;    }    ctx->process_list = NULL;    return ctx;}

void Server::serve(const char* address) {    void* context;    void* socket;    zmq_msg_t message;    POMAGMA_INFO("Starting server");    POMAGMA_ASSERT_C((context = zmq_ctx_new()));    POMAGMA_ASSERT_C((socket = zmq_socket(context, ZMQ_REP)));    POMAGMA_ASSERT_C(0 == zmq_bind(socket, address));    while (true) {        POMAGMA_DEBUG("waiting for request");        POMAGMA_ASSERT_C(0 == zmq_msg_init(&message));        POMAGMA_ASSERT_C(-1 != zmq_msg_recv(&message, socket, 0));        POMAGMA_DEBUG("parsing request");        protobuf::AnalystRequest request;        bool parsed = request.ParseFromArray(zmq_msg_data(&message),                                             zmq_msg_size(&message));        POMAGMA_ASSERT(parsed, "Failed to parse request");        POMAGMA_ASSERT_C(0 == zmq_msg_close(&message));        protobuf::AnalystResponse response = handle(*this, request);        POMAGMA_DEBUG("serializing response");        std::string response_str;        response.SerializeToString(&response_str);        const int size = response_str.length();        POMAGMA_ASSERT_C(0 == zmq_msg_init(&message));        POMAGMA_ASSERT_C(0 == zmq_msg_init_size(&message, size));        memcpy(zmq_msg_data(&message), response_str.c_str(), size);        POMAGMA_DEBUG("sending response");        POMAGMA_ASSERT_C(size == zmq_msg_send(&message, socket, 0));        POMAGMA_ASSERT_C(0 == zmq_msg_close(&message));    }}

static void *worker_proc(void *userdata){    zmq_thread_data_t *ztd = userdata;    ztd->ctx = zmq_ctx_new ();    ztd->socket = zmq_socket(ztd->ctx,ZMQ_PULL);    char url[256];    sprintf(url, "tcp://%s:5557", ztd->server_ip);    zmq_connect(ztd->socket, url);    char file_prefix[256];    config_t *conf = ztd->conf;    sprintf(file_prefix, "%s-%s", ztd->server_ip, conf->file_prefix);    ztd->logger = create_logger(conf->home_dir, file_prefix, conf->max_file_count, conf->file_size_limit);    struct tm *ptm;    char query_log[1024];    char datetime[100];    char buf[1024];    int msg_len;    time_t tt;    LogMessage *msg;    pthread_t _pid = pthread_self();    while(run)    {         msg_len = zmq_recv(ztd->socket, buf, MAX_MSG_SIZE, 0);         msg = log_message__unpack(NULL, msg_len, buf);            tt = msg->date;         ptm = localtime(&tt);         strftime(datetime, 100, "%Y-%m-%d %H:%M:%S ", ptm);         sprintf(query_log, "%s %s %s: %s %d %d/n", datetime, msg->cip, msg->view, msg->domain, msg->rtype, msg->rcode);         log_msg(ztd->logger, query_log);         log_message__free_unpacked(msg, NULL);         if (ztd->count++%100000== 0)         {             printf(query_log);             printf("!!! thread %d:msglen %d count....%d/n", (int)_pid, msg_len, ztd->count);         }    }}

void test_pair_tcp (extra_func_t extra_func_ = NULL){    size_t len = MAX_SOCKET_STRING;    char my_endpoint[MAX_SOCKET_STRING];    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_PAIR);    assert (sb);    if (extra_func_)        extra_func_ (sb);    int rc = zmq_bind (sb, "tcp://127.0.0.1:*");    assert (rc == 0);    rc = zmq_getsockopt (sb, ZMQ_LAST_ENDPOINT, my_endpoint, &len);    assert (rc == 0);    void *sc = zmq_socket (ctx, ZMQ_PAIR);    assert (sc);    if (extra_func_)        extra_func_ (sc);    rc = zmq_connect (sc, my_endpoint);    assert (rc == 0);    bounce (sb, sc);    rc = zmq_close (sc);    assert (rc == 0);    rc = zmq_close (sb);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);    const char *binds [] = { "inproc://a", "tcp://127.0.0.1:5555" };    const char *connects [] = { "inproc://a", "tcp://localhost:5555" };    for (int transport = 0; transport < 2; ++transport) {        bind_address = binds [transport];        connect_address = connects [transport];        // PUSH: SHALL route outgoing messages to connected peers using a        // round-robin strategy.        test_push_round_robin_out (ctx);        // PULL: SHALL receive incoming messages from its peers using a fair-queuing        // strategy.        test_pull_fair_queue_in (ctx);        // PUSH: SHALL block on sending, or return a suitable error, when it has no        // available peers.        test_push_block_on_send_no_peers (ctx);        // PUSH and PULL: SHALL create this queue when a peer connects to it. If        // this peer disconnects, the socket SHALL destroy its queue and SHALL        // discard any messages it contains.        // *** Test disabled until libzmq does this properly ***        // test_destroy_queue_on_disconnect (ctx);    }    int rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

/* * LastValueCache * * This class acts as an aggregating proxy for all statistics generated by the product code. * Statistics are internally sent over inproc:// connections encapsulated in Subscription * envelopes and externally are sent over a tcp:// publishing socket on port 6666. * * This proxy also caches the last known value for a statistic and re-publishes it when a * subscriber registers interest.  This allows a client to poll the last known value easily. */LastValueCache::LastValueCache(int statcount,                               std::string *statnames,                               long poll_timeout_ms) :  //< Poll period in milliseconds  _statcount(statcount),  _statnames(statnames),  _poll_timeout_ms(poll_timeout_ms),  _terminate(false){  LOG_DEBUG("Initializing statistics aggregator");  _context = zmq_ctx_new();  _subscriber = new void *[_statcount];  // Bind all the sockets first, before we try to connect. This is a  // limitation of inproc sockets. See  // http://zguide.zeromq.org/page:all#Unicast-Transports  // and the thread at  // http://lists.zeromq.org/pipermail/zeromq-dev/2010-November/008012.html  // for the issues leading to this design.  for (int ii = 0; ii < _statcount; ii++)  {    std::string statname = _statnames[ii];    void* publisher = zmq_socket(_context, ZMQ_PUB);    zmq_bind(publisher, ("inproc://" + statname).c_str());    _internal_publishers[statname] = publisher;    LOG_DEBUG("Opened statistics socket inproc://%s", statname.c_str());  }  int rc = pthread_create(&_cache_thread,                          NULL,                          &last_value_cache_entry_func,                          (void *)this);  if (rc < 0)  {    LOG_ERROR("Failed to start statistics aggregator, no statistics will be available");  }}

int main(void) {  void *context = zmq_ctx_new();  // set up sender to send messages to the sb  void *sender = zmq_socket(context, ZMQ_PUSH);  zmq_connect(sender, "ipc:///tmp/zero_sb_in");  printf("PO: Connected to SB to send messages/n");  // set up subscriber to get new messages from subscribed MIDs  void *subscriber = zmq_socket(context, ZMQ_SUB);  int rc = zmq_connect(subscriber, "ipc:///tmp/zero_sb_pub");  assert(rc == 0);  printf("PO: Connected to SB to receive messages/n");  char *filter = "10";  rc = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter));  assert(rc == 0);  printf("PO: /tAnd the filter is in place./n");  // Wait for a ping  char buffer[256];  rc = zmq_recv(subscriber, buffer, 256, 0);  if (rc > 255) {    printf("PO: I can't handle this/n");  } else {    buffer[rc] = '/0';    printf("PO: Received: %s/n", buffer);  }  // Respond with a pong  zmq_send(sender, "11 pong", 7, 0);    zmq_close(sender);  zmq_close(subscriber);  zmq_ctx_destroy(context);  return 0;}

void* ZmqClient::createSocket(){    int timeout = 100;    int result = 0;    void *context = zmq_ctx_new();    void *backend = zmq_socket(context, ZMQ_PUSH);    if (!backend)    {        return NULL;    }    result = zmq_connect(backend, ZBROKER_FRONTEND_SOCKET);    if (result != 0)    {        zmq_close(backend);        return NULL;    }    result = zmq_setsockopt(backend, ZMQ_SNDTIMEO, &timeout, sizeof(int));    if (result != 0)    {        zmq_disconnect(backend, ZBROKER_FRONTEND_SOCKET);        zmq_close(backend);        return NULL;    }    result = zmq_setsockopt(backend, ZMQ_RCVTIMEO, &timeout, sizeof(int));    if (result != 0)    {        zmq_disconnect(backend, ZBROKER_FRONTEND_SOCKET);        zmq_close(backend);        return NULL;    }    return backend;}

int main(int argc, char *argv[]) {    if (argc != 2) {        usage(1);    }    void *ctx = zmq_ctx_new();    void *publisher = zmq_socket(ctx, ZMQ_PUSH);        int rc = zmq_connect(publisher, argv[1]);    assert(rc == 0);    while (!feof(stdin)) {        if (fgets(buffer, BUFFER_SIZE, stdin) != NULL) {            zmq_send(publisher, buffer, strlen(buffer), 0);        }    }    zmq_close(publisher);    zmq_ctx_destroy(ctx);    return 0;}

int main(void){    void *context = zmq_ctx_new();    void *publisher = zmq_socket(context, ZMQ_PUB);    int rc = zmq_bind(publisher, "tcp://*:8080");    assert (rc == 0);        srandom((unsigned)time(NULL));    while(1)    {		int zipcode, temperature, relhumidity;		zipcode = randof(100000);		temperature = randof(215) - 80;		relhumidity = randof(50) + 10;				char update[20];		sprintf(update, "%05d %d %d", zipcode, temperature, relhumidity);		s_send(publisher, update);	}	zmq_close(publisher);	zmq_ctx_destroy(context);		return 0;}

/** * constructor */static void* zmq_new(void){   t_zmq *x = (t_zmq *)pd_new(zmq_class);   char v[64];   sprintf(v, "version: %i.%i.%i", ZMQ_VERSION_MAJOR, ZMQ_VERSION_MINOR, ZMQ_VERSION_PATCH);   post(v);#if ZMQ_VERSION_MAJOR > 2   x->zmq_context = zmq_ctx_new();#else   x->zmq_context = zmq_init(1);#endif   if(x->zmq_context) {      post("
C++ zmq_ctx_term函数代码示例
C++ zmq_ctx_destroy函数代码示例