communication inter-thread utilisant des messages ZeroMQ

Question

J'essaie d'utiliser zeroMQ comme un moyen d'implémenter un système de messagerie entre plusieurs threads. J'ai essayé le code ci-dessous mais ça ne marche pas; dans le spécifique, l'appel à zmq_recv dans chaque thread n'attend pas/bloque tout message à exécuter.

Pouvez-vous m'aider avec ce morceau de code?

J'utilise Linux OS et gcc

Cordialement

AFG

static void * 
    worker_routine (void *context) { 
     // Socket to talk to dispatcher 
     void *receiver = zmq_socket (context, ZMQ_REP); 
     zmq_connect (receiver, "inproc://workers"); 
     while (1) { 

      zmq_msg_t request; 
      zmq_msg_init(&request); 
      zmq_recv(receiver, &request, 0); 
      printf ("Received request\n"); 
      // Do some 'work' 
      usleep (1000); 
      // Send reply back to client 
      zmq_send (receiver, &request, 0); 
     } 
     zmq_close (receiver); 
     return NULL; 
    } 

    int main (void) { 

    void *context = zmq_init (1); 
    void *clients = zmq_socket (context, ZMQ_REP); 
    zmq_bind (clients, "inproc://workers"); 

    int thread_nbr; 
    for (thread_nbr = 0; thread_nbr < 5; thread_nbr++) { 
     pthread_t worker; 
     pthread_create (&worker, NULL, worker_routine, context); 
    } 

    zmq_close (clients); 
    zmq_term (context); 
    return 0; 
    } 

Answer 1

Vous êtes arrêter la prise et ZeroMQ juste après que vous créez les fils. Ils n'ont probablement pas le temps d'atteindre un état bloquant, et s'ils le faisaient, ils échoueraient dès que vous détruiriez le contexte zmq. De l'zmq_term man page:

terminaison de contexte est effectuée dans les étapes suivantes:

Toutes les opérations de blocage actuellement en cours sur les sockets ouverts dans le contexte doivent retourner immédiatement avec un code d'erreur de ETERM.

Answer 2

Les deux sockets sont REP. Ce que vous voulez, c'est REQ + REP.

Answer 3

Tout d'abord, comme @sustrik a noté que vous devez utiliser REQ et REP, le thread principal et les threads de travail ne peuvent pas être REP.

Deuxièmement, vous devez fournir une sorte de boucle de blocage dans votre thread principal:

int main (int argc, char **argv) 
{ 
    void *context = zmq_init (1); 
    void *clients = zmq_socket (context, ZMQ_REP); // use ZMQ_REQ on the clients 
    zmq_bind (clients, "inproc://workers"); 

    int thread_nbr; 
    for (thread_nbr = 0; thread_nbr < 5; thread_nbr++) { 
     pthread_t worker; 
     pthread_create (&worker, NULL, worker_routine, context); 
    } 

    while (TRUE) 
    { 
        // worker thread connected asking for work 
        zmq_msg_t request; 
        zmq_msg_init (&request); 
     zmq_recv (clients, &request, 0); 
     zmq_msg_close (&request); 

     // do whatever you need to do with the clients' request here 

     // send work to clients 
     zmq_msg_t reply; 
     zmq_msg_init_data (&reply, "Reply", 5, NULL, NULL); 
     zmq_send (clients, &reply, 0); 
     zmq_msg_close (&reply); 
    } 

    zmq_close (clients); 
    zmq_term (context); 
    return 0; 
}